Skip to content

unmarshals

Support for unmarshalling unstructured data into Python data structures.

Notes

"Unmarshalling" refers to the process of taking a "primitive" form of data, such as a basic dictionary or JSON string, and coercing it into a higher-order structured data type.

Tip

You may use this package directly, but we encourage you to work with the higher-level API provided by the typelib module.

Typical Usage

>>> import dataclasses
>>> import decimal
>>> from typelib import unmarshals
>>>
>>> @dataclasses.dataclass(slots=True, weakref_slot=True, kw_only=True)
... class Struct:
...     key: str
...     number: decimal.Decimal
...
>>>
>>> data = {"key": "some-key", "number": "3.14"}
>>> unmarshals.unmarshal(Struct, data)
Struct(key='some-key', number=decimal.Decimal('3.14'))
>>> unmarshaller = unmarshals.unmarshaller(Struct)
>>> unmarshaller(data)
Struct(key='some-key', number=decimal.Decimal('3.14'))
See Also

AbstractUnmarshaller

AbstractUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: ABC, Generic[T]

Abstract base class defining the common interface for unmarshallers.

Unmarshallers are custom callables which maintain type-specific information. They use this information to provide robust, performant logic for decoding and converting primtive Python objects or JSON-endcoded data into their target type.

Unmarshallers support contextual deserialization, which enables the unmarshalling of nested types.

Attributes:

  • t (type[T]) –

    The root type of this unmarshaller.

  • origin (type[T]) –

    If t is a generic, this will be an actionable runtime type related to t, otherwise it is the same as t.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None) –

    If this unmarshaller is used in a nested context, this will reference the field/parameter/index at which this unmarshaller should be used.

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__ abstractmethod

__call__(val: Any) -> T

Unmarshall a Python object into its target type.

Not implemented for the abstract base class.

Source code in src/typelib/unmarshals/routines.py
@abc.abstractmethod
def __call__(self, val: tp.Any) -> T:
    """Unmarshall a Python object into its target type.

    Not implemented for the abstract base class.
    """

BytesUnmarshaller

BytesUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[BytesT], Generic[BytesT]

Unmarshaller that encodes an input to bytes.

Note

We will format a member of the datetime module into ISO format before converting to bytes.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

CastUnmarshaller

CastUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[T]

Unmarshaller that converts an input to an instance of T with a direct cast.

Note

Before casting to the bound type, we will attempt to decode the value into a real Python object.

See Also

