lodum.cbor

  1# SPDX-FileCopyrightText: 2025-present Michael R. Bernstein <zopemaven@gmail.com>
  2#
  3# SPDX-License-Identifier: Apache-2.0
  4try:
  5    import cbor2
  6except ImportError:
  7    cbor2 = 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 CBOR bytes.
 22
 23    Args:
 24        obj: The object to encode. Must be lodum-enabled or a supported type.
 25
 26    Returns:
 27        The CBOR-encoded bytes.
 28
 29    Raises:
 30        ImportError: If cbor2 is not installed.
 31    """
 32    if cbor2 is None:
 33        raise ImportError(
 34            "cbor2 is required for CBOR serialization. Install it with 'pip install lodum[cbor]'."
 35        )
 36    dumper = CborDumper()
 37    dumped_data = dump(obj, dumper)
 38    return cbor2.dumps(dumped_data)
 39
 40
 41def loads(cls: Type[T], cbor_bytes: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
 42    """
 43    Decodes CBOR bytes into a Python object of the specified type.
 44
 45    Args:
 46        cls: The class to instantiate.
 47        cbor_bytes: The CBOR 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 cbor2 is not installed.
 56    """
 57    if len(cbor_bytes) > max_size:
 58        raise DeserializationError(
 59            f"Input size ({len(cbor_bytes)}) exceeds maximum allowed ({max_size})"
 60        )
 61
 62    if cbor2 is None:
 63        raise ImportError(
 64            "cbor2 is required for CBOR deserialization. Install it with 'pip install lodum[cbor]'."
 65        )
 66    try:
 67        data = cbor2.loads(cbor_bytes)
 68    except Exception as e:
 69        raise DeserializationError(f"Failed to parse CBOR: {e}")
 70    loader = CborLoader(data)
 71    return load(cls, loader)
 72
 73
 74# --- CBOR Dumper Implementation ---
 75
 76
 77class CborDumper(BaseDumper):
 78    pass
 79
 80
 81# --- CBOR Loader Implementation ---
 82
 83
 84class CborLoader(BaseLoader):
 85    def load_int(self) -> int:
 86        if not isinstance(self._data, int):
 87            raise DeserializationError(f"Expected int, got {type(self._data).__name__}")
 88        return self._data
 89
 90    def load_str(self) -> str:
 91        if not isinstance(self._data, str):
 92            raise DeserializationError(f"Expected str, got {type(self._data).__name__}")
 93        return self._data
 94
 95    def load_float(self) -> float:
 96        if not isinstance(self._data, (float, int)):
 97            raise DeserializationError(
 98                f"Expected float, got {type(self._data).__name__}"
 99            )
100        return float(self._data)
101
102    def load_bool(self) -> bool:
103        if not isinstance(self._data, bool):
104            raise DeserializationError(
105                f"Expected bool, got {type(self._data).__name__}"
106            )
107        return self._data
108
109    def load_list(self) -> Iterator["Loader"]:
110        if not isinstance(self._data, list):
111            raise DeserializationError(
112                f"Expected list, got {type(self._data).__name__}"
113            )
114        return (CborLoader(item) for item in self._data)
115
116    def load_dict(self) -> Iterator[tuple[str, "Loader"]]:
117        if not isinstance(self._data, dict):
118            raise DeserializationError(
119                f"Expected dict, got {type(self._data).__name__}"
120            )
121        return ((k, CborLoader(v)) for k, v in self._data.items())
122
123    def load_bytes_value(self, value: Any) -> bytes:
124        if not isinstance(value, bytes):
125            raise DeserializationError(f"Expected bytes, got {type(value).__name__}")
126        return value