Skip to content

API Reference

This page contains the technical documentation for all core modules in lodum.

Core API

lodum

lodum(cls=None, tag=None, tag_value=None)

A class decorator that marks a class as lodum-enabled. Field metadata is processed lazily during first serialization/deserialization to correctly handle forward references and circular dependencies.

Source code in src/lodum/core.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def lodum(
    cls: T | None = None,
    tag: str | None = None,
    tag_value: str | None = None,
) -> Any:
    """
    A class decorator that marks a class as lodum-enabled.
    Field metadata is processed lazily during first serialization/deserialization
    to correctly handle forward references and circular dependencies.
    """

    def decorator(c: T) -> T:
        c._lodum_enabled = True
        c._lodum_tag = tag
        c._lodum_tag_value = tag_value or c.__name__

        register_type(c)

        # Analysis is still officially lazy, but we perform it here
        # to ensure metadata is available for immediate use (e.g. in tests).
        from .compiler.analyzer import _analyze_class

        _analyze_class(c)
        return c

    if cls is None:
        return decorator
    return decorator(cls)

asdict(obj)

Recursively converts a lodum-enabled object into plain Python primitives (dict, list, etc.). This handles renaming, skipping fields, and converting enums/datetimes to values.

Source code in src/lodum/__init__.py 
16
17
18
19
20
21
22
23
24
def asdict(obj: Any) -> Any:
    """
    Recursively converts a lodum-enabled object into plain Python primitives (dict, list, etc.).
    This handles renaming, skipping fields, and converting enums/datetimes to values.
    """
    from .core import BaseDumper
    from .internal import dump

    return dump(obj, BaseDumper())

fromdict(cls, data)

Hydrates a lodum-enabled class from a dictionary or other plain Python primitives. This performs full type validation and nested object instantiation.

Source code in src/lodum/__init__.py 
27
28
29
30
31
32
33
34
35
def fromdict(cls: type[T], data: Any) -> T:
    """
    Hydrates a lodum-enabled class from a dictionary or other plain Python primitives.
    This performs full type validation and nested object instantiation.
    """
    from .core import BaseLoader
    from .internal import load

    return load(cls, BaseLoader(data))

lodum.core

Context

Holds the serialization/deserialization context, including registry and caches. Encapsulating this allows for isolated serialization environments.

Source code in src/lodum/core.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class Context:
    """
    Holds the serialization/deserialization context, including registry and caches.
    Encapsulating this allows for isolated serialization environments.
    """

    def __init__(self, registry: Optional["TypeRegistry"] = None) -> None:
        from .registry import registry as global_registry

        self.registry: TypeRegistry = (
            registry.copy() if registry else global_registry.copy()
        )
        self.dump_cache: dict[type[Any], DumpHandler] = {}
        self.load_cache: dict[type[Any], LoadHandler] = {}
        self.cache_lock = Lock()
        self.name_to_type_cache: dict[str, type[Any]] = {}

Dumper

Bases: Protocol

Defines the interface for a data format dumper (encoder).