Parameters:

  • t (type[T]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context (unused).

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context (unused).
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    self.caster: tp.Callable[[tp.Any], T] = self.origin  # type: ignore[assignment]

__call__

__call__(val: Any) -> T

Unmarshal a value into the bound T type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> T:
    """Unmarshal a value into the bound `T` type.

    Args:
        val: The input value to unmarshal.
    """
    # Try to load the string, if this is JSON or a literal expression.
    decoded = serdes.load(val)
    # Short-circuit cast if we have the type we want.
    if isinstance(decoded, self.t):
        return decoded
    # Cast the decoded value to the type.
    return self.caster(decoded)

DateTimeUnmarshaller

DateTimeUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[datetime], Generic[DateTimeT]

Unmarshaller that converts an input to a datetime.datetime (or subclasses).

Notes

This class tries to handle the 90% case:

  1. If we are already a datetime.datetime instance, return it.
  2. If we are a float or int instance, treat it as a unix timestamp, at UTC.
  3. Attempt to decode any bytes/string input into a real Python value.
  4. If we have a string value, parse it into either a datetime.date instance, a datetime.time instance or a datetime.datetime.
  5. If the parsed result is a datetime.time instance, then merge the parsed time with today, at the timezone specified in the time instance.
  6. If the parsed result is a datetime.date instance, create a datetime.datetime instance at midnight of the indicated date, UTC.
TL;DR

There are many ways to represent a datetime object over-the-wire. Your most fool-proof method is to rely upon ISO 8601 or RFC 3339.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> datetime

Unmarshal a value into the bound DateTimeT type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> datetime.datetime:
    """Unmarshal a value into the bound `DateTimeT` type.

    Args:
        val: The input value to unmarshal.
    """
    if isinstance(val, self.t):
        return val

    # Numbers can be treated as time since epoch.
    if isinstance(val, (int, float)):
        val = datetime.datetime.fromtimestamp(val, tz=datetime.timezone.utc)
    # Always decode bytes.
    decoded = serdes.decode(val)
    # Parse strings.
    dt: datetime.datetime | datetime.date | datetime.time = (
        serdes.dateparse(decoded, self.t) if isinstance(decoded, str) else decoded
    )
    # If we have a time object, default to today.
    if isinstance(dt, datetime.time):
        return self.t.now(tz=dt.tzinfo).replace(
            hour=dt.hour,
            minute=dt.minute,
            second=dt.second,
            microsecond=dt.microsecond,
            tzinfo=dt.tzinfo,
        )
    # Exact class matching.
    if dt.__class__ is self.t:
        return dt  # type: ignore[return-value]
    # Subclass check for datetimes.
    if isinstance(dt, datetime.datetime):
        return self.t(
            year=dt.year,
            month=dt.month,
            day=dt.day,
            hour=dt.hour,
            minute=dt.minute,
            second=dt.second,
            microsecond=dt.microsecond,
            tzinfo=dt.tzinfo,
            fold=dt.fold,
        )
    # Implicit: we have a date object.
    return self.t(
        year=dt.year, month=dt.month, day=dt.day, tzinfo=datetime.timezone.utc
    )

DateUnmarshaller

DateUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[DateT], Generic[DateT]

Unmarshaller that converts an input to a datetime.date (or subclasses).

Notes

This class tries to handle the 90% case:

  1. If we are already a datetime.date instance, return it.
  2. If we are a float or int instance, treat it as a unix timestamp, at UTC.
  3. Attempt to decode any bytes/string input into a real Python value.
  4. If we have a string value, parse it into either a datetime.date
  5. If the parsed result is a datetime.time instance, then return the result of datetime.datetime.now, at UTC, as a datetime.date.
TL;DR

There are many ways to represent a date object over-the-wire. Your most fool-proof method is to rely upon ISO 8601 or RFC 3339.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> DateT

Unmarshal a value into the bound DateT type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> DateT:
    """Unmarshal a value into the bound `DateT` type.

    Args:
        val: The input value to unmarshal.
    """
    if isinstance(val, self.t) and not isinstance(val, datetime.datetime):
        return val

    # Numbers can be treated as time since epoch.
    if isinstance(val, (int, float)):
        val = datetime.datetime.fromtimestamp(val, tz=datetime.timezone.utc)
    # Always decode bytes.
    decoded = serdes.decode(val)
    # Parse strings.
    date: datetime.date | datetime.time = (
        serdes.dateparse(decoded, self.t) if isinstance(decoded, str) else decoded
    )
    # Time-only construct is treated as today.
    if isinstance(date, datetime.time):
        date = datetime.datetime.now(tz=datetime.timezone.utc).today()
    # Exact class matching - the parser returns subclasses.
    if date.__class__ is self.t:
        return date  # type: ignore[return-value]
    # Reconstruct as the exact type.
    return self.t(year=date.year, month=date.month, day=date.day)

DelayedUnmarshaller

DelayedUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[T]

Delayed proxy for a given type's unmarshaller, used when we encounter a typing.ForwardRef.

Notes

This allows us to delay the resolution of the given type reference until call-time, enabling support for cyclic and recursive types.

Source code in src/typelib/unmarshals/api.py
def __init__(
    self, t: type[T], context: routines.ContextT, *, var: str | None = None
):
    super().__init__(t, context, var=var)
    self._resolved: routines.AbstractUnmarshaller[T] | None = None

resolved property

resolved: AbstractUnmarshaller[T]

The resolved unmarshaller.

FixedTupleUnmarshaller

FixedTupleUnmarshaller(t: type[TupleT], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[TupleT]

Unmarshaller for a "fixed" tuple (e.g., tuple[int, str, float]).

Note

Python supports two distinct uses for tuples, unlike in other languages:

  1. Tuples with a fixed number of members.
  2. Tuples of variable length (an immutable sequence).

"Fixed" tuples may have a distinct type for each member, while variable-length tuples may only have a single type (or union of types) for all members.

Variable-length tuples are handled by our generic iterable unmarshaller.

For "fixed" tuples, the algorithm is:

  1. Attempt to decode the input into a real Python object.
  2. zip the stack of member unmarshallers and the values in the decoded object.
  3. Unmarshal each value using the associated unmarshaller for that position.
  4. Pass the unmarshalling iterator in to the type's constructor.
Tip

If the input has more members than the type definition allows, those members will be dropped by nature of our unmarshalling algorithm.

See Also

Parameters:

  • t (type[TupleT]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context. Used to resolve the value unmarshaller stack.

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(
    self, t: type[compat.TupleT], context: ContextT, *, var: str | None = None
):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context. Used to resolve the value unmarshaller stack.
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    self.stack = inspection.args(t, evaluate=True)
    self.ordered_routines = [self.context[vt] for vt in self.stack]

__call__

__call__(val: Any) -> TupleT

Unmarshal a value into the bound tuple structure.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> compat.TupleT:
    """Unmarshal a value into the bound [`tuple`][] structure.

    Args:
        val: The input value to unmarshal.
    """
    decoded = serdes.load(val)
    return self.origin(
        routine(v)
        for routine, v in zip(self.ordered_routines, serdes.itervalues(decoded))
    )

LiteralUnmarshaller

LiteralUnmarshaller(t: type[LiteralT], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[LiteralT], Generic[LiteralT]

Unmarshaller that will enforce an input conform to a defined typing.Literal.

Note

We will attempt to decode the value into a real Python object if the input fails initial membership evaluation.

See Also

Parameters:

  • t (type[LiteralT]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context (unused).

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[LiteralT], context: ContextT, *, var: str | None = None):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context (unused).
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    self.values = inspection.args(t, evaluate=True)

NoOpUnmarshaller

NoOpUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[T]

Unmarshaller that does nothing.

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

NoneTypeUnmarshaller

NoneTypeUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[None]

Unmarshaller for null values.

Note

We will attempt to decode any string/bytes input before evaluating for None.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> None

Unmarshal the given input into a None value.

Parameters:

  • val (Any) –

    The value to unmarshal.

Raises:

  • ValueError

    If val is not None after decoding.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> None:
    """Unmarshal the given input into a `None` value.

    Args:
        val: The value to unmarshal.

    Raises:
        ValueError: If `val` is not `None` after decoding.
    """
    decoded = serdes.decode(val)
    if decoded is not None:
        raise ValueError(f"{val!r} is not of {types.NoneType!r}")
    return None

NumberUnmarshaller

NumberUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[NumberT], Generic[NumberT]

Unmarshaller that converts an input to a number.

Note

Number unmarshalling follows a best-effort strategy. We may extend type resolution to support more advanced type unmarshalling strategies in the future.

As of now: 1. Attempt to decode any bytes/string input into a real Python value. 2. If the input is a member of the datetime module, convert it to a number. 3. If the input is a mapping, unpack it into the number constructor. 4. If the input is an iterable, unpack it into the number constructor. 5. Otherwise, call the number constructor with the input.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> NumberT

Unmarshall a value into the bound Number type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> NumberT:
    """Unmarshall a value into the bound Number type.

    Args:
        val: The input value to unmarshal.
    """
    # Always decode bytes.
    decoded = serdes.decode(val)
    if isinstance(decoded, self.t):
        return decoded
    # Represent date/time objects as time since unix epoch.
    if isinstance(val, (datetime.date, datetime.time, datetime.timedelta)):
        decoded = serdes.unixtime(val)
    # Treat containers as constructor args.
    if inspection.ismappingtype(decoded.__class__):
        return self.t(**decoded)
    if inspection.isiterabletype(decoded.__class__) and not inspection.istexttype(
        decoded.__class__
    ):
        return self.t(*decoded)
    # Simple cast for non-containers.
    return self.t(decoded)  # type: ignore[call-arg]

PatternUnmarshaller

PatternUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[PatternT], Generic[PatternT]

Unmarshaller that converts an input to a re.Pattern.

Note

You can't instantiate a re.Pattern directly, so we don't have a good method for handling patterns from a different library out-of-the-box. We simply call re.compile() on the decoded input.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

StringUnmarshaller

StringUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[StringT], Generic[StringT]

Unmarshaller that converts an input to a string.

Note

We will format a member of the datetime module into ISO format.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

StructuredTypeUnmarshaller

StructuredTypeUnmarshaller(t: type[_ST], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[_ST]

Unmarshaller for a "structured" (user-defined) type.

Note

This unmarshaller supports the unmarshalling of any mapping or structured type into the targeted structured type. There are limitations.

The algorithm is:

  1. Attempt to decode the input into a real Python object.
  2. Using a mapping of the structured types "field" to the field-type's unmarshaller, iterate over the field->value pairs of the input, skipping fields in the input which are not present in the field mapping.
  3. Store each unmarshalled value in a keyword-argument mapping.
  4. Unpack the keyword argument mapping into the bound type's constructor.
Tip

While we don't currently support arbitrary collections, we may add this functionality at a later date. Doing so requires more advanced introspection and parameter-binding that would lead to a significant loss in performance if not done carefully.

See Also

Parameters:

  • t (type[_ST]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context. Used to resolve the value field-to-unmarshaller mapping.

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[_ST], context: ContextT, *, var: str | None = None):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context. Used to resolve the value field-to-unmarshaller mapping.
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    self.fields_by_var = self._fields_by_var()

__call__

__call__(val: Any) -> _ST

Unmarshal a value into the bound type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> _ST:
    """Unmarshal a value into the bound type.

    Args:
        val: The input value to unmarshal.
    """
    decoded = serdes.load(val)
    fields = self.fields_by_var
    kwargs = {f: fields[f](v) for f, v in serdes.iteritems(decoded) if f in fields}
    return self.t(**kwargs)

SubscriptedIterableUnmarshaller

SubscriptedIterableUnmarshaller(t: type[IterableT], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[IterableT], Generic[IterableT]

Unmarshaller for a subscripted iterable type.

Note

This unmarshaller handles standard simple iterable types. We leverage our own generic itervalues to allow for translating other collections or structured objects into the target iterable.

The algorithm is as follows:

  1. We attempt to decode the input into a real Python object.
  2. We iterate over values in the decoded input.
  3. We call the value-type's unmarshaller on the value members.
  4. We pass the unmarshalling iterator in to the type's constructor.
See Also

Parameters:

  • t (type[IterableT]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context. Used to resolve the member unmarshaller.

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(
    self, t: type[IterableT], context: ContextT, *, var: str | None = None
):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context. Used to resolve the member unmarshaller.
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t=t, context=context, var=var)
    # supporting tuple[str, ...]
    (value_t, *_) = inspection.args(t, evaluate=True)
    self.values = context[value_t]

__call__

__call__(val: Any) -> IterableT

Unmarshal a value into the bound IterableT.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> IterableT:
    """Unmarshal a value into the bound `IterableT`.

    Args:
        val: The input value to unmarshal.
    """
    # Always decode bytes.
    decoded = serdes.load(val)
    values = self.values
    return self.origin((values(v) for v in serdes.itervalues(decoded)))  # type: ignore[call-arg]

SubscriptedIteratorUnmarshaller

SubscriptedIteratorUnmarshaller(t: type[IteratorT], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[IteratorT], Generic[IteratorT]

Unmarshaller for a subscripted iterator type.

Note

This unmarshaller handles standard simple iterable types. We leverage our own generic itervalues to allow for translating other collections or structured objects into the target iterator.

The algorithm is as follows:

  1. We attempt to decode the input into a real Python object.
  2. We iterate over values in the decoded input.
  3. We call the value-type's unmarshaller on the value members.
  4. We return a new, unmarshalling iterator.
See Also

Parameters:

  • t (type[IteratorT]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context. Used to resolve the member unmarshaller.

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(
    self, t: type[IteratorT], context: ContextT, *, var: str | None = None
):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context. Used to resolve the member unmarshaller.
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    (value_t,) = inspection.args(t, evaluate=True)
    self.values = context[value_t]

__call__

__call__(val: Any) -> IteratorT

Unmarshal a value into the bound IteratorT.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> IteratorT:
    """Unmarshal a value into the bound `IteratorT`.

    Args:
        val: The input value to unmarshal.
    """
    # Always decode bytes.
    decoded = serdes.load(val)
    values = self.values
    it: IteratorT = (values(v) for v in serdes.itervalues(decoded))  # type: ignore[assignment]
    return it

SubscriptedMappingUnmarshaller

SubscriptedMappingUnmarshaller(t: type[MappingT], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[MappingT], Generic[MappingT]

Unmarshaller for a subscripted mapping type.

Note

This unmarshaller handles standard key->value mappings. We leverage our own generic iteritems to allow for translating other collections or structured objects into the target mapping.

The algorithm is as follows:

  1. We attempt to decode the input into a real Python object.
  2. We iterate over key->value pairs.
  3. We call the key-type's unmarshaller on the key members.
  4. We call the value-type's unmarshaller on the value members.
  5. We pass the unmarshalling iterator in to the type's constructor.
See Also

Parameters:

  • t (type[MappingT]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context. Used to resolve the member unmarshallers.

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[MappingT], context: ContextT, *, var: str | None = None):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context. Used to resolve the member unmarshallers.
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    key_t, value_t = inspection.args(t, evaluate=True)
    self.keys = context[key_t]
    self.values = context[value_t]

__call__

__call__(val: Any) -> MappingT

Unmarshal a value into the bound MappingT.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> MappingT:
    """Unmarshal a value into the bound `MappingT`.

    Args:
        val: The input value to unmarshal.
    """
    # Always decode bytes.
    decoded = serdes.load(val)
    keys = self.keys
    values = self.values
    return self.origin(  # type: ignore[call-arg]
        ((keys(k), values(v)) for k, v in serdes.iteritems(decoded))
    )

TimeDeltaUnmarshaller

TimeDeltaUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[TimeDeltaT], Generic[TimeDeltaT]

Unmarshaller that converts an input to a datetime.timedelta (or subclasses).

Notes

This class tries to handle the 90% case:

  1. If we are already a datetime.timedelta instance, return it.
  2. If we are a float or int instance, treat it as total seconds for a delta.
  3. Attempt to decode any bytes/string input into a real Python value.
  4. If we have a string value, parse it into a datetime.timedelta instance.
  5. If the parsed result is not exactly the bound TimeDeltaT type, convert it.
TL;DR

There are many ways to represent a time object over-the-wire. Your most fool-proof method is to rely upon ISO 8601 or RFC 3339.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> TimeDeltaT

Unmarshal a value into the bound TimeDeltaT type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> TimeDeltaT:
    """Unmarshal a value into the bound `TimeDeltaT` type.

    Args:
        val: The input value to unmarshal.
    """
    if isinstance(val, (int, float)):
        return self.t(seconds=int(val))

    decoded = serdes.decode(val)
    td: datetime.timedelta = (
        serdes.dateparse(decoded, t=datetime.timedelta)
        if isinstance(decoded, str)
        else decoded
    )

    if td.__class__ is self.t:
        return td  # type: ignore[return-value]

    return self.t(seconds=td.total_seconds())

TimeUnmarshaller

TimeUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[TimeT], Generic[TimeT]

Unmarshaller that converts an input to adatetime.time (or subclasses).

Notes

This class tries to handle the 90% case:

  1. If we are already a datetime.time instance, return it.
  2. If we are a float or int instance, treat it as a unix timestamp, at UTC.
  3. Attempt to decode any bytes/string input into a real Python value.
  4. If we have a string value, parse it into either a datetime.date instance, a datetime.time instance or a datetime.datetime.
  5. If the parsed result is a datetime.datetime instance, then extract the time portion, preserving the associated timezone.
  6. If the parsed result is a datetime.date instance, create a time instance at midnight, UTC.
TL;DR

There are many ways to represent a time object over-the-wire. Your most fool-proof method is to rely upon ISO 8601 or RFC 3339.

See Also

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> TimeT

Unmarshal a value into the bound TimeT type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> TimeT:
    """Unmarshal a value into the bound `TimeT` type.

    Args:
        val: The input value to unmarshal.
    """
    if isinstance(val, self.t):
        return val

    decoded = serdes.decode(val)
    if isinstance(decoded, (int, float)):
        decoded = (
            datetime.datetime.fromtimestamp(val, tz=datetime.timezone.utc)
            .time()
            # datetime.time() strips tzinfo...
            .replace(tzinfo=datetime.timezone.utc)
        )
    dt: datetime.datetime | datetime.date | datetime.time = (
        serdes.dateparse(decoded, self.t) if isinstance(decoded, str) else decoded
    )

    if isinstance(dt, datetime.datetime):
        # datetime.time() strips tzinfo...
        dt = dt.time().replace(tzinfo=dt.tzinfo)
    elif isinstance(dt, datetime.date):
        dt = self.t(tzinfo=datetime.timezone.utc)

    if dt.__class__ is self.t:
        return dt  # type: ignore[return-value]

    return self.t(
        hour=dt.hour,
        minute=dt.minute,
        second=dt.second,
        microsecond=dt.microsecond,
        tzinfo=dt.tzinfo,
        fold=dt.fold,
    )

UUIDUnmarshaller

UUIDUnmarshaller(t: type[T], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[UUIDT], Generic[UUIDT]

Unmarshaller that converts an input to a uuid.UUID (or subclasses).

Note

The resolution algorithm is intentionally simple:

  1. Attempt to decode any bytes/string input into a real Python object.
  2. If the value is an integer, pass it into the constructor via the int= param.
  3. Otherwise, pass into the constructor directly.
Tip

While the uuid.UUID constructor supports many different keyword inputs for different types of UUID formats/encodings, we don't have a great method for detecting the correct input. We have moved with the assumption that the two most common formats are a standard string encoding, or an integer encoding.

Parameters:

  • t (type[T]) –

    The root type of this unmarshaller.

  • context (ContextT) –

    The complete type context for this unmarshaller.

  • var (str | None, default: None ) –

    The associated field or parameter name for this unmarshaller (optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[T], context: ContextT, *, var: str | None = None):
    """Construct an unmarshaller instance.

    Args:
        t: The root type of this unmarshaller.
        context: The complete type context for this unmarshaller.
        var: The associated field or parameter name for this unmarshaller (optional).
    """
    self.t = t
    self.origin = inspection.origin(self.t)
    self.context = context
    self.var = var

__call__

__call__(val: Any) -> UUIDT

Unmarshal a value into the bound UUIDT type.

Parameters:

  • val (Any) –

    The input value to unmarshal.

See Also
Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> UUIDT:
    """Unmarshal a value into the bound `UUIDT` type.

    Args:
        val: The input value to unmarshal.

    See Also:
        - [`typelib.serdes.load`][]
    """
    decoded = serdes.load(val)
    if isinstance(decoded, int):
        return self.t(int=decoded)
    if isinstance(decoded, self.t):
        return decoded
    return self.t(decoded)  # type: ignore[arg-type]

UnionUnmarshaller

UnionUnmarshaller(t: type[UnionT], context: ContextT, *, var: str | None = None)

Bases: AbstractUnmarshaller[UnionT], Generic[UnionT]

Unmarshaller that will convert an input to one of the types defined in a typing.Union.

Note

Union deserialization is messy and violates a static type-checking mechanism - for static type-checkers, str | int is equivalent to int | str. This breaks down during unmarshalling for the simple fact that casting something to str will always succeed, so we would never actually unmarshal the input it an int, even if that is the "correct" result.

Our algorithm is intentionally simple:

  1. We iterate through each union member from top to bottom and call the resolved unmarshaller, returning the result.
  2. If any of (ValueError, TypeError, SyntaxError), try again with the next unmarshaller.
  3. If all unmarshallers fail, then we have an invalid input, raise an error.
TL;DR

In order to ensure correctness, you should treat your union members as a stack, sorted from most-strict initialization to least-strict.

Parameters:

  • t (type[UnionT]) –

    The type to unmarshal into.

  • context (ContextT) –

    Any nested type context. Used to resolve the member unmarshallers.

  • var (str | None, default: None ) –

    A variable name for the indicated type annotation (unused, optional).

Source code in src/typelib/unmarshals/routines.py
def __init__(self, t: type[UnionT], context: ContextT, *, var: str | None = None):
    """Constructor.

    Args:
        t: The type to unmarshal into.
        context: Any nested type context. Used to resolve the member unmarshallers.
        var: A variable name for the indicated type annotation (unused, optional).
    """
    super().__init__(t, context, var=var)
    self.stack = inspection.args(t, evaluate=True)
    if inspection.isoptionaltype(t):
        self.stack = (self.stack[-1], *self.stack[:-1])

    self.ordered_routines = [self.context[typ] for typ in self.stack]

__call__

__call__(val: Any) -> UnionT

Unmarshal a value into the bound UnionT.

Parameters:

  • val (Any) –

    The input value to unmarshal.

Raises:

  • ValueError

    If val cannot be unmarshalled into any member type.

Source code in src/typelib/unmarshals/routines.py
def __call__(self, val: tp.Any) -> UnionT:
    """Unmarshal a value into the bound `UnionT`.

    Args:
        val: The input value to unmarshal.

    Raises:
        ValueError: If `val` cannot be unmarshalled into any member type.
    """
    for routine in self.ordered_routines:
        with contextlib.suppress(
            ValueError, TypeError, SyntaxError, AttributeError
        ):
            unmarshalled = routine(val)
            return unmarshalled

    raise ValueError(f"{val!r} is not one of types {self.stack!r}")

unmarshal

unmarshal(t: type[T] | ForwardRef | str, value: Any) -> T

Unmarshal value into typ.

Parameters:

  • t (type[T] | ForwardRef | str) –

    The type annotation or reference to unmarshal into.

  • value (Any) –

    The value to unmarshal.

Source code in src/typelib/unmarshals/api.py
def unmarshal(t: type[T] | refs.ForwardRef | str, value: tp.Any) -> T:
    """Unmarshal `value` into `typ`.

    Args:
        t: The type annotation or reference to unmarshal into.
        value: The value to unmarshal.
    """
    routine = unmarshaller(t)
    unmarshalled = routine(value)
    return unmarshalled

unmarshaller

unmarshaller(t: type[T] | ForwardRef | TypeAliasType | str) -> AbstractUnmarshaller[T]

Get an un-marshaller routine for a given type.

Parameters:

  • t (type[T] | ForwardRef | TypeAliasType | str) –

    The type annotation to generate an unmarshaller for. May be a type, type alias, typing.ForwardRef, or string reference.

Source code in src/typelib/unmarshals/api.py
@compat.cache
def unmarshaller(
    t: type[T] | refs.ForwardRef | compat.TypeAliasType | str,
) -> routines.AbstractUnmarshaller[T]:
    """Get an un-marshaller routine for a given type.

    Args:
        t: The type annotation to generate an unmarshaller for.
             May be a type, type alias, [`typing.ForwardRef`][], or string reference.
    """
    nodes = graph.static_order(t)
    context: ctx.TypeContext[routines.AbstractUnmarshaller] = ctx.TypeContext()
    if not nodes:
        return routines.NoOpUnmarshaller(t=t, context=context, var=None)  # type: ignore[arg-type]

    # "root" type will always be the final node in the sequence.
    root = nodes[-1]
    for node in nodes:
        context[node.type] = _get_unmarshaller(node, context=context)
        context[node.unwrapped] = context[node.type]

    return context[root.type]