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 and processes field metadata.

Parameters:

Name Type Description Default
cls Optional[T]

The class to decorate.

 None
tag Optional[str]

An optional field name to use as a tag for identifying the class in a Union.

 None
tag_value Optional[str]

An optional value for the tag field. Defaults to the class name.

 None
Source code in src/lodum/core.py 
 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
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
def lodum(
    cls: Optional[T] = None,
    tag: Optional[str] = None,
    tag_value: Optional[str] = None,
) -> Any:
    """
    A class decorator that marks a class as lodum-enabled and processes field metadata.

    Args:
        cls: The class to decorate.
        tag: An optional field name to use as a tag for identifying the class in a Union.
        tag_value: An optional value for the tag field. Defaults to the class name.
    """

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

        original_init = c.__init__
        init_sig = inspect.signature(original_init)
        fields: Dict[str, Field] = {}

        for param in init_sig.parameters.values():
            if param.name == "self":
                continue

            is_field_spec = isinstance(param.default, Field)

            if is_field_spec:
                field_info = param.default
            else:
                # Create a default Field for params without one, preserving its default value
                default = (
                    param.default if param.default is not param.empty else _MISSING
                )
                field_info = Field(default=default)

            field_info.name = param.name
            field_info.type = param.annotation
            fields[param.name] = field_info

        setattr(c, "_lodum_fields", fields)

        @functools.wraps(original_init)
        def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
            bound_args = init_sig.bind(self, *args, **kwargs)
            bound_args.apply_defaults()

            resolved_args = {}
            for name, value in bound_args.arguments.items():
                if name == "self":
                    continue

                if isinstance(value, Field):
                    if value.has_default:
                        resolved_args[name] = value.get_default()
                else:
                    resolved_args[name] = value

            original_init(self, **resolved_args)

        c.__init__ = new_init  # type: ignore[method-assign]
        register_type(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 
15
16
17
18
19
20
21
22
23
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 .internal import dump
    from .core import BaseDumper

    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 
26
27
28
29
30
31
32
33
34
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 .internal import load
    from .core import BaseLoader

    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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class Dumper(Protocol):
    """
    Defines the interface for a data format dumper (encoder).
    """

    def dump_int(self, value: int) -> Any: ...
    def dump_str(self, value: str) -> Any: ...
    def dump_float(self, value: float) -> Any: ...
    def dump_bool(self, value: bool) -> Any: ...
    def dump_bytes(self, value: bytes) -> Any: ...
    def dump_list(self, value: List[Any]) -> Any: ...
    def dump_dict(self, value: Dict[str, Any]) -> Any: ...
    def begin_struct(self, cls: Type) -> Any: ...
    def end_struct(self) -> Any: ...

Loader

Bases: Protocol

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

Source code in src/lodum/core.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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) -> Optional[TypingUnion[Dict[str, Any], List[Any]]]: ...
    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 Optional[str]

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 Optional[Callable[[], Any]]

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

 None
serializer Optional[Callable[[Any], Any]]

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

 None
deserializer Optional[Callable[[Any], Any]]

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

 None
validate Optional[Union[Callable[[Any], None], List[Callable[[Any], None]]]]

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

 None
