Question: How to make litestar apply msgspec custom enc_hook / dec_hook functions to serialize/deserialize custom types?

[stewit]

Using msgspec for modelling my entities, I need to control serialization/deserialization for some custom types. Msgspec provides a mechanism for this, which involves writing custom enc_hook / dec_hook functions (see msgspec extending docs), which then can be passed to the encoding/decoding methods of msgspec.

My question is: When I now want to use these models in litestar, where / how can I tell litestar to use my custom enc_hook / dec_hook functions?

[provinzkraut]

You can pass type_encoders / type_decoders on every layer. These allow you to define custom msgspec hooks.

https://docs.litestar.dev/latest/reference/app.html#litestar.app.Litestar.params.type_encoders

[stewit]

Thank you @provinzkraut!

Maybe it's worth a section in the usage part of the documentation how to use these.

For everyone else here is a usage example:

from msgspec import Struct

from typing import Type, Any
from litestar import Litestar, post
from litestar.testing import TestClient


class MyClass:
    """Custom class that litestar/msgspec cannot know how to de/serialize"""

    def __init__(self, a, b):
        self.A = a
        self.B = b


def my_class_predicate(type: Type) -> bool:
    return type is MyClass


def my_class_enc_hook(m: MyClass) -> Any:
    return {"A": str(m.A), "B": str(m.B)}


def my_class_dec_hook(obj: dict) -> MyClass:
    try:
        return MyClass(a=obj["A"], b=["B"])
    except KeyError:
        raise ValueError("Missing key to decode into MyClass")


def general_dec_hook(type: Type, obj: Any) -> Any:
    if my_class_predicate(type):
        return my_class_dec_hook(obj)

    raise NotImplementedError(f"Encountered unknown type during decoding: {str(type)}")


class User(Struct):
    user_id: int
    user_name: str

    my_obj: MyClass


@post("/user", sync_to_thread=False)
def create_user(
    data: User,
) -> User:
    assert isinstance(data.my_obj, MyClass)
    return data


with TestClient(
    Litestar(
        [create_user],
        debug=True,
        type_encoders={
            MyClass: my_class_enc_hook
        },  # tell litestar how to encode other types
        type_decoders=[
            (my_class_predicate, general_dec_hook)
        ],  # tell litestar how to decode
    )
) as client:
    response = client.post(
        "/user",
        json={
            "user_id": 1,
            "user_name": "Some Body",
            "my_obj": my_class_enc_hook(MyClass(a=42, b="t")),
        },
    )
    print(response.text)
    print(f"Status code: {response.status_code}")

[provinzkraut]

Maybe it's worth a section in the usage part of the documentation how to use these.

Definitely. PRs are welcome if you want to add something :)

[stewit]

documentation PR: #3603