Skip to content

hololinked.serializers.msgpack.MsgpackSerializer

Bases: BaseSerializer

Serializer that wraps the msgspec MessagePack serialization protocol.

Recommended serializer for highspeed applications.

Source code in repo/hololinked/hololinked/serializers/msgpack.py 
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class MsgpackSerializer(BaseSerializer):
    """
    Serializer that wraps the msgspec MessagePack serialization protocol.

    Recommended serializer for highspeed applications.
    """

    def __init__(self) -> None:
        super().__init__()
        self.type = msgpack

    codes = dict(NDARRAY_EXT=1)

    def loads(self, data: bytearray | memoryview | bytes) -> Any:
        return msgpack.decode(self.convert_to_bytes(data), ext_hook=self.ext_decode)

    def dumps(self, data: Any) -> bytes:
        return msgpack.encode(data, enc_hook=self.default_encode)

    @classmethod
    def default_encode(cls, obj) -> Any:
        """
        Encode types that MessagePack does not support natively, currently numpy arrays.

        Parameters
        ----------
        obj: Any
            the object to be serialized

        Returns
        -------
        Any
            a MessagePack serializable representation of the object

        Raises
        ------
        TypeError
            if the object cannot be serialized to MessagePack
        """
        if "numpy" in globals() and isinstance(obj, numpy.ndarray):
            buf = io.BytesIO()
            numpy.save(buf, obj, allow_pickle=False)  # use .npy. which stores dtype, shape, order, endianness
            return msgpack.Ext(MsgpackSerializer.codes["NDARRAY_EXT"], buf.getvalue())
        raise TypeError("Given type cannot be converted to MessagePack : {}".format(type(obj)))

    @classmethod
    def ext_decode(cls, code: int, obj: memoryview) -> Any:
        """
        Decode a MessagePack extension type, currently numpy arrays.

        Parameters
        ----------
        code: int
            the extension type code
        obj: memoryview
            the payload of the extension type

        Returns
        -------
        Any
            the decoded object, or the payload itself for an unknown extension code

        Raises
        ------
        ValueError
            if a numpy array is encountered but numpy is not installed
        """
        if code == MsgpackSerializer.codes["NDARRAY_EXT"]:
            if "numpy" in globals():
                return numpy.load(io.BytesIO(obj), allow_pickle=False)
            else:
                raise ValueError("numpy is required to decode numpy array from MessagePack")
        return obj

    @property
    def content_type(self) -> str:
        return "application/msgpack"

Attributes

type instance-attribute 

type = msgpack

content_type property 

content_type: str

codes class-attribute instance-attribute 

codes = dict(NDARRAY_EXT=1)

Functions

__init__ 

__init__() -> None
Source code in repo/hololinked/hololinked/serializers/msgpack.py 
26
27
28
def __init__(self) -> None:
    super().__init__()
    self.type = msgpack

dumps 

dumps(data: Any) -> bytes
Source code in repo/hololinked/hololinked/serializers/msgpack.py 
35
36
def dumps(self, data: Any) -> bytes:
    return msgpack.encode(data, enc_hook=self.default_encode)

loads 

loads(data: bytearray | memoryview | bytes) -> Any
Source code in repo/hololinked/hololinked/serializers/msgpack.py 
32
33
def loads(self, data: bytearray | memoryview | bytes) -> Any:
    return msgpack.decode(self.convert_to_bytes(data), ext_hook=self.ext_decode)

default_encode classmethod 

default_encode(obj) -> Any

Encode types that MessagePack does not support natively, currently numpy arrays.

Parameters:

Name Type Description Default

obj

the object to be serialized

 required

Returns:

Type Description
Any

a MessagePack serializable representation of the object

Raises:

Type Description
TypeError

if the object cannot be serialized to MessagePack
Source code in repo/hololinked/hololinked/serializers/msgpack.py 
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
@classmethod
def default_encode(cls, obj) -> Any:
    """
    Encode types that MessagePack does not support natively, currently numpy arrays.

    Parameters
    ----------
    obj: Any
        the object to be serialized

    Returns
    -------
    Any
        a MessagePack serializable representation of the object

    Raises
    ------
    TypeError
        if the object cannot be serialized to MessagePack
    """
    if "numpy" in globals() and isinstance(obj, numpy.ndarray):
        buf = io.BytesIO()
        numpy.save(buf, obj, allow_pickle=False)  # use .npy. which stores dtype, shape, order, endianness
        return msgpack.Ext(MsgpackSerializer.codes["NDARRAY_EXT"], buf.getvalue())
    raise TypeError("Given type cannot be converted to MessagePack : {}".format(type(obj)))

ext_decode classmethod 

ext_decode(code: int, obj: memoryview) -> Any

Decode a MessagePack extension type, currently numpy arrays.

Parameters:

Name Type Description Default

code

int

the extension type code

 required

obj

memoryview

the payload of the extension type

 required

Returns:

Type Description
Any

the decoded object, or the payload itself for an unknown extension code

Raises:

Type Description
ValueError

if a numpy array is encountered but numpy is not installed
Source code in repo/hololinked/hololinked/serializers/msgpack.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@classmethod
def ext_decode(cls, code: int, obj: memoryview) -> Any:
    """
    Decode a MessagePack extension type, currently numpy arrays.

    Parameters
    ----------
    code: int
        the extension type code
    obj: memoryview
        the payload of the extension type

    Returns
    -------
    Any
        the decoded object, or the payload itself for an unknown extension code

    Raises
    ------
    ValueError
        if a numpy array is encountered but numpy is not installed
    """
    if code == MsgpackSerializer.codes["NDARRAY_EXT"]:
        if "numpy" in globals():
            return numpy.load(io.BytesIO(obj), allow_pickle=False)
        else:
            raise ValueError("numpy is required to decode numpy array from MessagePack")
    return obj