Custom types

Data serialization / deserialization (encoding / decoding) and validation are important parts of any API framework.

In addition to being capable to encode / decode and validate many standard types, litestar supports Python’s builtin dataclasses and libraries like Pydantic and msgspec.

However, sometimes you may need to employ a custom type.

Using type encoders / decoders

Litestar supports a mechanism where you provide encoding and decoding hook functions which translate your type in / to a type that it knows. You can provide them via the type_encoders and type_decoders parameters which can be defined on every layer. For example see the litestar app reference.

Layered architecture

type_encoders and type_decoders are part of Litestar’s layered architecture, which means you can set them on every layer of the application. If you set them on multiple layers, the layer closest to the route handler will take precedence.

You can read more about this here: Layered architecture

Here is an example:

Tell Litestar how to encode and decode a custom type

from typing import Any, Type

from msgspec import Struct

from litestar import Litestar, post


class TenantUser:
    """Custom Type that represents a user associated to a tenant

    Parsed from / serialized to a combined tenant + user id string of the form

        TENANTPREFIX_USERID

    i.e. separated by underscore.
    """

    tenant_prefix: str
    user_id: str

    def __init__(self, tenant_prefix: str, user_id: str) -> None:
        self.tenant_prefix = tenant_prefix
        self.user_id = user_id

    @classmethod
    def from_string(cls, s: str) -> "TenantUser":
        splits = s.split("_", maxsplit=1)
        if len(splits) < 2:
            raise ValueError(
                "Could not split up tenant user id string. "
                "Expecting underscore for separation of tenant prefix and user id."
            )
        return cls(tenant_prefix=splits[0], user_id=splits[1])

    def to_combined_str(self) -> str:
        return self.tenant_prefix + "_" + self.user_id


def tenant_user_type_predicate(type: Type) -> bool:
    return type is TenantUser


def tenant_user_enc_hook(u: TenantUser) -> Any:
    return u.to_combined_str()


def tenant_user_dec_hook(tenant_user_id_str: str) -> TenantUser:
    return TenantUser.from_string(tenant_user_id_str)


def general_dec_hook(type: Type, obj: Any) -> Any:
    if tenant_user_type_predicate(type):
        return tenant_user_dec_hook(obj)

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


class UserAsset(Struct):
    user: TenantUser
    name: str


@post("/asset", sync_to_thread=False)
def create_asset(
    data: UserAsset,
) -> UserAsset:
    assert isinstance(data.user, TenantUser)
    return data


app = Litestar(
    [create_asset],
    type_encoders={TenantUser: tenant_user_enc_hook},  # tell litestar how to encode TenantUser
    type_decoders=[(tenant_user_type_predicate, general_dec_hook)],  # tell litestar how to decode TenantUser
)

Run it

> curl http://127.0.0.1:8000/asset -X POST -H Content-Type: application/json -d {"name":"SomeAsset","user":"TenantA_Somebody"}
{"user":"TenantA_Somebody","name":"SomeAsset"}

Custom Pydantic types

If you use a custom Pydantic type you can use it directly:

Tell Litestar how to encode and decode a custom Pydantic type

from pydantic import BaseModel, BeforeValidator, ConfigDict, PlainSerializer, WithJsonSchema
from typing_extensions import Annotated

from litestar import Litestar, post


class TenantUser:
    """Custom Type that represents a user associated to a tenant

    Parsed from / serialized to a combined tenant + user id string of the form

        TENANTPREFIX_USERID

    i.e. separated by underscore.
    """

    tenant_prefix: str
    user_id: str

    def __init__(self, tenant_prefix: str, user_id: str) -> None:
        self.tenant_prefix = tenant_prefix
        self.user_id = user_id

    @classmethod
    def from_string(cls, s: str) -> "TenantUser":
        splits = s.split("_", maxsplit=1)
        if len(splits) < 2:
            raise ValueError(
                "Could not split up tenant user id string. "
                "Expecting underscore for separation of tenant prefix and user id."
            )
        return cls(tenant_prefix=splits[0], user_id=splits[1])

    def to_combined_str(self) -> str:
        return self.tenant_prefix + "_" + self.user_id


PydAnnotatedTenantUser = Annotated[
    TenantUser,
    BeforeValidator(lambda x: TenantUser.from_string(x)),
    PlainSerializer(lambda x: x.to_combined_str(), return_type=str),
    WithJsonSchema({"type": "string"}, mode="serialization"),
]


class UserAsset(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    user: PydAnnotatedTenantUser
    name: str


@post("/asset", sync_to_thread=False)
def create_asset(
    data: UserAsset,
) -> UserAsset:
    assert isinstance(data.user, TenantUser)
    return data


app = Litestar(
    [create_asset],
)

Run it

> curl http://127.0.0.1:8000/asset -X POST -H Content-Type: application/json -d {"name":"SomeAsset","user":"TenantA_Somebody"}
{"user":"TenantA_Somebody","name":"SomeAsset"}