lodum.msgpack

View Source

  1# SPDX-FileCopyrightText: 2025-present Michael R. Bernstein <zopemaven@gmail.com>
  2#
  3# SPDX-License-Identifier: Apache-2.0
  4try:
  5    import msgpack  # type: ignore[import-untyped]
  6except ImportError:
  7    msgpack = None  # type: ignore
  8from typing import Any, Iterator, Type, TypeVar
  9
 10from .core import Loader, BaseDumper, BaseLoader
 11from .exception import DeserializationError
 12from .internal import dump, load, DEFAULT_MAX_SIZE
 13
 14T = TypeVar("T")
 15
 16# --- Public API ---
 17
 18
 19def dumps(obj: Any) -> bytes:
 20    """
 21    Encodes a Python object to MsgPack bytes.
 22
 23    Args:
 24        obj: The object to encode. Must be lodum-enabled or a supported type.
 25
 26    Returns:
 27        The MsgPack-encoded bytes.
 28
 29    Raises:
 30        ImportError: If msgpack is not installed.
 31    """
 32    if msgpack is None:
 33        raise ImportError(
 34            "msgpack is required for MsgPack serialization. Install it with 'pip install lodum[msgpack]'."
 35        )
 36    dumper = MsgPackDumper()
 37    dumped_data = dump(obj, dumper)
 38    return msgpack.packb(dumped_data, use_bin_type=True)
 39
 40
 41def loads(cls: Type[T], packed_bytes: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
 42    """
 43    Decodes MsgPack bytes into a Python object of the specified type.
 44
 45    Args:
 46        cls: The class to instantiate.
 47        packed_bytes: The MsgPack data to decode.
 48        max_size: Maximum allowed size of the input bytes.
 49
 50    Returns:
 51        An instance of cls populated with the decoded data.
 52
 53    Raises:
 54        DeserializationError: If the input is invalid or exceeds max_size.
 55        ImportError: If msgpack is not installed.
 56    """
 57    if len(packed_bytes) > max_size:
 58        raise DeserializationError(
 59            f"Input size ({len(packed_bytes)}) exceeds maximum allowed ({max_size})"
 60        )
 61
 62    if msgpack is None:
 63        raise ImportError(
 64            "msgpack is required for MsgPack deserialization. Install it with 'pip install lodum[msgpack]'."
 65        )
 66    try:
 67        data = msgpack.unpackb(packed_bytes, raw=False)
 68    except Exception as e:
 69        raise DeserializationError(f"Failed to parse MsgPack: {e}")
 70    loader = MsgPackLoader(data)
 71    return load(cls, loader)
 72
 73
 74# --- MsgPack Dumper Implementation ---
 75
 76
 77class MsgPackDumper(BaseDumper):
 78    pass
 79
 80
 81# --- MsgPack Loader Implementation ---
 82
 83
 84class MsgPackLoader(BaseLoader):
 85    def load_list(self) -> Iterator["Loader"]:
 86        if not isinstance(self._data, list):
 87            raise DeserializationError(
 88                f"Expected list, got {type(self._data).__name__}"
 89            )
 90        return (MsgPackLoader(item) for item in self._data)
 91
 92    def load_dict(self) -> Iterator[tuple[str, "Loader"]]:
 93        if not isinstance(self._data, dict):
 94            raise DeserializationError(
 95                f"Expected dict, got {type(self._data).__name__}"
 96            )
 97        return ((k, MsgPackLoader(v)) for k, v in self._data.items())
 98
 99    def load_bytes_value(self, value: Any) -> bytes:
100        if not isinstance(value, bytes):
101            raise DeserializationError(f"Expected bytes, got {type(value).__name__}")
102        return value

def dumps(obj: Any) -> bytes: View Source

20def dumps(obj: Any) -> bytes:
21    """
22    Encodes a Python object to MsgPack bytes.
23
24    Args:
25        obj: The object to encode. Must be lodum-enabled or a supported type.
26
27    Returns:
28        The MsgPack-encoded bytes.
29
30    Raises:
31        ImportError: If msgpack is not installed.
32    """
33    if msgpack is None:
34        raise ImportError(
35            "msgpack is required for MsgPack serialization. Install it with 'pip install lodum[msgpack]'."
36        )
37    dumper = MsgPackDumper()
38    dumped_data = dump(obj, dumper)
39    return msgpack.packb(dumped_data, use_bin_type=True)

Encodes a Python object to MsgPack bytes.

Args: obj: The object to encode. Must be lodum-enabled or a supported type.

Returns: The MsgPack-encoded bytes.

Raises: ImportError: If msgpack is not installed.

def loads(cls: Type[~T], packed_bytes: bytes, max_size: int = 10485760) -> ~T: View Source

42def loads(cls: Type[T], packed_bytes: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
43    """
44    Decodes MsgPack bytes into a Python object of the specified type.
45
46    Args:
47        cls: The class to instantiate.
48        packed_bytes: The MsgPack data to decode.
49        max_size: Maximum allowed size of the input bytes.
50
51    Returns:
52        An instance of cls populated with the decoded data.
53
54    Raises:
55        DeserializationError: If the input is invalid or exceeds max_size.
56        ImportError: If msgpack is not installed.
57    """
58    if len(packed_bytes) > max_size:
59        raise DeserializationError(
60            f"Input size ({len(packed_bytes)}) exceeds maximum allowed ({max_size})"
61        )
62
63    if msgpack is None:
64        raise ImportError(
65            "msgpack is required for MsgPack deserialization. Install it with 'pip install lodum[msgpack]'."
66        )
67    try:
68        data = msgpack.unpackb(packed_bytes, raw=False)
69    except Exception as e:
70        raise DeserializationError(f"Failed to parse MsgPack: {e}")
71    loader = MsgPackLoader(data)
72    return load(cls, loader)

Decodes MsgPack bytes into a Python object of the specified type.

Args: cls: The class to instantiate. packed_bytes: The MsgPack data to decode. max_size: Maximum allowed size of the input bytes.

Returns: An instance of cls populated with the decoded data.

Raises: DeserializationError: If the input is invalid or exceeds max_size. ImportError: If msgpack is not installed.

class MsgPackDumper(lodum.core.BaseDumper): View Source

78class MsgPackDumper(BaseDumper):
79    pass

Base implementation of the Dumper protocol to reduce duplication.

class MsgPackLoader(lodum.core.BaseLoader): View Source

 85class MsgPackLoader(BaseLoader):
 86    def load_list(self) -> Iterator["Loader"]:
 87        if not isinstance(self._data, list):
 88            raise DeserializationError(
 89                f"Expected list, got {type(self._data).__name__}"
 90            )
 91        return (MsgPackLoader(item) for item in self._data)
 92
 93    def load_dict(self) -> Iterator[tuple[str, "Loader"]]:
 94        if not isinstance(self._data, dict):
 95            raise DeserializationError(
 96                f"Expected dict, got {type(self._data).__name__}"
 97            )
 98        return ((k, MsgPackLoader(v)) for k, v in self._data.items())
 99
100    def load_bytes_value(self, value: Any) -> bytes:
101        if not isinstance(value, bytes):
102            raise DeserializationError(f"Expected bytes, got {type(value).__name__}")
103        return value

Base implementation of the Loader protocol to reduce duplication.

def load_list(self) -> Iterator[lodum.core.Loader]: View Source

86    def load_list(self) -> Iterator["Loader"]:
87        if not isinstance(self._data, list):
88            raise DeserializationError(
89                f"Expected list, got {type(self._data).__name__}"
90            )
91        return (MsgPackLoader(item) for item in self._data)

def load_dict(self) -> Iterator[tuple[str, lodum.core.Loader]]: View Source

93    def load_dict(self) -> Iterator[tuple[str, "Loader"]]:
94        if not isinstance(self._data, dict):
95            raise DeserializationError(
96                f"Expected dict, got {type(self._data).__name__}"
97            )
98        return ((k, MsgPackLoader(v)) for k, v in self._data.items())

def load_bytes_value(self, value: Any) -> bytes: View Source

100    def load_bytes_value(self, value: Any) -> bytes:
101        if not isinstance(value, bytes):
102            raise DeserializationError(f"Expected bytes, got {type(value).__name__}")
103        return value