codekingpro/portable-devtools

1"""Utility methods for marshmallow."""2 3# ruff: noqa: T201, T2034from __future__ import annotations5 6import collections7import datetime as dt8import functools9import inspect10import json11import re12import typing13import warnings14from collections.abc import Mapping15from email.utils import format_datetime, parsedate_to_datetime16from pprint import pprint as py_pprint17 18from marshmallow.base import FieldABC19from marshmallow.exceptions import FieldInstanceResolutionError20from marshmallow.warnings import RemovedInMarshmallow4Warning21 22if typing.TYPE_CHECKING:23    from marshmallow.fields import Field24 25 26EXCLUDE = "exclude"27INCLUDE = "include"28RAISE = "raise"29_UNKNOWN_VALUES = {EXCLUDE, INCLUDE, RAISE}30 31 32class _Missing:33    def __bool__(self):34        return False35 36    def __copy__(self):37        return self38 39    def __deepcopy__(self, _):40        return self41 42    def __repr__(self):43        return "<marshmallow.missing>"44 45 46# Singleton value that indicates that a field's value is missing from input47# dict passed to `Schema.load <marshmallow.Schema.load>`. If the field's value is not required,48# it's ``default`` value is used.49missing = _Missing()50 51 52def is_generator(obj) -> bool:53    """Return True if ``obj`` is a generator"""54    return inspect.isgeneratorfunction(obj) or inspect.isgenerator(obj)55 56 57def is_iterable_but_not_string(obj) -> bool:58    """Return True if ``obj`` is an iterable object that isn't a string."""59    return (hasattr(obj, "__iter__") and not hasattr(obj, "strip")) or is_generator(obj)60 61 62def is_collection(obj) -> bool:63    """Return True if ``obj`` is a collection type, e.g list, tuple, queryset."""64    return is_iterable_but_not_string(obj) and not isinstance(obj, Mapping)65 66 67def is_instance_or_subclass(val, class_) -> bool:68    """Return True if ``val`` is either a subclass or instance of ``class_``."""69    try:70        return issubclass(val, class_)71    except TypeError:72        return isinstance(val, class_)73 74 75def is_keyed_tuple(obj) -> bool:76    """Return True if ``obj`` has keyed tuple behavior, such as77    namedtuples or SQLAlchemy's KeyedTuples.78    """79    return isinstance(obj, tuple) and hasattr(obj, "_fields")80 81 82def pprint(obj, *args, **kwargs) -> None:83    """Pretty-printing function that can pretty-print OrderedDicts84    like regular dictionaries. Useful for printing the output of85    :meth:`marshmallow.Schema.dump`.86 87    .. deprecated:: 3.7.088        marshmallow.pprint will be removed in marshmallow 4.89    """90    warnings.warn(91        "marshmallow's pprint function is deprecated and will be removed in marshmallow 4.",92        RemovedInMarshmallow4Warning,93        stacklevel=2,94    )95    if isinstance(obj, collections.OrderedDict):96        print(json.dumps(obj, *args, **kwargs))97    else:98        py_pprint(obj, *args, **kwargs)99 100 101# https://stackoverflow.com/a/27596917102def is_aware(datetime: dt.datetime) -> bool:103    return (104        datetime.tzinfo is not None and datetime.tzinfo.utcoffset(datetime) is not None105    )106 107 108def from_rfc(datestring: str) -> dt.datetime:109    """Parse a RFC822-formatted datetime string and return a datetime object.110 111    https://stackoverflow.com/questions/885015/how-to-parse-a-rfc-2822-date-time-into-a-python-datetime  # noqa: B950112    """113    return parsedate_to_datetime(datestring)114 115 116def rfcformat(datetime: dt.datetime) -> str:117    """Return the RFC822-formatted representation of a datetime object.118 119    :param datetime: The datetime.120    """121    return format_datetime(datetime)122 123 124# Hat tip to Django for ISO8601 deserialization functions125 126_iso8601_datetime_re = re.compile(127    r"(?P<year>\d{4})-(?P<month>\d{1,2})-(?P<day>\d{1,2})"128    r"[T ](?P<hour>\d{1,2}):(?P<minute>\d{1,2})"129    r"(?::(?P<second>\d{1,2})(?:\.(?P<microsecond>\d{1,6})\d{0,6})?)?"130    r"(?P<tzinfo>Z|[+-]\d{2}(?::?\d{2})?)?$"131)132 133_iso8601_date_re = re.compile(r"(?P<year>\d{4})-(?P<month>\d{1,2})-(?P<day>\d{1,2})$")134 135_iso8601_time_re = re.compile(136    r"(?P<hour>\d{1,2}):(?P<minute>\d{1,2})"137    r"(?::(?P<second>\d{1,2})(?:\.(?P<microsecond>\d{1,6})\d{0,6})?)?"138)139 140 141def get_fixed_timezone(offset: float | dt.timedelta) -> dt.timezone:142    """Return a tzinfo instance with a fixed offset from UTC."""143    if isinstance(offset, dt.timedelta):144        offset = offset.total_seconds() // 60145    sign = "-" if offset < 0 else "+"146    hhmm = "{:02d}{:02d}".format(*divmod(abs(offset), 60))147    name = sign + hhmm148    return dt.timezone(dt.timedelta(minutes=offset), name)149 150 151def from_iso_datetime(value):152    """Parse a string and return a datetime.datetime.153 154    This function supports time zone offsets. When the input contains one,155    the output uses a timezone with a fixed offset from UTC.156    """157    match = _iso8601_datetime_re.match(value)158    if not match:159        raise ValueError("Not a valid ISO8601-formatted datetime string")160    kw = match.groupdict()161    kw["microsecond"] = kw["microsecond"] and kw["microsecond"].ljust(6, "0")162    tzinfo = kw.pop("tzinfo")163    if tzinfo == "Z":164        tzinfo = dt.timezone.utc165    elif tzinfo is not None:166        offset_mins = int(tzinfo[-2:]) if len(tzinfo) > 3 else 0167        offset = 60 * int(tzinfo[1:3]) + offset_mins168        if tzinfo[0] == "-":169            offset = -offset170        tzinfo = get_fixed_timezone(offset)171    kw = {k: int(v) for k, v in kw.items() if v is not None}172    kw["tzinfo"] = tzinfo173    return dt.datetime(**kw)  # noqa: DTZ001174 175 176def from_iso_time(value):177    """Parse a string and return a datetime.time.178 179    This function doesn't support time zone offsets.180    """181    match = _iso8601_time_re.match(value)182    if not match:183        raise ValueError("Not a valid ISO8601-formatted time string")184    kw = match.groupdict()185    kw["microsecond"] = kw["microsecond"] and kw["microsecond"].ljust(6, "0")186    kw = {k: int(v) for k, v in kw.items() if v is not None}187    return dt.time(**kw)188 189 190def from_iso_date(value):191    """Parse a string and return a datetime.date."""192    match = _iso8601_date_re.match(value)193    if not match:194        raise ValueError("Not a valid ISO8601-formatted date string")195    kw = {k: int(v) for k, v in match.groupdict().items()}196    return dt.date(**kw)197 198 199def from_timestamp(value: typing.Any) -> dt.datetime:200    if value is True or value is False:201        raise ValueError("Not a valid POSIX timestamp")202    value = float(value)203    if value < 0:204        raise ValueError("Not a valid POSIX timestamp")205 206    # Load a timestamp with utc as timezone to prevent using system timezone.207    # Then set timezone to None, to let the Field handle adding timezone info.208    try:209        return dt.datetime.fromtimestamp(value, tz=dt.timezone.utc).replace(tzinfo=None)210    except OverflowError as exc:211        raise ValueError("Timestamp is too large") from exc212    except OSError as exc:213        raise ValueError("Error converting value to datetime") from exc214 215 216def from_timestamp_ms(value: typing.Any) -> dt.datetime:217    value = float(value)218    return from_timestamp(value / 1000)219 220 221def timestamp(222    value: dt.datetime,223) -> float:224    if not is_aware(value):225        # When a date is naive, use UTC as zone info to prevent using system timezone.226        value = value.replace(tzinfo=dt.timezone.utc)227    return value.timestamp()228 229 230def timestamp_ms(value: dt.datetime) -> float:231    return timestamp(value) * 1000232 233 234def isoformat(datetime: dt.datetime) -> str:235    """Return the ISO8601-formatted representation of a datetime object.236 237    :param datetime: The datetime.238    """239    return datetime.isoformat()240 241 242def to_iso_time(time: dt.time) -> str:243    return dt.time.isoformat(time)244 245 246def to_iso_date(date: dt.date) -> str:247    return dt.date.isoformat(date)248 249 250def ensure_text_type(val: str | bytes) -> str:251    if isinstance(val, bytes):252        val = val.decode("utf-8")253    return str(val)254 255 256def pluck(dictlist: list[dict[str, typing.Any]], key: str):257    """Extracts a list of dictionary values from a list of dictionaries.258    ::259 260        >>> dlist = [{'id': 1, 'name': 'foo'}, {'id': 2, 'name': 'bar'}]261        >>> pluck(dlist, 'id')262        [1, 2]263    """264    return [d[key] for d in dictlist]265 266 267# Various utilities for pulling keyed values from objects268 269 270def get_value(obj, key: int | str, default=missing):271    """Helper for pulling a keyed value off various types of objects. Fields use272    this method by default to access attributes of the source object. For object `x`273    and attribute `i`, this method first tries to access `x[i]`, and then falls back to274    `x.i` if an exception is raised.275 276    .. warning::277        If an object `x` does not raise an exception when `x[i]` does not exist,278        `get_value` will never check the value `x.i`. Consider overriding279        `marshmallow.fields.Field.get_value` in this case.280    """281    if not isinstance(key, int) and "." in key:282        return _get_value_for_keys(obj, key.split("."), default)283    return _get_value_for_key(obj, key, default)284 285 286def _get_value_for_keys(obj, keys, default):287    if len(keys) == 1:288        return _get_value_for_key(obj, keys[0], default)289    return _get_value_for_keys(290        _get_value_for_key(obj, keys[0], default), keys[1:], default291    )292 293 294def _get_value_for_key(obj, key, default):295    if not hasattr(obj, "__getitem__"):296        return getattr(obj, key, default)297 298    try:299        return obj[key]300    except (KeyError, IndexError, TypeError, AttributeError):301        return getattr(obj, key, default)302 303 304def set_value(dct: dict[str, typing.Any], key: str, value: typing.Any):305    """Set a value in a dict. If `key` contains a '.', it is assumed306    be a path (i.e. dot-delimited string) to the value's location.307 308    ::309 310        >>> d = {}311        >>> set_value(d, 'foo.bar', 42)312        >>> d313        {'foo': {'bar': 42}}314    """315    if "." in key:316        head, rest = key.split(".", 1)317        target = dct.setdefault(head, {})318        if not isinstance(target, dict):319            raise ValueError(320                f"Cannot set {key} in {head} due to existing value: {target}"321            )322        set_value(target, rest, value)323    else:324        dct[key] = value325 326 327def callable_or_raise(obj):328    """Check that an object is callable, else raise a :exc:`TypeError`."""329    if not callable(obj):330        raise TypeError(f"Object {obj!r} is not callable.")331    return obj332 333 334def _signature(func: typing.Callable) -> list[str]:335    return list(inspect.signature(func).parameters.keys())336 337 338def get_func_args(func: typing.Callable) -> list[str]:339    """Given a callable, return a list of argument names. Handles340    `functools.partial` objects and class-based callables.341 342    .. versionchanged:: 3.0.0a1343        Do not return bound arguments, eg. ``self``.344    """345    if inspect.isfunction(func) or inspect.ismethod(func):346        return _signature(func)347    if isinstance(func, functools.partial):348        return _signature(func.func)349    # Callable class350    return _signature(func)351 352 353def resolve_field_instance(cls_or_instance: type[Field] | Field) -> Field:354    """Return a field instance from a field class or instance.355 356    :param cls_or_instance: Field class or instance.357    """358    if isinstance(cls_or_instance, type):359        if not issubclass(cls_or_instance, FieldABC):360            raise FieldInstanceResolutionError361        return cls_or_instance()362    if not isinstance(cls_or_instance, FieldABC):363        raise FieldInstanceResolutionError364    return cls_or_instance365 366 367def timedelta_to_microseconds(value: dt.timedelta) -> int:368    """Compute the total microseconds of a timedelta369 370    https://github.com/python/cpython/blob/bb3e0c240bc60fe08d332ff5955d54197f79751c/Lib/datetime.py#L665-L667  # noqa: B950371    """372    return (value.days * (24 * 3600) + value.seconds) * 1000000 + value.microseconds373 374 375def validate_unknown_parameter_value(obj: typing.Any) -> str:376    if obj not in _UNKNOWN_VALUES:377        raise ValueError(378            f"Object {obj!r} is not a valid value for the 'unknown' parameter"379        )380    return obj381