Source code in src/lodum/field.py 
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
140
141
142
def field(
    *,
    rename: Optional[str] = None,
    skip_serializing: bool = False,
    default: Any = _MISSING,
    default_factory: Optional[Callable[[], Any]] = None,
    serializer: Optional[Callable[[Any], Any]] = None,
    deserializer: Optional[Callable[[Any], Any]] = None,
    validate: Optional[
        Union[Callable[[Any], None], List[Callable[[Any], 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

Functions

dumps(obj)

Encodes a Python object to a JSON string.

Parameters:

Name Type Description Default
obj Any

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

 required

Returns:

Type Description
str

A JSON string representation of the object.
Source code in src/lodum/json.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
def dumps(obj: Any) -> str:
    """
    Encodes a Python object to a JSON string.

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

    Returns:
        A JSON string representation of the object.
    """
    dumper = JsonDumper()
    dumped_data = dump(obj, dumper)
    return json.dumps(dumped_data)

loads(cls, json_string, max_size=DEFAULT_MAX_SIZE)

Decodes a JSON string into a Python object of the specified type.

Parameters:

Name Type Description Default
cls Type[T]

The class to instantiate.

 required
json_string str

The JSON data to decode.

 required
max_size int

Maximum allowed size of the input string in bytes.

 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.
Source code in src/lodum/json.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def loads(cls: Type[T], json_string: str, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    Decodes a JSON string into a Python object of the specified type.

    Args:
        cls: The class to instantiate.
        json_string: The JSON data to decode.
        max_size: Maximum allowed size of the input string in bytes.

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

    Raises:
        DeserializationError: If the input is invalid or exceeds max_size.
    """
    if len(json_string) > max_size:
        raise DeserializationError(
            f"Input size ({len(json_string)}) exceeds maximum allowed ({max_size})"
        )
    data = json.loads(json_string)
    loader = JsonLoader(data)
    return load(cls, loader)

schema(cls)

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

Source code in src/lodum/json.py 
55
56
57
def schema(cls: Type[Any]) -> Dict[str, Any]:
    """Generates a JSON Schema for a given lodum-enabled class."""
    return generate_schema(cls)

lodum.yaml

Classes

YamlDumper

Bases: BaseDumper

Encodes Python objects into a YAML-compatible intermediate representation.

Source code in src/lodum/yaml.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class YamlDumper(BaseDumper):
    """
    Encodes Python objects into a YAML-compatible intermediate representation.
    """

    def dump_bytes(self, value: bytes) -> 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")

YamlLoader

Bases: BaseLoader

Decodes a YAML-compatible intermediate representation into Python objects.

Source code in src/lodum/yaml.py 
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
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

dumps(obj)

Encodes a Python object to a YAML string.

Parameters:

Name Type Description Default
obj Any

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

 required

Returns:

Type Description
str

A YAML string representation of the object.

Raises:

Type Description
ImportError

If ruamel.yaml is not installed.
Source code in src/lodum/yaml.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def dumps(obj: Any) -> str:
    """
    Encodes a Python object to a YAML string.

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

    Returns:
        A YAML string representation of the object.

    Raises:
        ImportError: If ruamel.yaml is not installed.
    """
    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(obj, dumper)

    with io.StringIO() as string_stream:
        yaml.dump(dumped_data, string_stream)
        return string_stream.getvalue()

loads(cls, yaml_string, max_size=DEFAULT_MAX_SIZE)

Decodes a YAML string into a Python object of the specified type.

Parameters:

Name Type Description Default
cls Type[T]

The class to instantiate.

 required
yaml_string str

The YAML data to decode.

 required
max_size int

Maximum allowed size of the input string in bytes.

 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 ruamel.yaml is not installed.
Source code in src/lodum/yaml.py 
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
def loads(cls: Type[T], yaml_string: str, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    Decodes a YAML string into a Python object of the specified type.

    Args:
        cls: The class to instantiate.
        yaml_string: The YAML data to decode.
        max_size: Maximum allowed size of the input string in bytes.

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

    Raises:
        DeserializationError: If the input is invalid or exceeds max_size.
        ImportError: If ruamel.yaml is not installed.
    """
    if len(yaml_string) > max_size:
        raise DeserializationError(
            f"Input size ({len(yaml_string)}) exceeds maximum allowed ({max_size})"
        )

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

    data = yaml.load(yaml_string)
    loader = YamlLoader(data)
    return load(cls, loader)

schema(cls)

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

Source code in src/lodum/yaml.py 
85
86
87
def schema(cls: Type[Any]) -> Dict[str, Any]:
    """Generates a JSON Schema for a given lodum-enabled class."""
    return generate_schema(cls)

lodum.msgpack

Classes

Functions

dumps(obj)

Encodes a Python object to MsgPack bytes.

Parameters:

Name Type Description Default
obj Any

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

 required

Returns:

Type Description
bytes

The MsgPack-encoded bytes.

Raises:

Type Description
ImportError

If msgpack is not installed.
Source code in src/lodum/msgpack.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def dumps(obj: Any) -> bytes:
    """
    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.
    """
    if msgpack is None:
        raise ImportError(
            "msgpack is required for MsgPack serialization. Install it with 'pip install lodum[msgpack]'."
        )
    dumper = MsgPackDumper()
    dumped_data = dump(obj, dumper)
    return msgpack.packb(dumped_data, use_bin_type=True)

loads(cls, packed_bytes, max_size=DEFAULT_MAX_SIZE)

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

Parameters:

Name Type Description Default
cls Type[T]

The class to instantiate.

 required
packed_bytes bytes

The MsgPack data to decode.

 required
max_size int

Maximum allowed size of the input bytes.

 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 msgpack is not installed.
Source code in src/lodum/msgpack.py 
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
def loads(cls: Type[T], packed_bytes: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    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.
    """
    if len(packed_bytes) > max_size:
        raise DeserializationError(
            f"Input size ({len(packed_bytes)}) exceeds maximum allowed ({max_size})"
        )

    if msgpack is None:
        raise ImportError(
            "msgpack is required for MsgPack deserialization. Install it with 'pip install lodum[msgpack]'."
        )
    try:
        data = msgpack.unpackb(packed_bytes, raw=False)
    except Exception as e:
        raise DeserializationError(f"Failed to parse MsgPack: {e}")
    loader = MsgPackLoader(data)
    return load(cls, loader)

lodum.cbor

Classes

Functions

dumps(obj)

Encodes a Python object to CBOR bytes.

Parameters:

Name Type Description Default
obj Any

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

 required

Returns:

Type Description
bytes

The CBOR-encoded bytes.

Raises:

Type Description
ImportError

If cbor2 is not installed.
Source code in src/lodum/cbor.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def dumps(obj: Any) -> bytes:
    """
    Encodes a Python object to CBOR bytes.

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

    Returns:
        The CBOR-encoded bytes.

    Raises:
        ImportError: If cbor2 is not installed.
    """
    if cbor2 is None:
        raise ImportError(
            "cbor2 is required for CBOR serialization. Install it with 'pip install lodum[cbor]'."
        )
    dumper = CborDumper()
    dumped_data = dump(obj, dumper)
    return cbor2.dumps(dumped_data)

loads(cls, cbor_bytes, max_size=DEFAULT_MAX_SIZE)

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

Parameters:

Name Type Description Default
cls Type[T]

The class to instantiate.

 required
cbor_bytes bytes

The CBOR data to decode.

 required
max_size int

Maximum allowed size of the input bytes.

 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 cbor2 is not installed.
Source code in src/lodum/cbor.py 
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
def loads(cls: Type[T], cbor_bytes: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    Decodes CBOR bytes into a Python object of the specified type.

    Args:
        cls: The class to instantiate.
        cbor_bytes: The CBOR 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 cbor2 is not installed.
    """
    if len(cbor_bytes) > max_size:
        raise DeserializationError(
            f"Input size ({len(cbor_bytes)}) exceeds maximum allowed ({max_size})"
        )

    if cbor2 is None:
        raise ImportError(
            "cbor2 is required for CBOR deserialization. Install it with 'pip install lodum[cbor]'."
        )
    try:
        data = cbor2.loads(cbor_bytes)
    except Exception as e:
        raise DeserializationError(f"Failed to parse CBOR: {e}")
    loader = CborLoader(data)
    return load(cls, loader)

lodum.bson

Classes

Functions

dumps(obj)

Encodes a Python object to BSON bytes.

Parameters:

Name Type Description Default
obj Any

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

 required

Returns:

Type Description
bytes

The BSON-encoded bytes.

Raises:

Type Description
ImportError

If bson (pymongo) is not installed.
Source code in src/lodum/bson.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def dumps(obj: Any) -> bytes:
    """
    Encodes a Python object to BSON bytes.

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

    Returns:
        The BSON-encoded bytes.

    Raises:
        ImportError: If bson (pymongo) is not installed.
    """
    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(obj, dumper)
    # BSON requires a dictionary at the root
    if not isinstance(dumped_data, dict):
        dumped_data = {"_v": dumped_data}
    return bson.encode(dumped_data)

loads(cls, bson_bytes, max_size=DEFAULT_MAX_SIZE)

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

Parameters:

Name Type Description Default
cls Type[T]

The class to instantiate.

 required
bson_bytes bytes

The BSON data to decode.

 required
max_size int

Maximum allowed size of the input bytes.

 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 bson (pymongo) is not installed.
Source code in src/lodum/bson.py 
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
def loads(cls: Type[T], bson_bytes: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    Decodes BSON bytes into a Python object of the specified type.

    Args:
        cls: The class to instantiate.
        bson_bytes: The BSON 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 bson (pymongo) is not installed.
    """
    if len(bson_bytes) > max_size:
        raise DeserializationError(
            f"Input size ({len(bson_bytes)}) exceeds maximum allowed ({max_size})"
        )

    if bson is None:
        raise ImportError(
            "bson (pymongo) is required for BSON deserialization. Install it with 'pip install lodum[bson]'."
        )
    try:
        data = bson.decode(bson_bytes)
    except Exception as e:
        raise DeserializationError(f"Failed to parse BSON: {e}")

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

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

lodum.toml

Classes

Functions

dumps(obj)

Encodes a Python object to a TOML string.

Parameters:

Name Type Description Default
obj Any

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

 required

Returns:

Type Description
str

A TOML string representation of the object.

Raises:

Type Description
ImportError

If tomli-w is not installed.
Source code in src/lodum/toml.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def dumps(obj: Any) -> str:
    """
    Encodes a Python object to a TOML string.

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

    Returns:
        A TOML string representation of the object.

    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(obj, dumper)
    return tomli_w.dumps(dumped_data)

loads(cls, toml_string, max_size=DEFAULT_MAX_SIZE)

Decodes a TOML string into a Python object of the specified type.

Parameters:

Name Type Description Default
cls Type[T]

The class to instantiate.

 required
toml_string str

The TOML data to decode.

 required
max_size int

Maximum allowed size of the input string in bytes.

 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 
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
def loads(cls: Type[T], toml_string: str, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    Decodes a TOML string into a Python object of the specified type.

    Args:
        cls: The class to instantiate.
        toml_string: The TOML data to decode.
        max_size: Maximum allowed size of the input string in bytes.

    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 len(toml_string) > max_size:
        raise DeserializationError(
            f"Input size ({len(toml_string)}) exceeds maximum allowed ({max_size})"
        )

    if tomllib is None:
        raise ImportError(
            "tomllib (or tomli) is required for TOML deserialization. Install it with 'pip install lodum[toml]'."
        )
    try:
        data = tomllib.loads(toml_string)
    except Exception as e:
        if isinstance(e, DeserializationError):
            raise e
        raise DeserializationError(f"Failed to parse TOML: {e}")
    loader = TomlLoader(data)
    return load(cls, loader)

schema(cls)

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

Source code in src/lodum/toml.py 
84
85
86
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 
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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 
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
class ValidationDumper(Dumper):
    """A no-op dumper used only for validation."""

    def dump_int(self, value: int) -> None:
        pass

    def dump_str(self, value: str) -> None:
        pass

    def dump_float(self, value: float) -> None:
        pass

    def dump_bool(self, value: bool) -> None:
        pass

    def dump_bytes(self, value: bytes) -> None:
        pass

    def dump_list(self, value: list[Any]) -> None:
        pass

    def dump_dict(self, value: dict[str, Any]) -> None:
        pass

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

    def end_struct(self) -> None:
        pass

Functions

dumps(obj)

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

Source code in src/lodum/pickle.py 
52
53
54
55
56
57
58
def dumps(obj: Any) -> bytes:
    """
    Encodes a Python object to a pickle byte string, ensuring it is safe.
    """
    validator = ValidationDumper()
    validate_lodum_structure(obj, validator)
    return pickle.dumps(obj)

loads(cls, data, max_size=DEFAULT_MAX_SIZE)

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

Source code in src/lodum/pickle.py 
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
def loads(cls: Type[T], data: bytes, max_size: int = DEFAULT_MAX_SIZE) -> T:
    """
    Decodes a pickle byte string to a Python object, ensuring it is safe.
    """
    if len(data) > max_size:
        raise DeserializationError(
            f"Input size ({len(data)}) exceeds maximum allowed ({max_size})"
        )

    with io.BytesIO(data) as f:
        unpickler = SafeUnpickler(f)
        try:
            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

Validation

lodum.validators

Classes