Source code in src/lodum/core.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class Dumper(Protocol):
    """
    Defines the interface for a data format dumper (encoder).
    """

    def dump_int(self, value: int, depth: int = 0, seen: set | None = None) -> Any: ...
    def dump_str(self, value: str, depth: int = 0, seen: set | None = None) -> Any: ...
    def dump_float(
        self, value: float, depth: int = 0, seen: set | None = None
    ) -> Any: ...
    def dump_bool(
        self, value: bool, depth: int = 0, seen: set | None = None
    ) -> Any: ...
    def dump_bytes(
        self, value: bytes, depth: int = 0, seen: set | None = None
    ) -> Any: ...
    def dump_buffer(
        self, value: Any, depth: int = 0, seen: set | None = None
    ) -> Any: ...
    def dump_none(self, depth: int = 0, seen: set | None = None) -> Any: ...
    def dump_list(
        self, value: list[Any], depth: int = 0, seen: set | None = None
    ) -> Any: ...
    def dump_dict(
        self, value: dict[str, Any], depth: int = 0, seen: set | None = None
    ) -> Any: ...
    def begin_struct(self, cls: type) -> Any: ...
    def end_struct(self) -> Any: ...
    def field(
        self,
        name: str,
        value: Any,
        handler: Callable[[Any, "Dumper", int, set | None], Any],
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        """Processes a single struct field."""
        ...

    def begin_list(self) -> None:
        """Starts a sequence/list."""
        ...

    def end_list(self) -> Any:
        """Ends a sequence/list."""
        ...

    def list_item(
        self,
        value: Any,
        handler: Callable[[Any, "Dumper", int, set | None], Any],
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        """Processes a single list item."""
        ...

begin_list()

Starts a sequence/list.

Source code in src/lodum/core.py 
147
148
149
def begin_list(self) -> None:
    """Starts a sequence/list."""
    ...

end_list()

Ends a sequence/list.

Source code in src/lodum/core.py 
151
152
153
def end_list(self) -> Any:
    """Ends a sequence/list."""
    ...

field(name, value, handler, depth=0, seen=None)

Processes a single struct field.

Source code in src/lodum/core.py 
136
137
138
139
140
141
142
143
144
145
def field(
    self,
    name: str,
    value: Any,
    handler: Callable[[Any, "Dumper", int, set | None], Any],
    depth: int = 0,
    seen: set | None = None,
) -> None:
    """Processes a single struct field."""
    ...

list_item(value, handler, depth=0, seen=None)

Processes a single list item.

Source code in src/lodum/core.py 
155
156
157
158
159
160
161
162
163
def list_item(
    self,
    value: Any,
    handler: Callable[[Any, "Dumper", int, set | None], Any],
    depth: int = 0,
    seen: set | None = None,
) -> None:
    """Processes a single list item."""
    ...

StreamingDumper

Bases: Dumper

Base class for dumpers that write directly to an IO target.

Source code in src/lodum/core.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
class StreamingDumper(Dumper):
    """
    Base class for dumpers that write directly to an IO target.
    """

    def __init__(self, target: IO[Any]) -> None:
        self._target = target
        self._depth = 0
        self._first_item_stack: list[bool] = [True]

    def write_raw(self, chunk: Any) -> None:
        """Writes pre-encoded data directly to the stream."""
        self._target.write(chunk)

    def begin_struct(self, cls: type) -> Any:
        self._depth += 1
        self._first_item_stack.append(True)
        return None

    def end_struct(self) -> Any:
        self._depth -= 1
        self._first_item_stack.pop()
        return None

    def field(
        self,
        name: str,
        value: Any,
        handler: Callable[[Any, "Dumper", int, set | None], Any],
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        handler(value, self, depth, seen)

    def begin_list(self) -> None:
        self._depth += 1
        self._first_item_stack.append(True)

    def end_list(self) -> Any:
        self._depth -= 1
        self._first_item_stack.pop()
        return None

    def list_item(
        self,
        value: Any,
        handler: Callable[[Any, "Dumper", int, set | None], Any],
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        handler(value, self, depth, seen)

write_raw(chunk)

Writes pre-encoded data directly to the stream.

Source code in src/lodum/core.py 
258
259
260
def write_raw(self, chunk: Any) -> None:
    """Writes pre-encoded data directly to the stream."""
    self._target.write(chunk)

Loader

Bases: Protocol

Defines the interface for a data format loader (decoder).

Source code in src/lodum/core.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
class Loader(Protocol):
    """
    Defines the interface for a data format loader (decoder).
    """

    def load_int(self) -> int: ...
    def load_str(self) -> str: ...
    def load_float(self) -> float: ...
    def load_bool(self) -> bool: ...
    def load_bytes(self) -> bytes: ...
    def load_list(self) -> Iterator["Loader"]: ...
    def load_dict(self) -> Iterator[tuple[str, "Loader"]]: ...
    def load_any(self) -> Any: ...
    def mark(self) -> Any: ...
    def rewind(self, marker: Any) -> None: ...
    def get_dict(self) -> dict[str, Any] | list[Any] | None: ...
    def load_bytes_value(self, value: Any) -> bytes: ...

lodum.field

field(*, rename=None, skip_serializing=False, default=_MISSING, default_factory=None, serializer=None, deserializer=None, validate=None)

Provides metadata to the @lodum decorator for a single field.

Parameters:

Name Type Description Default
rename str | None

The name to use for the field in the output.

 None
skip_serializing bool

If True, the field will not be included in the output.

 False
default Any

A default value to use for the field during decoding if it is missing from the input data.

 _MISSING
default_factory Callable[[], Any] | None

A zero-argument function that will be called to create a default value for a missing field.

 None
serializer Callable[[Any], Any] | None

A function to call to encode the field's value.

 None
deserializer Callable[[Any], Any] | None

A function to call to decode the field's value.

 None
validate Callable[[Any], None] | list[Callable[[Any], None]] | None

A callable or list of callables to validate the field's value during decoding.

 None
Source code in src/lodum/field.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def field(
    *,
    rename: str | None = None,
    skip_serializing: bool = False,
    default: Any = _MISSING,
    default_factory: Callable[[], Any] | None = None,
    serializer: Callable[[Any], Any] | None = None,
    deserializer: Callable[[Any], Any] | None = None,
    validate: Callable[[Any], None] | list[Callable[[Any], None]] | None = None,
) -> Any:
    """
    Provides metadata to the `@lodum` decorator for a single field.

    Args:
        rename: The name to use for the field in the output.
        skip_serializing: If `True`, the field will not be included in the
            output.
        default: A default value to use for the field during decoding
            if it is missing from the input data.
        default_factory: A zero-argument function that will be called to
            create a default value for a missing field.
        serializer: A function to call to encode the field's value.
        deserializer: A function to call to decode the field's value.
        validate: A callable or list of callables to validate the field's value during decoding.
    """
    return Field(
        rename=rename,
        skip_serializing=skip_serializing,
        default=default,
        default_factory=default_factory,
        serializer=serializer,
        deserializer=deserializer,
        validate=validate,
    )

lodum.concurrency

WorkerThread

Bases: SequentialThread

A faux thread intended to use Web Workers or Node worker_threads.

Note: Full implementation requires a complex JS bridge to bootstrap a new Pyodide environment in the worker. This currently serves as a placeholder that executes sequentially to maintain compatibility.

Source code in src/lodum/concurrency.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class WorkerThread(SequentialThread):
    """
    A faux thread intended to use Web Workers or Node worker_threads.

    Note: Full implementation requires a complex JS bridge to bootstrap a
    new Pyodide environment in the worker. This currently serves as a
    placeholder that executes sequentially to maintain compatibility.
    """

    def start(self):
        # TODO: Implement actual Worker spawning when state-sharing is not required.
        super().start()

Data Formats

lodum.json

Classes

JsonStreamingDumper

Bases: StreamingDumper

Writes JSON tokens directly to a stream.

Source code in src/lodum/json.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class JsonStreamingDumper(StreamingDumper):
    """
    Writes JSON tokens directly to a stream.
    """

    def dump_int(self, value: int, depth: int = 0, seen: set | None = None) -> Any:
        self.write_raw(str(value))

    def dump_str(self, value: str, depth: int = 0, seen: set | None = None) -> Any:
        self.write_raw(json.dumps(value))

    def dump_float(self, value: float, depth: int = 0, seen: set | None = None) -> Any:
        self.write_raw(str(value))

    def dump_bool(self, value: bool, depth: int = 0, seen: set | None = None) -> Any:
        self.write_raw("true" if value else "false")

    def dump_none(self, depth: int = 0, seen: set | None = None) -> Any:
        self.write_raw("null")

    def dump_bytes(self, value: bytes, depth: int = 0, seen: set | None = None) -> Any:
        import base64

        encoded = base64.b64encode(value).decode("ascii")
        self.write_raw(json.dumps(encoded))

    def dump_buffer(self, value: Any, depth: int = 0, seen: set | None = None) -> Any:
        if hasattr(value, "tolist"):
            from .internal import dump as _dump

            return _dump(value.tolist(), self, depth + 1, seen)
        if isinstance(value, (bytes, bytearray, memoryview)):
            return self.dump_bytes(bytes(value), depth, seen)
        return value

    def dump_list(
        self, value: list[Any], depth: int = 0, seen: set | None = None
    ) -> Any:
        # Should normally not be called directly if orchestration is used
        from .internal import dump as _dump

        self.begin_list()
        for item in value:
            self.list_item(item, _dump, depth + 1, seen)
        return self.end_list()

    def dump_dict(
        self, value: dict[str, Any], depth: int = 0, seen: set | None = None
    ) -> Any:
        # Should normally not be called directly if orchestration is used
        from .internal import dump as _dump

        self.begin_struct(dict)
        for k, v in value.items():
            self.field(str(k), v, _dump, depth + 1, seen)
        return self.end_struct()

    def begin_struct(self, cls: type) -> Any:
        super().begin_struct(cls)
        self.write_raw("{")
        return None

    def end_struct(self) -> Any:
        self.write_raw("}")
        return super().end_struct()

    def field(
        self,
        name: str,
        value: Any,
        handler: Any,
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        if not self._first_item_stack[-1]:
            self.write_raw(",")
        self._first_item_stack[-1] = False

        self.write_raw(json.dumps(name))
        self.write_raw(":")
        handler(value, self, depth, seen)

    def begin_list(self) -> None:
        super().begin_list()
        self.write_raw("[")

    def end_list(self) -> Any:
        self.write_raw("]")
        return super().end_list()

    def list_item(
        self,
        value: Any,
        handler: Any,
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        if not self._first_item_stack[-1]:
            self.write_raw(",")
        self._first_item_stack[-1] = False

        handler(value, self, depth, seen)

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to JSON.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[str] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for json.dump(s) (e.g., indent). Note: If target is provided, O(1) streaming is used and some formatting kwargs might be ignored in the current implementation.

 {}

Returns:

Type Description
str | None

The JSON string if target is None, otherwise None.
Source code in src/lodum/json.py 
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
def dump(obj: Any, target: IO[str] | Path | None = None, **kwargs: Any) -> str | None:
    """
    Encodes a Python object to JSON.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for json.dump(s) (e.g., indent).
                 Note: If target is provided, O(1) streaming is used and some
                 formatting kwargs might be ignored in the current implementation.

    Returns:
        The JSON string if target is None, otherwise None.
    """
    if target is None:
        # IR Mode for string output
        dumper = JsonDumper()
        dumped_data = dump_internal(obj, dumper)
        return json.dumps(dumped_data, **kwargs)

    # O(1) Streaming Mode
    with _resolve_target(target, "w") as out:
        assert out is not None
        dumper_stream = JsonStreamingDumper(out)
        dump_internal(obj, dumper_stream)
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/json.py 
57
58
59
def dumps(obj: Any, **kwargs: Any) -> str:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes JSON from a string, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source str | IO[Any] | Path

JSON string, file-like object, or Path.

 required
max_size int

Maximum allowed size for string input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls.
Source code in src/lodum/json.py 
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
def load(
    cls: type[T], source: str | IO[Any] | Path, max_size: int = DEFAULT_MAX_SIZE
) -> T:
    """
    Decodes JSON from a string, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: JSON string, file-like object, or Path.
        max_size: Maximum allowed size for string input.

    Returns:
        An instance of cls.
    """
    try:
        with _resolve_source(source, "r") as src:
            if isinstance(src, str):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                data = json.loads(src)
            elif hasattr(src, "read"):
                data = json.load(src)  # type: ignore[arg-type]
            else:
                raise DeserializationError(f"Unsupported source type: {type(src)}")
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise
        raise DeserializationError(f"Failed to parse JSON: {e}")

    loader = JsonLoader(data)
    return load_internal(cls, loader)

load_stream(cls, stream_io)

Legacy alias for stream(cls, source).

Source code in src/lodum/json.py 
130
131
132
def load_stream(cls: type[T], stream_io: IO[bytes]) -> Iterator[T]:
    """Legacy alias for stream(cls, source)."""
    return stream(cls, stream_io)

loads(cls, json_string, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/json.py 
97
98
99
def loads(cls: type[T], json_string: str, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, json_string, **kwargs)

schema(cls)

Generates a JSON Schema for a given lodum-enabled class.

Source code in src/lodum/json.py 
135
136
137
def schema(cls: type[Any]) -> dict[str, Any]:
    """Generates a JSON Schema for a given lodum-enabled class."""
    return generate_schema(cls)

stream(cls, source)

Lazily decodes a stream of JSON objects into instances of cls. Intended for sources containing a top-level array of objects.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate for each item.

 required
source IO[bytes] | Path

A binary stream, file-like object, or Path to a JSON array.

 required

Returns:

Type Description
Iterator[T]

An iterator yielding instances of cls.
Source code in src/lodum/json.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def stream(cls: type[T], source: IO[bytes] | Path) -> Iterator[T]:
    """
    Lazily decodes a stream of JSON objects into instances of `cls`.
    Intended for sources containing a top-level array of objects.

    Args:
        cls: The class to instantiate for each item.
        source: A binary stream, file-like object, or Path to a JSON array.

    Returns:
        An iterator yielding instances of `cls`.
    """
    try:
        import ijson  # type: ignore[import-untyped]
    except ImportError:
        raise RuntimeError(
            "Streaming support requires 'ijson'. Install it with: pip install lodum[ijson]"
        )

    with _resolve_source(source, "rb") as src:
        try:
            # ijson.items yields standard Python dicts for each element.
            for item in ijson.items(src, "item"):
                yield load_internal(cls, JsonLoader(item))
        except ijson.common.JSONError as e:
            raise DeserializationError(f"Streaming JSON error: {e}")

lodum.yaml

Classes

YamlDumper

Bases: BaseDumper

Encodes Python objects into a YAML-compatible intermediate representation.

Source code in src/lodum/yaml.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class YamlDumper(BaseDumper):
    """
    Encodes Python objects into a YAML-compatible intermediate representation.
    """

    def dump_bytes(self, value: bytes, depth: int = 0, seen: set | None = None) -> Any:
        # YAML can handle bytes natively if using certain tags,
        # but for simplicity and cross-format consistency, we'll use base64 like JSON.
        import base64

        return base64.b64encode(value).decode("ascii")

    def dump_buffer(self, value: Any, depth: int = 0, seen: set | None = None) -> Any:
        if hasattr(value, "tolist"):
            return value.tolist()
        if isinstance(value, (bytes, bytearray, memoryview)):
            return self.dump_bytes(bytes(value), depth, seen)
        return value

YamlLoader

Bases: BaseLoader

Decodes a YAML-compatible intermediate representation into Python objects.

Source code in src/lodum/yaml.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class YamlLoader(BaseLoader):
    """
    Decodes a YAML-compatible intermediate representation into Python objects.
    """

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

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

    def load_bytes_value(self, value: Any) -> bytes:
        if isinstance(value, bytes):
            return value
        if not isinstance(value, str):
            raise DeserializationError(f"Expected str, got {type(value).__name__}")
        import base64

        try:
            return base64.b64decode(value)
        except Exception as e:
            raise DeserializationError(f"Failed to decode base64: {e}")

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to YAML.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[str] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for yaml.dump.

 {}

Returns:

Type Description
str | None

The YAML string if target is None, otherwise None.
Source code in src/lodum/yaml.py 
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
def dump(obj: Any, target: IO[str] | Path | None = None, **kwargs: Any) -> str | None:
    """
    Encodes a Python object to YAML.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for yaml.dump.

    Returns:
        The YAML string if target is None, otherwise None.
    """
    if not yaml_available:
        raise ImportError(
            "ruamel.yaml is required for YAML serialization. Install it with 'pip install lodum[yaml]'."
        )

    dumper = YamlDumper()
    dumped_data = dump_internal(obj, dumper)

    with _resolve_target(target, "w") as out:
        if out is None:
            with io.StringIO() as string_stream:
                _get_yaml().dump(dumped_data, string_stream, **kwargs)
                return string_stream.getvalue()
        _get_yaml().dump(dumped_data, out, **kwargs)
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/yaml.py 
76
77
78
def dumps(obj: Any, **kwargs: Any) -> str:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes YAML from a string, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source str | IO[Any] | Path

YAML string, file-like object, or Path.

 required
max_size int

Maximum allowed size for string input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls.
Source code in src/lodum/yaml.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def load(
    cls: type[T], source: str | IO[Any] | Path, max_size: int = DEFAULT_MAX_SIZE
) -> T:
    """
    Decodes YAML from a string, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: YAML string, file-like object, or Path.
        max_size: Maximum allowed size for string input.

    Returns:
        An instance of cls.
    """
    if not yaml_available:
        raise ImportError(
            "ruamel.yaml is required for YAML deserialization. Install it with 'pip install lodum[yaml]'."
        )

    try:
        with _resolve_source(source, "r") as src:
            if isinstance(src, str):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                data = _get_yaml().load(src)
            else:
                data = _get_yaml().load(src)
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise
        raise DeserializationError(f"Failed to parse YAML: {e}")

    loader = YamlLoader(data)
    return load_internal(cls, loader)

loads(cls, yaml_string, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/yaml.py 
119
120
121
def loads(cls: type[T], yaml_string: str, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, yaml_string, **kwargs)

schema(cls)

Generates a JSON Schema for a given lodum-enabled class.

Source code in src/lodum/yaml.py 
146
147
148
def schema(cls: type[Any]) -> dict[str, Any]:
    """Generates a JSON Schema for a given lodum-enabled class."""
    return generate_schema(cls)

stream(cls, source)

Lazily decodes a stream of YAML documents.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate for each item.

 required
source IO[Any] | Path

A stream, file-like object, or Path.

 required

Returns:

Type Description
Iterator[T]

An iterator yielding instances of cls.
Source code in src/lodum/yaml.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def stream(cls: type[T], source: IO[Any] | Path) -> Iterator[T]:
    """
    Lazily decodes a stream of YAML documents.

    Args:
        cls: The class to instantiate for each item.
        source: A stream, file-like object, or Path.

    Returns:
        An iterator yielding instances of `cls`.
    """
    if not yaml_available:
        raise ImportError(
            "ruamel.yaml is required for YAML deserialization. Install it with 'pip install lodum[yaml]'."
        )

    with _resolve_source(source, "r") as src:
        # load_all handles multi-document YAML streams
        for data in _get_yaml().load_all(src):
            yield load_internal(cls, YamlLoader(data))

lodum.msgpack

Classes

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to MsgPack.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[bytes] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for msgpack.packb.

 {}

Returns:

Type Description
bytes | None

The MsgPack bytes if target is None, otherwise None.
Source code in src/lodum/msgpack.py 
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
def dump(
    obj: Any, target: IO[bytes] | Path | None = None, **kwargs: Any
) -> bytes | None:
    """
    Encodes a Python object to MsgPack.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for msgpack.packb.

    Returns:
        The MsgPack bytes if target is None, otherwise None.
    """
    if msgpack is None:
        raise ImportError(
            "msgpack is required for MsgPack serialization. Install it with 'pip install lodum[msgpack]'."
        )
    dumper = MsgPackDumper()
    dumped_data = dump_internal(obj, dumper)

    kwargs.setdefault("use_bin_type", True)

    with _resolve_target(target, "wb") as out:
        if out is None:
            return msgpack.packb(dumped_data, **kwargs)
        out.write(msgpack.packb(dumped_data, **kwargs))
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/msgpack.py 
61
62
63
def dumps(obj: Any, **kwargs: Any) -> bytes:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes MsgPack from bytes, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source bytes | IO[bytes] | Path

MsgPack bytes, file-like object, or Path.

 required
max_size int

Maximum allowed size for bytes input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls.
Source code in src/lodum/msgpack.py 
 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
 96
 97
 98
 99
100
101
102
103
def load(
    cls: type[T],
    source: bytes | IO[bytes] | Path,
    max_size: int = DEFAULT_MAX_SIZE,
) -> T:
    """
    Decodes MsgPack from bytes, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: MsgPack bytes, file-like object, or Path.
        max_size: Maximum allowed size for bytes input.

    Returns:
        An instance of cls.
    """
    if msgpack is None:
        raise ImportError(
            "msgpack is required for MsgPack deserialization. Install it with 'pip install lodum[msgpack]'."
        )

    try:
        with _resolve_source(source, "rb") as src:
            if isinstance(src, (bytes, bytearray)):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                data = msgpack.unpackb(src, raw=False)
            else:
                data = msgpack.unpack(src, raw=False)
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise
        raise DeserializationError(f"Failed to parse MsgPack: {e}")

    loader = MsgPackLoader(data)
    return load_internal(cls, loader)

loads(cls, packed_bytes, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/msgpack.py 
106
107
108
def loads(cls: type[T], packed_bytes: bytes, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, packed_bytes, **kwargs)

stream(cls, source)

Lazily decodes a stream of MsgPack objects. Supports concatenated MsgPack objects (ND-MsgPack style).

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate for each item.

 required
source IO[bytes] | Path

A binary stream, file-like object, or Path.

 required

Returns:

Type Description
Iterator[T]

An iterator yielding instances of cls.
Source code in src/lodum/msgpack.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def stream(cls: type[T], source: IO[bytes] | Path) -> Iterator[T]:
    """
    Lazily decodes a stream of MsgPack objects.
    Supports concatenated MsgPack objects (ND-MsgPack style).

    Args:
        cls: The class to instantiate for each item.
        source: A binary stream, file-like object, or Path.

    Returns:
        An iterator yielding instances of `cls`.
    """
    if msgpack is None:
        raise ImportError(
            "msgpack is required for MsgPack deserialization. Install it with 'pip install lodum[msgpack]'."
        )

    with _resolve_source(source, "rb") as src:
        # Use Unpacker for streaming multiple objects
        unpacker = msgpack.Unpacker(src, raw=False)
        for data in unpacker:
            yield load_internal(cls, MsgPackLoader(data))

lodum.cbor

Classes

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to CBOR.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[bytes] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for cbor2.dump(s).

 {}

Returns:

Type Description
bytes | None

The CBOR bytes if target is None, otherwise None.
Source code in src/lodum/cbor.py 
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
def dump(
    obj: Any, target: IO[bytes] | Path | None = None, **kwargs: Any
) -> bytes | None:
    """
    Encodes a Python object to CBOR.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for cbor2.dump(s).

    Returns:
        The CBOR bytes if target is None, otherwise None.
    """
    if cbor2 is None:
        raise ImportError(
            "cbor2 is required for CBOR serialization. Install it with 'pip install lodum[cbor]'."
        )
    dumper = CborDumper()
    dumped_data = dump_internal(obj, dumper)

    with _resolve_target(target, "wb") as out:
        if out is None:
            return cbor2.dumps(dumped_data, **kwargs)
        cbor2.dump(dumped_data, out, **kwargs)
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/cbor.py 
61
62
63
def dumps(obj: Any, **kwargs: Any) -> bytes:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes CBOR from bytes, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source bytes | IO[bytes] | Path

CBOR bytes, file-like object, or Path.

 required
max_size int

Maximum allowed size for bytes input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls.
Source code in src/lodum/cbor.py 
 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
 96
 97
 98
 99
100
101
102
103
104
105
def load(
    cls: type[T],
    source: bytes | IO[bytes] | Path,
    max_size: int = DEFAULT_MAX_SIZE,
) -> T:
    """
    Decodes CBOR from bytes, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: CBOR bytes, file-like object, or Path.
        max_size: Maximum allowed size for bytes input.

    Returns:
        An instance of cls.
    """
    if cbor2 is None:
        raise ImportError(
            "cbor2 is required for CBOR deserialization. Install it with 'pip install lodum[cbor]'."
        )

    try:
        with _resolve_source(source, "rb") as src:
            if isinstance(src, (bytes, bytearray)):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                data = cbor2.loads(src)
            elif hasattr(src, "read"):
                data = cbor2.load(src)  # type: ignore[arg-type]
            else:
                raise DeserializationError(f"Unsupported source type: {type(src)}")
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise
        raise DeserializationError(f"Failed to parse CBOR: {e}")

    loader = CborLoader(data)
    return load_internal(cls, loader)

loads(cls, cbor_bytes, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/cbor.py 
108
109
110
def loads(cls: type[T], cbor_bytes: bytes, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, cbor_bytes, **kwargs)

stream(cls, source)

Lazily decodes a stream of CBOR objects. Supports concatenated CBOR objects.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate for each item.

 required
source IO[bytes] | Path

A binary stream, file-like object, or Path.

 required

Returns:

Type Description
Iterator[T]

An iterator yielding instances of cls.
Source code in src/lodum/cbor.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def stream(cls: type[T], source: IO[bytes] | Path) -> Iterator[T]:
    """
    Lazily decodes a stream of CBOR objects.
    Supports concatenated CBOR objects.

    Args:
        cls: The class to instantiate for each item.
        source: A binary stream, file-like object, or Path.

    Returns:
        An iterator yielding instances of `cls`.
    """
    if cbor2 is None:
        raise ImportError(
            "cbor2 is required for CBOR deserialization. Install it with 'pip install lodum[cbor]'."
        )

    with _resolve_source(source, "rb") as src:
        if isinstance(src, (bytes, bytearray)):
            src = io.BytesIO(src)

        if not hasattr(src, "read"):
            raise DeserializationError(
                f"Unsupported source type for streaming: {type(src)}"
            )

        decoder = cbor2.CBORDecoder(src)  # type: ignore[arg-type]
        try:
            while True:
                data = decoder.decode()
                yield load_internal(cls, CborLoader(data))
        except (EOFError, cbor2.CBORDecodeEOF):
            pass

lodum.bson

Classes

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to BSON.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[bytes] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for bson.encode.

 {}

Returns:

Type Description
bytes | None

The BSON bytes if target is None, otherwise None.
Source code in src/lodum/bson.py 
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
def dump(
    obj: Any, target: IO[bytes] | Path | None = None, **kwargs: Any
) -> bytes | None:
    """
    Encodes a Python object to BSON.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for bson.encode.

    Returns:
        The BSON bytes if target is None, otherwise None.
    """
    if bson is None:
        raise ImportError(
            "bson (pymongo) is required for BSON serialization. Install it with 'pip install lodum[bson]'."
        )
    dumper = BsonDumper()
    dumped_data = dump_internal(obj, dumper)

    # BSON requires a dictionary at the root
    if not isinstance(dumped_data, dict):
        dumped_data = {"_v": dumped_data}

    with _resolve_target(target, "wb") as out:
        if out is None:
            return bson.encode(dumped_data, **kwargs)
        out.write(bson.encode(dumped_data, **kwargs))
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/bson.py 
65
66
67
def dumps(obj: Any, **kwargs: Any) -> bytes:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes BSON from bytes, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source bytes | IO[bytes] | Path

BSON bytes, file-like object, or Path.

 required
max_size int

Maximum allowed size for bytes input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls.
Source code in src/lodum/bson.py 
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def load(
    cls: type[T],
    source: bytes | IO[bytes] | Path,
    max_size: int = DEFAULT_MAX_SIZE,
) -> T:
    """
    Decodes BSON from bytes, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: BSON bytes, file-like object, or Path.
        max_size: Maximum allowed size for bytes input.

    Returns:
        An instance of cls.
    """
    if bson is None:
        raise ImportError(
            "bson (pymongo) is required for BSON deserialization. Install it with 'pip install lodum[bson]'."
        )

    try:
        with _resolve_source(source, "rb") as src:
            if isinstance(src, (bytes, bytearray)):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                data = bson.decode(src)
            elif hasattr(src, "read"):
                data = bson.decode(src.read())  # type: ignore[arg-type]
            else:
                raise DeserializationError(f"Unsupported source type: {type(src)}")
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise
        raise DeserializationError(f"Failed to parse BSON: {e}")

    # Check if we wrapped a primitive
    if isinstance(data, dict) and "_v" in data and len(data) == 1:
        data = data["_v"]

    loader = BsonLoader(data)
    return load_internal(cls, loader)

loads(cls, bson_bytes, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/bson.py 
116
117
118
def loads(cls: type[T], bson_bytes: bytes, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, bson_bytes, **kwargs)

stream(cls, source)

Lazily decodes a stream of BSON objects. Supports concatenated BSON objects.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate for each item.

 required
source IO[bytes] | Path

A binary stream, file-like object, or Path.

 required

Returns:

Type Description
Iterator[T]

An iterator yielding instances of cls.
Source code in src/lodum/bson.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def stream(cls: type[T], source: IO[bytes] | Path) -> Iterator[T]:
    """
    Lazily decodes a stream of BSON objects.
    Supports concatenated BSON objects.

    Args:
        cls: The class to instantiate for each item.
        source: A binary stream, file-like object, or Path.

    Returns:
        An iterator yielding instances of `cls`.
    """
    if bson is None:
        raise ImportError(
            "bson (pymongo) is required for BSON deserialization. Install it with 'pip install lodum[bson]'."
        )

    with _resolve_source(source, "rb") as src:
        if isinstance(src, (bytes, bytearray)):
            src = io.BytesIO(src)

        if not hasattr(src, "read"):
            raise DeserializationError(
                f"Unsupported source type for streaming: {type(src)}"
            )

        # decode_file_iter handles concatenated BSON documents
        for data in bson.decode_file_iter(src):  # type: ignore[arg-type]
            if isinstance(data, dict) and "_v" in data and len(data) == 1:
                data = data["_v"]
            yield load_internal(cls, BsonLoader(data))

lodum.toml

Classes

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to TOML.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[str] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for tomli_w.dump(s).

 {}

Returns:

Type Description
str | None

The TOML string if target is None, otherwise None.

Raises:

Type Description
ImportError

If tomli-w is not installed.
Source code in src/lodum/toml.py 
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
def dump(obj: Any, target: IO[str] | Path | None = None, **kwargs: Any) -> str | None:
    """
    Encodes a Python object to TOML.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for tomli_w.dump(s).

    Returns:
        The TOML string if target is None, otherwise None.

    Raises:
        ImportError: If tomli-w is not installed.
    """
    if tomli_w is None:
        raise ImportError(
            "tomli-w is required for TOML serialization. Install it with 'pip install lodum[toml]'."
        )
    dumper = TomlDumper()
    dumped_data = dump_internal(obj, dumper)

    with _resolve_target(target, "w") as out:
        if out is None:
            return tomli_w.dumps(dumped_data, **kwargs)
        tomli_w.dump(dumped_data, out, **kwargs)
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/toml.py 
69
70
71
def dumps(obj: Any, **kwargs: Any) -> str:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes TOML from a string, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source str | IO[Any] | Path

TOML string, file-like object, or Path.

 required
max_size int

Maximum allowed size for string input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls populated with the decoded data.

Raises:

Type Description
DeserializationError

If the input is invalid or exceeds max_size.
ImportError

If tomllib (or tomli) is not installed.
Source code in src/lodum/toml.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def load(
    cls: type[T], source: str | IO[Any] | Path, max_size: int = DEFAULT_MAX_SIZE
) -> T:
    """
    Decodes TOML from a string, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: TOML string, file-like object, or Path.
        max_size: Maximum allowed size for string input.

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

    Raises:
        DeserializationError: If the input is invalid or exceeds max_size.
        ImportError: If tomllib (or tomli) is not installed.
    """
    if tomllib is None:
        raise ImportError(
            "tomllib (or tomli) is required for TOML deserialization. Install it with 'pip install lodum[toml]'."
        )

    try:
        with _resolve_source(source, "rb") as src:
            if isinstance(src, str):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                data = tomllib.loads(src)
            elif hasattr(src, "read"):
                data = tomllib.load(src)  # type: ignore[arg-type]
            else:
                raise DeserializationError(f"Unsupported source type: {type(src)}")
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise
        raise DeserializationError(f"Failed to parse TOML: {e}")

    loader = TomlLoader(data)
    return load_internal(cls, loader)

loads(cls, toml_string, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/toml.py 
118
119
120
def loads(cls: type[T], toml_string: str, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, toml_string, **kwargs)

schema(cls)

Generates a JSON Schema for a given lodum-enabled class.

Source code in src/lodum/toml.py 
123
124
125
def schema(cls: type[Any]) -> dict[str, Any]:
    """Generates a JSON Schema for a given lodum-enabled class."""
    return generate_schema(cls)

lodum.pickle

Classes

SafeUnpickler

Bases: Unpickler

A custom unpickler that only allows safe, lodum-enabled classes to be loaded.

Source code in src/lodum/pickle.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class SafeUnpickler(pickle.Unpickler):
    """
    A custom unpickler that only allows safe, lodum-enabled classes to be loaded.
    """

    def find_class(self, module_name: str, class_name: str) -> type:
        if "os" in module_name or "sys" in module_name or "subprocess" in module_name:
            raise pickle.UnpicklingError(f"Unsafe module '{module_name}' is forbidden.")

        SAFE_BUILTINS = {
            "int",
            "float",
            "str",
            "bool",
            "bytes",
            "bytearray",
            "list",
            "tuple",
            "dict",
            "set",
            "frozenset",
            "complex",
            "NoneType",
            "type",
        }

        if module_name == "builtins":
            if class_name in SAFE_BUILTINS and hasattr(builtins, class_name):
                return getattr(builtins, class_name)
            raise pickle.UnpicklingError(f"Unsafe builtin '{class_name}' is forbidden.")

        if module_name == "collections" and class_name in (
            "defaultdict",
            "OrderedDict",
            "Counter",
        ):
            import collections

            return getattr(collections, class_name)

        if module_name == "array" and class_name in ("array", "_array_reconstructor"):
            import array

            return getattr(array, class_name)

        cls = super().find_class(module_name, class_name)

        if getattr(cls, "_lodum_enabled", False):
            return cls

        raise pickle.UnpicklingError(
            f"Attempted to unpickle a non-lodum type: {module_name}.{class_name}"
        )

ValidationDumper

Bases: Dumper

A no-op dumper used only for validation.

Source code in src/lodum/pickle.py 
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
class ValidationDumper(Dumper):
    """A no-op dumper used only for validation."""

    def dump_int(self, value: int, depth: int = 0, seen: set | None = None) -> Any:
        pass

    def dump_str(self, value: str, depth: int = 0, seen: set | None = None) -> Any:
        pass

    def dump_float(self, value: float, depth: int = 0, seen: set | None = None) -> Any:
        pass

    def dump_bool(self, value: bool, depth: int = 0, seen: set | None = None) -> Any:
        pass

    def dump_bytes(self, value: bytes, depth: int = 0, seen: set | None = None) -> Any:
        pass

    def dump_buffer(self, value: Any, depth: int = 0, seen: set | None = None) -> Any:
        pass

    def dump_list(
        self, value: list[Any], depth: int = 0, seen: set | None = None
    ) -> Any:
        pass

    def dump_dict(
        self, value: dict[str, Any], depth: int = 0, seen: set | None = None
    ) -> Any:
        pass

    def begin_struct(self, cls: type[Any]) -> Any:
        return {}  # Return a dummy dict

    def end_struct(self) -> Any:
        pass

    def field(
        self,
        name: str,
        value: Any,
        handler: Any,
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        handler(value, self, depth, seen)

    def begin_list(self) -> None:
        pass

    def end_list(self) -> Any:
        pass

    def list_item(
        self,
        value: Any,
        handler: Any,
        depth: int = 0,
        seen: set | None = None,
    ) -> None:
        handler(value, self, depth, seen)

    def dump_none(self, depth: int = 0, seen: set | None = None) -> Any:
        pass

Functions:

dump(obj, target=None, **kwargs)

Encodes a Python object to a pickle byte string, ensuring it is safe.

Parameters:

Name Type Description Default
obj Any

The object to encode.

 required
target IO[bytes] | Path | None

Optional file-like object or Path to write to.

 None
**kwargs Any

Additional arguments for pickle.dump(s).

 {}

Returns:

Type Description
bytes | None

The pickle bytes if target is None, otherwise None.
Source code in src/lodum/pickle.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def dump(
    obj: Any, target: IO[bytes] | Path | None = None, **kwargs: Any
) -> bytes | None:
    """
    Encodes a Python object to a pickle byte string, ensuring it is safe.

    Args:
        obj: The object to encode.
        target: Optional file-like object or Path to write to.
        **kwargs: Additional arguments for pickle.dump(s).

    Returns:
        The pickle bytes if target is None, otherwise None.
    """
    validator = ValidationDumper()
    validate_lodum_structure(obj, validator)

    with _resolve_target(target, "wb") as out:
        if out is None:
            return pickle.dumps(obj, **kwargs)
        pickle.dump(obj, out, **kwargs)
        return None

dumps(obj, **kwargs)

Legacy alias for dump(obj).

Source code in src/lodum/pickle.py 
116
117
118
def dumps(obj: Any, **kwargs: Any) -> bytes:
    """Legacy alias for dump(obj)."""
    return dump(obj, **kwargs)  # type: ignore

load(cls, source, max_size=DEFAULT_MAX_SIZE)

Decodes a pickle from bytes, stream, or file into a Python object.

Parameters:

Name Type Description Default
cls type[T]

The class to instantiate.

 required
source bytes | IO[bytes] | Path

pickle bytes, file-like object, or Path.

 required
max_size int

Maximum allowed size for bytes input.

 DEFAULT_MAX_SIZE

Returns:

Type Description
T

An instance of cls.
Source code in src/lodum/pickle.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def load(
    cls: type[T],
    source: bytes | IO[bytes] | Path,
    max_size: int = DEFAULT_MAX_SIZE,
) -> T:
    """
    Decodes a pickle from bytes, stream, or file into a Python object.

    Args:
        cls: The class to instantiate.
        source: pickle bytes, file-like object, or Path.
        max_size: Maximum allowed size for bytes input.

    Returns:
        An instance of cls.
    """
    try:
        with _resolve_source(source, "rb") as src:
            f: IO[bytes]
            if isinstance(src, (bytes, bytearray)):
                if len(src) > max_size:
                    raise DeserializationError(
                        f"Input size ({len(src)}) exceeds maximum allowed ({max_size})"
                    )
                # Unpickler doesn't take bytes directly
                f = io.BytesIO(src)
            elif hasattr(src, "read"):
                f = src  # type: ignore[assignment]
            else:
                raise DeserializationError(f"Unsupported source type: {type(src)}")

            unpickler = SafeUnpickler(f)
            obj = unpickler.load()
    except (
        pickle.UnpicklingError,
        AttributeError,
        ImportError,
        IndexError,
        TypeError,
    ) as e:
        raise DeserializationError(f"Failed to unpickle data: {e}")

    if not isinstance(obj, cls):
        raise DeserializationError(
            f"Deserialized object is of type {type(obj).__name__}, but expected {cls.__name__}"
        )

    return obj

loads(cls, data, **kwargs)

Legacy alias for load(cls, source).

Source code in src/lodum/pickle.py 
229
230
231
def loads(cls: type[T], data: bytes, **kwargs: Any) -> T:
    """Legacy alias for load(cls, source)."""
    return load(cls, data, **kwargs)

Validation

lodum.validators

Classes