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
)
Tell Litestar how to encode and decode a custom type#
from typing import Any
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],
)
Tell Litestar how to encode and decode a custom Pydantic type#
from pydantic import BaseModel, BeforeValidator, ConfigDict, PlainSerializer, WithJsonSchema
from typing 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"}