Recursive conversion for msgspec

msgspec_recurse.py

from typing import Any, Final

import msgspec
from msgspec import inspect


class _RecurseMarker:
    def __repr__(self):
        return "RECURSE"


RECURSE: Any = _RecurseMarker()
"""Initialize nested struct from parent fields. Use with
``convert_recursive()``."""


COMPOSITE_TYPES: Final = (inspect.StructType, inspect.DataclassType)


def convert_recursive(obj, type, *, from_attributes=True):
    """
    Convert the input object to the specified type, or error accordingly.
    Recurses into nested structs if default=RECURSE is set.
    """

    it = msgspec.convert(obj, type, from_attributes=from_attributes)

    type_info = inspect.type_info(type)
    if not isinstance(type_info, COMPOSITE_TYPES):
        return it

    for fld in type_info.fields:
        if fld.default is not RECURSE:
            continue

        if isinstance(fld.type, COMPOSITE_TYPES):
            value = convert_recursive(
                obj,
                fld.type.cls,
                from_attributes=from_attributes,
            )
        else:
            raise TypeError("RECURSE fields must be structs or dataclasses")

        setattr(it, fld.name, value)

    return it


# Example:
if __name__ == "__main__":
    class ServiceSpec(msgspec.Struct):
        name: str
        labels: dict[str, str]
        image: str
        environment: list[str]

    class DockerContainerSpec(msgspec.Struct):
        image: str = msgspec.field(name="Image")
        environment: list[str] = msgspec.field(name="Env")

    class DockerTaskTemplate(msgspec.Struct):
        _container_spec: DockerContainerSpec = msgspec.field(default=RECURSE, name="ContainerSpec")


    class DockerService(msgspec.Struct):
        name: str = msgspec.field(name="Name")
        labels: dict[str, str] = msgspec.field(name="Labels")
        _task_template: DockerTaskTemplate = msgspec.field(default=RECURSE, name="TaskTemplate")

    assert msgspec.json.encode(
        convert_recursive(
            ServiceSpec(
                name="test",
                labels={},
                image="nginx:alpine",
                environment=["HELLO=world"],
            ),
            type=DockerService
        )
    ) == (
        b'{"Name":"test","Labels":{},"TaskTemplate":{'
        b'"ContainerSpec":{"Image":"nginx:alpine","Env":["HELLO=world"]}}}'
    )