Skip to content

codecs

Interfaces for managing type-enforced wire protocols (codecs).

CodecT module-attribute 

CodecT = TypeAliasType('CodecT', Codec, type_params=(T))

Generic type alias with an upper bound of Codec.

DecoderT module-attribute 

DecoderT: TypeAlias = Callable[[bytes], MarshalledValueT]

Protocol for a wire deserializer.

EncoderT module-attribute 

EncoderT: TypeAlias = Callable[[MarshalledValueT], bytes]

Protocol for a wire serializer.

Codec dataclass 

Codec(marshal: AbstractMarshaller[T], unmarshal: AbstractUnmarshaller[T], encoder: EncoderT, decoder: DecoderT)

Bases: Generic[T]

A standard wire protocol (codec).

This codec enables you to directly encode and decode your data model into your wire protocol.

decoder instance-attribute 

decoder: DecoderT

The decoder used to deserialize a bytes-like object into a Python data structure for marshalling into T.

encoder instance-attribute 

encoder: EncoderT

The encoder used to serialize a marshalled T into bytes.

marshal instance-attribute 

marshal: AbstractMarshaller[T]

The marshaller used to convert an instance of T to a serializable object.

unmarshal instance-attribute 

unmarshal: AbstractUnmarshaller[T]

The unmarshaller used to convert a deserialized object into an instance of T.

decode 

decode(value: bytes) -> T

Decode an instance of T from bytes.

We will first decode the data from bytes using the decoder, then unmarshal the data into an instance of T using unmarshal.

Parameters:

  • value (bytes) –

    The bytes to decode.

Source code in src/typelib/codecs.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
def decode(self, value: bytes) -> T:
    """Decode an instance of `T` from bytes.

    We will first decode the data from bytes using the
    [`decoder`][typelib.Codec.decoder], then unmarshal the data into an
    instance of `T` using [`unmarshal`][typelib.Codec.unmarshal].

    Args:
        value: The bytes to decode.
    """
    decoded = self.decoder(value)
    unmarshalled = self.unmarshal(decoded)
    return unmarshalled

encode 

encode(value: T) -> bytes

Encode an instance of T to bytes.

We will first marshal the given instance using the marshal, then encode the marshalled data into bytes using the encoder.

Parameters:

  • value (T) –

    The instance to encode.

Source code in src/typelib/codecs.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def encode(self, value: T) -> bytes:
    """Encode an instance of `T` to bytes.

    We will first marshal the given instance using the
    [`marshal`][typelib.Codec.marshal], then encode the marshalled data
    into bytes using the [`encoder`][typelib.Codec.encoder].

    Args:
        value: The instance to encode.
    """
    marshalled = self.marshal(value)
    encoded = self.encoder(marshalled)
    return encoded

codec 

codec(t: type[T], *, marshaller: AbstractMarshaller[T] | None = None, unmarshaller: AbstractUnmarshaller[T] | None = None, encoder: EncoderT = compat.json.dumps, decoder: DecoderT = compat.json.loads, codec_cls: type[CodecT[T]] | None = None) -> CodecT[T]

Factory function for creating a Codec instance.

Note

In the simplest case, all that needs be provided is the first parameter (t). We will generate a marshaller, unmarshaller and build a codec. However, we provide ample means for customization:

  • You can pass in a subclass of Codec to codec_cls.
  • You may supply custom marshaller or unmarshaller callables - we will generate one using the high-level APIs from marshals and unmarshals if not supplied.
  • The encoder and decoder default to JSON, using either stdlib json or orjson if available. You may provide custom encoder and decoder callables, the only requirement is they ser/des to/from bytes.

Tip

If you installed the json extra when you installed this library, then you have installed orjson.

Parameters:

  • t (type[T]) –

    The type to create the interchange protocol for.

  • marshaller (AbstractMarshaller[T] | None, default: None ) –

    The marshaller used to marshal inputs into the associated type. (optional)

  • unmarshaller (AbstractUnmarshaller[T] | None, default: None ) –

    The unmarshaller used to unmarshal inputs into the associated type. (optional)

  • encoder (EncoderT, default: dumps ) –

    The encoder for encoding data for over-the-wire (defaults to JSON).

  • decoder (DecoderT, default: loads ) –

    The decoder for decoding data from over-the-wire (defaults to JSON).

  • codec_cls (type[CodecT[T]] | None, default: None ) –

    The codec class definition, if overloading (optional).

Source code in src/typelib/codecs.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@compat.cache
def codec(
    t: type[T],
    *,
    marshaller: marshals.AbstractMarshaller[T] | None = None,
    unmarshaller: unmarshals.AbstractUnmarshaller[T] | None = None,
    encoder: EncoderT = compat.json.dumps,
    decoder: DecoderT = compat.json.loads,
    codec_cls: type[CodecT[T]] | None = None,
) -> CodecT[T]:
    """Factory function for creating a [`Codec`][typelib.Codec] instance.

    Note:
        In the simplest case, all that needs be provided is the first parameter (`t`). We will
        generate a marshaller, unmarshaller and build a codec. However, we provide ample
        means for customization:

        - You can pass in a subclass of [`Codec`][typelib.codecs.Codec] to `codec_cls`.
        - You may supply custom `marshaller` or `unmarshaller` callables - we will generate
          one using the high-level APIs from [`marshals`][typelib.marshals] and
          [`unmarshals`][typelib.unmarshals] if not supplied.
        - The `encoder` and `decoder` default to JSON, using either
          stdlib [`json`][] or [`orjson`](https://github.com/ijl/orjson){.external}
          if available. You may provide custom `encoder` and `decoder` callables, the only
          requirement is they ser/des to/from `bytes`.

        /// tip
        If you installed the `json` extra when you installed this library, then you have
        installed [`orjson`](https://github.com/ijl/orjson){.external}.
        ///

    Args:
        t: The type to create the interchange protocol for.
        marshaller: The marshaller used to marshal inputs into the associated type. (optional)
        unmarshaller: The unmarshaller used to unmarshal inputs into the associated type. (optional)
        encoder: The encoder for encoding data for over-the-wire (defaults to JSON).
        decoder: The decoder for decoding data from over-the-wire (defaults to JSON).
        codec_cls: The codec class definition, if overloading (optional).

    """
    marshal = marshaller or marshals.marshaller(t=t)
    unmarshal = unmarshaller or unmarshals.unmarshaller(t=t)
    cls = codec_cls or Codec
    if inspection.isbytestype(t):
        cdc = cls(
            marshal=marshal,
            unmarshal=unmarshal,
            encoder=lambda v: v,  # type: ignore[arg-type,return-value]
            decoder=lambda v: v,  # type: ignore[arg-type,return-value]
        )
        return cdc
    cdc = cls(
        marshal=marshal,
        unmarshal=unmarshal,
        encoder=encoder,
        decoder=decoder,
    )
    return cdc