serde.msgpack

  1"""
  2Serialize and Deserialize in MsgPack format. This module depends on
  3[msgpack](https://pypi.org/project/msgpack/) package.
  4"""
  5
  6from typing import Any, Optional, overload
  7
  8import msgpack
  9
 10from .compat import T
 11from .compat import SerdeError
 12from .de import Deserializer, from_dict, from_tuple
 13from .numpy import encode_numpy
 14from .se import Serializer, to_dict, to_tuple
 15
 16__all__ = ["from_msgpack", "to_msgpack"]
 17
 18
 19class MsgPackSerializer(Serializer[bytes]):
 20    @classmethod
 21    def serialize(
 22        cls, obj: Any, use_bin_type: bool = True, ext_type_code: Optional[int] = None, **opts: Any
 23    ) -> bytes:
 24        if "default" not in opts:
 25            opts["default"] = encode_numpy
 26        if ext_type_code is not None:
 27            obj_bytes = msgpack.packb(obj, use_bin_type=use_bin_type, **opts)
 28            obj_or_ext = msgpack.ExtType(ext_type_code, obj_bytes)
 29        else:
 30            obj_or_ext = obj
 31        return msgpack.packb(obj_or_ext, use_bin_type=use_bin_type, **opts)
 32
 33
 34class MsgPackDeserializer(Deserializer[bytes]):
 35    @classmethod
 36    def deserialize(
 37        cls, data: bytes, raw: bool = False, use_list: bool = False, **opts: Any
 38    ) -> Any:
 39        return msgpack.unpackb(data, raw=raw, use_list=use_list, **opts)
 40
 41
 42def to_msgpack(
 43    obj: Any,
 44    cls: Optional[Any] = None,
 45    se: type[Serializer[bytes]] = MsgPackSerializer,
 46    named: bool = True,
 47    ext_dict: Optional[dict[type[Any], int]] = None,
 48    reuse_instances: bool = False,
 49    convert_sets: bool = True,
 50    **opts: Any,
 51) -> bytes:
 52    """
 53    Serialize the object into MsgPack.
 54
 55    You can pass any serializable `obj`. If `ext_dict` option is specified, `obj` is encoded
 56    as a `msgpack.ExtType` If you supply other keyword arguments, they will be passed in
 57    `msgpack.packb` function.
 58
 59    If `named` is True, field names are preserved, namely the object is encoded as `dict` then
 60    serialized into MsgPack.  If `named` is False, the object is encoded as `tuple` then serialized
 61    into MsgPack. `named=False` will produces compact binary.
 62
 63    If you want to use the other msgpack package, you can subclass `MsgPackSerializer` and
 64    implement your own logic.
 65    """
 66    ext_type_code = None
 67    if ext_dict is not None:
 68        obj_type = type(obj)
 69        ext_type_code = ext_dict.get(obj_type)
 70        if ext_type_code is None:
 71            raise SerdeError(f"Could not find type code for {obj_type.__name__} in ext_dict")
 72
 73    kwargs: Any = {"c": cls, "reuse_instances": reuse_instances, "convert_sets": convert_sets}
 74    dict_or_tuple = to_dict(obj, **kwargs) if named else to_tuple(obj, **kwargs)
 75    return se.serialize(
 76        dict_or_tuple,
 77        ext_type_code=ext_type_code,
 78        **opts,
 79    )
 80
 81
 82@overload
 83def from_msgpack(
 84    c: type[T],
 85    s: bytes,
 86    de: type[Deserializer[bytes]] = MsgPackDeserializer,
 87    named: bool = True,
 88    ext_dict: Optional[dict[int, type[Any]]] = None,
 89    **opts: Any,
 90) -> T: ...
 91
 92
 93@overload
 94def from_msgpack(
 95    c: Any,
 96    s: bytes,
 97    de: type[Deserializer[bytes]] = MsgPackDeserializer,
 98    named: bool = True,
 99    ext_dict: Optional[dict[int, type[Any]]] = None,
100    **opts: Any,
101) -> Any: ...
102
103
104def from_msgpack(
105    c: Any,
106    s: bytes,
107    de: type[Deserializer[bytes]] = MsgPackDeserializer,
108    named: bool = True,
109    ext_dict: Optional[dict[int, type[Any]]] = None,
110    **opts: Any,
111) -> Any:
112    """
113    Deserialize from MsgPack into the object.
114
115    `c` is a class obejct and `s` is MsgPack binary. If `ext_dict` option is specified,
116    `c` is ignored and type is inferred from `msgpack.ExtType` If you supply other keyword
117    arguments, they will be passed in `msgpack.unpackb` function.
118
119    If you want to use the other msgpack package, you can subclass `MsgPackDeserializer`
120    and implement your own logic.
121    """
122    if ext_dict is not None:
123        ext = de.deserialize(s, **opts)
124        ext_type = ext_dict.get(ext.code)
125        if ext_type is None:
126            raise SerdeError(f"Could not find type for code {ext.code} in ext_dict")
127        return from_msgpack(ext_type, ext.data, de, named, **opts)
128    else:
129        from_func = from_dict if named else from_tuple
130        return from_func(c, de.deserialize(s, **opts), reuse_instances=False)