AbstractDTO#
Litestar maintains a suite of DTO factory types that can be used to create DTOs for use with popular data modelling
libraries, such as ORMs. These take a model type as a generic type argument, and create subtypes of
AbstractDTO
that support conversion of that model type to
and from raw bytes.
The following factories are currently available:
Using DTO Factories#
DTO factories are used to create DTOs for use with a particular data modelling library. The following example creates a DTO for use with a SQLAlchemy model:
from datetime import datetime
from sqlalchemy.orm import Mapped
from litestar import Litestar, post
from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
from .my_lib import Base
class User(Base):
name: Mapped[str]
password: Mapped[str]
created_at: Mapped[datetime]
UserDTO = SQLAlchemyDTO[User]
@post("/users", dto=UserDTO, sync_to_thread=False)
def create_user(data: User) -> User:
return data
app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"name":"Litestar User","password":"xyz","created_at":"2023-04-24T00:00:00Z"}
{"id":"371331cd-f0b1-4ce9-8d4c-73fea3639761","name":"Litestar User","password":"xyz","created_at":"2023-04-24T00:00:00Z"}
Here we see that a SQLAlchemy model is used as both the data
and return annotation for the handler, and while
Litestar does not natively support encoding/decoding to/from SQLAlchemy models, through
SQLAlchemyDTO
we can do this.
However, we do have some issues with the above example. Firstly, the user’s password has been returned to them in the
response from the handler. Secondly, the user is able to set the created_at
field on the model, which should only
ever be set once, and defined internally.
Let’s explore how we can configure DTOs to manage scenarios like these.
Marking fields#
The dto_field
function can be used to mark model attributes with DTO-based
configuration.
Fields marked as "private"
or "read-only"
will not be parsed from client data into the user model, and
"private"
fields are never serialized into return data.
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped, mapped_column
4
5from litestar import Litestar, post
6from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
7from litestar.dto import dto_field
8
9from .my_lib import Base
10
11
12class User(Base):
13 name: Mapped[str]
14 password: Mapped[str] = mapped_column(info=dto_field("private"))
15 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
16
17
18UserDTO = SQLAlchemyDTO[User]
19
20
21@post("/users", dto=UserDTO, sync_to_thread=False)
22def create_user(data: User) -> User:
23 # even though the client sent the password and created_at field, it is not in the data object
24 assert "password" not in vars(data)
25 assert "created_at" not in vars(data)
26 # normally the database would set the created_at timestamp
27 data.created_at = datetime.min
28 return data # the response includes the created_at field
29
30
31app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"name":"Litestar User","password":"xyz","created_at":"2023-04-24T00:00:00Z"}
{"created_at":"0001-01-01T00:00:00","id":"135de9ba-ae2c-4443-9838-3e3235921539","name":"Litestar User"}
Excluding fields#
Fields can be explicitly excluded using DTOConfig
.
The following example demonstrates excluding attributes from the serialized response, including excluding fields from nested models.
1from datetime import datetime
2from typing import List
3from uuid import UUID
4
5from sqlalchemy import ForeignKey
6from sqlalchemy.orm import Mapped, mapped_column, relationship
7from typing_extensions import Annotated
8
9from litestar import Litestar, post
10from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
11from litestar.dto import DTOConfig, dto_field
12
13from .my_lib import Base
14
15
16class Address(Base):
17 street: Mapped[str]
18 city: Mapped[str]
19 state: Mapped[str]
20 zip: Mapped[str]
21
22
23class Pets(Base):
24 name: Mapped[str]
25 user_id: Mapped[UUID] = mapped_column(ForeignKey("user.id"))
26
27
28class User(Base):
29 name: Mapped[str]
30 password: Mapped[str] = mapped_column(info=dto_field("private"))
31 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
32 address_id: Mapped[UUID] = mapped_column(ForeignKey("address.id"), info=dto_field("private"))
33 address: Mapped[Address] = relationship(info=dto_field("read-only"))
34 pets: Mapped[List[Pets]] = relationship(info=dto_field("read-only"))
35
36
37UserDTO = SQLAlchemyDTO[User]
38config = DTOConfig(
39 exclude={
40 "id",
41 "address.id",
42 "address.street",
43 "pets.0.id",
44 "pets.0.user_id",
45 }
46)
47ReadUserDTO = SQLAlchemyDTO[Annotated[User, config]]
48
49
50@post("/users", dto=UserDTO, return_dto=ReadUserDTO, sync_to_thread=False)
51def create_user(data: User) -> User:
52 data.created_at = datetime.min
53 data.address = Address(street="123 Main St", city="Anytown", state="NY", zip="12345")
54 data.pets = [Pets(id=1, name="Fido"), Pets(id=2, name="Spot")]
55 return data
56
57
58app = Litestar(route_handlers=[create_user])
1from datetime import datetime
2from uuid import UUID
3
4from sqlalchemy import ForeignKey
5from sqlalchemy.orm import Mapped, mapped_column, relationship
6from typing import Annotated
7
8from litestar import Litestar, post
9from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
10from litestar.dto import DTOConfig, dto_field
11
12from .my_lib import Base
13
14
15class Address(Base):
16 street: Mapped[str]
17 city: Mapped[str]
18 state: Mapped[str]
19 zip: Mapped[str]
20
21
22class Pets(Base):
23 name: Mapped[str]
24 user_id: Mapped[UUID] = mapped_column(ForeignKey("user.id"))
25
26
27class User(Base):
28 name: Mapped[str]
29 password: Mapped[str] = mapped_column(info=dto_field("private"))
30 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
31 address_id: Mapped[UUID] = mapped_column(ForeignKey("address.id"), info=dto_field("private"))
32 address: Mapped[Address] = relationship(info=dto_field("read-only"))
33 pets: Mapped[list[Pets]] = relationship(info=dto_field("read-only"))
34
35
36UserDTO = SQLAlchemyDTO[User]
37config = DTOConfig(
38 exclude={
39 "id",
40 "address.id",
41 "address.street",
42 "pets.0.id",
43 "pets.0.user_id",
44 }
45)
46ReadUserDTO = SQLAlchemyDTO[Annotated[User, config]]
47
48
49@post("/users", dto=UserDTO, return_dto=ReadUserDTO, sync_to_thread=False)
50def create_user(data: User) -> User:
51 data.created_at = datetime.min
52 data.address = Address(street="123 Main St", city="Anytown", state="NY", zip="12345")
53 data.pets = [Pets(id=1, name="Fido"), Pets(id=2, name="Spot")]
54 return data
55
56
57app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"name":"Litestar User","password":"xyz","created_at":"2023-04-24T00:00:00Z"}
{"created_at":"0001-01-01T00:00:00","address":{"city":"Anytown","state":"NY","zip":"12345"},"pets":[{"name":"Fido"},{"name":"Spot"}],"name":"Litestar User"}
Here, the config is created with the exclude parameter, which is a set of strings. Each string represents the path to a
field in the User
object that should be excluded from the output DTO.
config = DTOConfig(
exclude={
"id",
"address.id",
"address.street",
"pets.0.id",
"pets.0.user_id",
}
)
In this example, "id"
represents the id field of the User
object, "address.id"
and "address.street"
represent fields of the Address
object nested inside the User
object, and "pets.0.id"
and
"pets.0.user_id"
represent fields of the Pets
objects nested within the list of User.pets
.
Note
Given a generic type, with an arbitrary number of type parameters (e.g., GenericType[Type0, Type1, ..., TypeN]
),
we use the index of the type parameter to indicate which type the exclusion should refer to. For example, a.0.b
,
excludes the b
field from the first type parameter of a
, a.1.b
excludes the b
field from the second
type parameter of a
, and so on.
Renaming fields#
Fields can be renamed using DTOConfig
. The following example uses the name
userName
client-side, and user
internally.
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped, mapped_column
4from typing_extensions import Annotated
5
6from litestar import Litestar, post
7from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
8from litestar.dto import DTOConfig, dto_field
9
10from .my_lib import Base
11
12
13class User(Base):
14 name: Mapped[str]
15 password: Mapped[str] = mapped_column(info=dto_field("private"))
16 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
17
18
19config = DTOConfig(rename_fields={"name": "userName"})
20UserDTO = SQLAlchemyDTO[Annotated[User, config]]
21
22
23@post("/users", dto=UserDTO, sync_to_thread=False)
24def create_user(data: User) -> User:
25 assert data.name == "Litestar User"
26 data.created_at = datetime.min
27 return data
28
29
30app = Litestar(route_handlers=[create_user])
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped, mapped_column
4from typing import Annotated
5
6from litestar import Litestar, post
7from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
8from litestar.dto import DTOConfig, dto_field
9
10from .my_lib import Base
11
12
13class User(Base):
14 name: Mapped[str]
15 password: Mapped[str] = mapped_column(info=dto_field("private"))
16 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
17
18
19config = DTOConfig(rename_fields={"name": "userName"})
20UserDTO = SQLAlchemyDTO[Annotated[User, config]]
21
22
23@post("/users", dto=UserDTO, sync_to_thread=False)
24def create_user(data: User) -> User:
25 assert data.name == "Litestar User"
26 data.created_at = datetime.min
27 return data
28
29
30app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"userName":"Litestar User","password":"xyz","created_at":"2023-04-24T00:00:00Z"}
{"created_at":"0001-01-01T00:00:00","id":"6e212ffd-26a7-4b4c-bd87-5f834eae8327","userName":"Litestar User"}
Fields can also be renamed using a renaming strategy that will be applied to all fields. The following example uses a pre-defined rename strategy that will convert all field names to camel case on client-side.
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped, mapped_column
4from typing_extensions import Annotated
5
6from litestar import Litestar, post
7from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
8from litestar.dto import DTOConfig, dto_field
9
10from .my_lib import Base
11
12
13class User(Base):
14 first_name: Mapped[str]
15 password: Mapped[str] = mapped_column(info=dto_field("private"))
16 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
17
18
19config = DTOConfig(rename_strategy="camel")
20# another rename strategy with a custom callback:
21# config = DTOConfig(rename_strategy=lambda x: f"-{x}-")
22UserDTO = SQLAlchemyDTO[Annotated[User, config]]
23
24
25@post("/users", dto=UserDTO, sync_to_thread=False)
26def create_user(data: User) -> User:
27 assert data.first_name == "Litestar User"
28 data.created_at = datetime.min
29 return data
30
31
32app = Litestar(route_handlers=[create_user])
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped, mapped_column
4from typing import Annotated
5
6from litestar import Litestar, post
7from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
8from litestar.dto import DTOConfig, dto_field
9
10from .my_lib import Base
11
12
13class User(Base):
14 first_name: Mapped[str]
15 password: Mapped[str] = mapped_column(info=dto_field("private"))
16 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
17
18
19config = DTOConfig(rename_strategy="camel")
20# another rename strategy with a custom callback:
21# config = DTOConfig(rename_strategy=lambda x: f"-{x}-")
22UserDTO = SQLAlchemyDTO[Annotated[User, config]]
23
24
25@post("/users", dto=UserDTO, sync_to_thread=False)
26def create_user(data: User) -> User:
27 assert data.first_name == "Litestar User"
28 data.created_at = datetime.min
29 return data
30
31
32app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"firstName":"Litestar User","password":"xyz","createdAt":"2023-04-24T00:00:00Z"}
{"createdAt":"0001-01-01T00:00:00","id":"5933d91c-68a8-43f3-94b0-7a37dacf419c","firstName":"Litestar User"}
Fields that are directly renamed using rename_fields mapping will be excluded from rename_strategy.
The rename strategy either accepts one of the pre-defined strategies: “camel”, “pascal”, “upper”, “lower”, or it can be provided a callback that accepts the field name as an argument and should return a string.
Type checking#
Factories check that the types to which they are assigned are a subclass of the type provided as the generic type to the
DTO factory. This means that if you have a handler that accepts a User
model, and you assign a UserDTO
factory
to it, the DTO will only accept User
types for “data” and return types.
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped, mapped_column
4
5from litestar import Litestar, post
6from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
7from litestar.dto import dto_field
8
9from .my_lib import Base
10
11
12class User(Base):
13 name: Mapped[str]
14 password: Mapped[str] = mapped_column(info=dto_field("private"))
15 created_at: Mapped[datetime] = mapped_column(info=dto_field("read-only"))
16
17
18class Foo(Base):
19 foo: Mapped[str]
20
21
22UserDTO = SQLAlchemyDTO[User]
23
24
25@post("/users", dto=UserDTO)
26def create_user(data: Foo) -> Foo:
27 return data
28
29
30# This will raise an exception at handler registration time.
31app = Litestar(route_handlers=[create_user])
In the above example, the handler is declared to use UserDTO
which has been type-narrowed with the User
type.
However, we annotate the handler with the Foo
type. This will raise an error such as this at runtime:
litestar.exceptions.dto.InvalidAnnotationException: DTO narrowed with ‘<class ‘docs.examples.data_transfer_objects.factory.type_checking.User’>’, handler type is ‘<class ‘docs.examples.data_transfer_objects.factory.type_checking.Foo’>’
Nested fields#
The depth of related items parsed from client data and serialized into return data can be controlled using the
max_nested_depth
parameter to DTOConfig
.
In this example, we set max_nested_depth=0
for the DTO that handles inbound client data, and leave it at the default
of 1
for the return DTO.
1from __future__ import annotations
2
3from uuid import UUID
4
5from sqlalchemy import ForeignKey
6from sqlalchemy.orm import Mapped, mapped_column, relationship
7from typing_extensions import Annotated
8
9from litestar import Litestar, put
10from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
11from litestar.dto import DTOConfig
12
13from .my_lib import Base
14
15
16class A(Base):
17 b_id: Mapped[UUID] = mapped_column(ForeignKey("b.id"))
18 b: Mapped[B] = relationship(back_populates="a")
19
20
21class B(Base):
22 a: Mapped[A] = relationship(back_populates="b")
23
24
25data_config = DTOConfig(max_nested_depth=0)
26DataDTO = SQLAlchemyDTO[Annotated[A, data_config]]
27
28# default config sets max_nested_depth to 1
29ReturnDTO = SQLAlchemyDTO[A]
30
31
32@put("/a", dto=DataDTO, return_dto=ReturnDTO, sync_to_thread=False)
33def update_a(data: A) -> A:
34 # this shows that "b" was not parsed out of the inbound data
35 assert "b" not in vars(data)
36 # Now we'll create an instance of B and assign it"
37 # This includes a reference back to ``a`` which is not serialized in the return data
38 # because default ``max_nested_depth`` is set to 1
39 data.b = B(id=data.b_id, a=data)
40 return data
41
42
43app = Litestar(route_handlers=[update_a])
1from __future__ import annotations
2
3from uuid import UUID
4
5from sqlalchemy import ForeignKey
6from sqlalchemy.orm import Mapped, mapped_column, relationship
7from typing import Annotated
8
9from litestar import Litestar, put
10from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
11from litestar.dto import DTOConfig
12
13from .my_lib import Base
14
15
16class A(Base):
17 b_id: Mapped[UUID] = mapped_column(ForeignKey("b.id"))
18 b: Mapped[B] = relationship(back_populates="a")
19
20
21class B(Base):
22 a: Mapped[A] = relationship(back_populates="b")
23
24
25data_config = DTOConfig(max_nested_depth=0)
26DataDTO = SQLAlchemyDTO[Annotated[A, data_config]]
27
28# default config sets max_nested_depth to 1
29ReturnDTO = SQLAlchemyDTO[A]
30
31
32@put("/a", dto=DataDTO, return_dto=ReturnDTO, sync_to_thread=False)
33def update_a(data: A) -> A:
34 # this shows that "b" was not parsed out of the inbound data
35 assert "b" not in vars(data)
36 # Now we'll create an instance of B and assign it"
37 # This includes a reference back to ``a`` which is not serialized in the return data
38 # because default ``max_nested_depth`` is set to 1
39 data.b = B(id=data.b_id, a=data)
40 return data
41
42
43app = Litestar(route_handlers=[update_a])
Run it
> curl http://127.0.0.1:8000/a -H Content-Type: application/json -X PUT -d {"id": "6955e63c-c2bc-4707-8fa4-2144d1764746", "b_id": "9cf3518d-7e19-4215-9ec2-e056cac55bf7", "b": {"id": "9cf3518d-7e19-4215-9ec2-e056cac55bf7"}}
{"b_id":"9cf3518d-7e19-4215-9ec2-e056cac55bf7","b":{"id":"9cf3518d-7e19-4215-9ec2-e056cac55bf7"},"id":"6955e63c-c2bc-4707-8fa4-2144d1764746"}
When the handler receives the client data, we can see that the b
field has not been parsed into the A
model that
is injected for our data parameter (line 35).
We then add a B
instance to the data (line 39), which includes a reference back to a
, and from inspection of the
return data can see that b
is included in the response data, however b.a
is not, due to the default
max_nested_depth
of 1
.
Handling unknown fields#
By default, DTOs will silently ignore unknown fields in the source data. This behaviour
can be configured using the forbid_unknown_fields
parameter of the
DTOConfig
. When set to True
a validation
error response will be returned if the data contains a field not defined on the model:
1from __future__ import annotations
2
3from dataclasses import dataclass
4
5from typing_extensions import Annotated
6
7from litestar import Litestar, post
8from litestar.dto import DataclassDTO, DTOConfig
9
10
11@dataclass
12class User:
13 id: str
14
15
16UserDTO = DataclassDTO[Annotated[User, DTOConfig(forbid_unknown_fields=True)]]
17
18
19@post("/users", dto=UserDTO)
20async def create_user(data: User) -> User:
21 return data
22
23
24app = Litestar([create_user])
1from __future__ import annotations
2
3from dataclasses import dataclass
4
5from typing import Annotated
6
7from litestar import Litestar, post
8from litestar.dto import DataclassDTO, DTOConfig
9
10
11@dataclass
12class User:
13 id: str
14
15
16UserDTO = DataclassDTO[Annotated[User, DTOConfig(forbid_unknown_fields=True)]]
17
18
19@post("/users", dto=UserDTO)
20async def create_user(data: User) -> User:
21 return data
22
23
24app = Litestar([create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"id": "1", "name": "Peter"}
{"status_code":400,"detail":"Object contains unknown field `name`"}
DTO Data#
Sometimes we need to be able to access the data that has been parsed and validated by the DTO, but not converted into an instance of our data model.
In the following example, we create a Person
model, that is a dataclass
with 3
required fields, id
, name
, and age
.
We also create a DTO that doesn’t allow clients to set the id
field on the Person
model and set it on the
handler.
1from __future__ import annotations
2
3from dataclasses import dataclass, field
4from uuid import UUID, uuid4
5
6from litestar import Litestar, post
7from litestar.dto import DataclassDTO, DTOConfig
8
9
10@dataclass
11class User:
12 name: str
13 email: str
14 age: int
15 id: UUID = field(default_factory=uuid4)
16
17
18class UserWriteDTO(DataclassDTO[User]):
19 """Don't allow client to set the id."""
20
21 config = DTOConfig(exclude={"id"})
22
23
24# We need a dto for the handler to parse the request data per the configuration, however,
25# we don't need a return DTO as we are returning a dataclass, and Litestar already knows
26# how to serialize dataclasses.
27@post("/users", dto=UserWriteDTO, return_dto=None, sync_to_thread=False)
28def create_user(data: User) -> User:
29 """Create an user."""
30 return data
31
32
33app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"name":"Peter","email": "peter@example.com", "age":41}
{"name":"Peter","email":"peter@example.com","age":41,"id":"7ea9bfda-ee23-4b5d-8af3-bdfcea0aff2f"}
Notice that we get a 500
response from the handler - this is because the DTO has attempted to convert the request
data into a Person
object and failed because it has no value for the required id
field.
One way to handle this is to create different models, e.g., we might create a CreatePerson
model that has no id
field, and decode the client data into that. However, this method can become quite cumbersome when we have a lot of
variability in the data that we accept from clients, for example,
PATCH requests.
This is where the DTOData
class comes in. It is a generic class that accepts the
type of the data that it will contain, and provides useful methods for interacting with that data.
1from dataclasses import dataclass
2from uuid import UUID, uuid4
3
4from litestar import Litestar, post
5from litestar.dto import DataclassDTO, DTOConfig, DTOData
6
7
8@dataclass
9class User:
10 name: str
11 email: str
12 age: int
13 id: UUID
14
15
16class UserWriteDTO(DataclassDTO[User]):
17 """Don't allow client to set the id."""
18
19 config = DTOConfig(exclude={"id"})
20
21
22@post("/users", dto=UserWriteDTO, return_dto=None, sync_to_thread=False)
23def create_user(data: DTOData[User]) -> User:
24 """Create an user."""
25 return data.create_instance(id=uuid4())
26
27
28app = Litestar(route_handlers=[create_user])
Run it
> curl http://127.0.0.1:8000/users -H Content-Type: application/json -d {"name":"Peter", "email": "peter@example.com", "age":41}
{"name":"Peter","email":"peter@example.com","age":41,"id":"6917dfc6-18d3-46b5-815c-896cd99d8d14"}
In the above example, we’ve injected an instance of DTOData
into our handler,
and have used that to create our Person
instance, after augmenting the client data with a server generated id
value.
Consult the Reference Docs
for more information on the methods available.
Providing values for nested data#
To augment data used to instantiate our model instances, we can provide keyword arguments to the
create_instance()
method.
Sometimes we need to provide values for nested data, for example, when creating a new instance of a model that has a nested model with excluded fields.
1from __future__ import annotations
2
3from dataclasses import dataclass
4
5from litestar import Litestar, post
6from litestar.dto import DataclassDTO, DTOConfig, DTOData
7
8
9@dataclass
10class Address:
11 id: int
12 street: str
13
14
15@dataclass
16class Person:
17 id: int
18 name: str
19 age: int
20 address: Address
21
22
23class ReadDTO(DataclassDTO[Person]):
24 config = DTOConfig()
25
26
27class WriteDTO(DataclassDTO[Person]):
28 config = DTOConfig(exclude={"id", "address.id"})
29
30
31@post("/person", dto=WriteDTO, return_dto=ReadDTO, sync_to_thread=False)
32def create_person(data: DTOData[Person]) -> Person:
33 # Logic for persisting the person goes here
34 return data.create_instance(id=1, address__id=2)
35
36
37app = Litestar(route_handlers=[create_person])
Run it
> curl http://127.0.0.1:8000/person -H Content-Type: application/json -d {"name":"Peter","age":41, "address": {"street": "Fake Street"}}
{"id":1,"name":"Peter","age":41,"address":{"id":2,"street":"Fake Street"}}
The double-underscore syntax address__id
passed as a keyword argument to the
create_instance()
method call is used to specify a value for a
nested attribute. In this case, it’s used to provide a value for the id
attribute of the Address
instance nested
within the Person
instance.
This is a common convention in Python for dealing with nested structures. The double underscore can be interpreted as
“traverse through”, so address__id
means “traverse through address to get to id”.
In the context of this script, create_instance(id=1, address__id=2)
is saying “create a new Person
instance from
the client data given an id of 1
, and supplement the client address data with an id of 2
”.
DTO Factory and PATCH requests#
PATCH requests are a special case when it comes to data transfer objects. The reason for this is that we need to be able to accept and validate any subset of the model attributes in the client payload, which requires some special handling internally.
1from __future__ import annotations
2
3from dataclasses import dataclass
4from uuid import UUID
5
6from litestar import Litestar, patch
7from litestar.dto import DataclassDTO, DTOConfig, DTOData
8
9
10@dataclass
11class Person:
12 id: UUID
13 name: str
14 age: int
15
16
17class PatchDTO(DataclassDTO[Person]):
18 """Don't allow client to set the id, and allow partial updates."""
19
20 config = DTOConfig(exclude={"id"}, partial=True)
21
22
23database = {
24 UUID("f32ff2ce-e32f-4537-9dc0-26e7599f1380"): Person(
25 id=UUID("f32ff2ce-e32f-4537-9dc0-26e7599f1380"), name="Peter", age=40
26 )
27}
28
29
30@patch("/person/{person_id:uuid}", dto=PatchDTO, return_dto=None, sync_to_thread=False)
31def update_person(person_id: UUID, data: DTOData[Person]) -> Person:
32 """Create a person."""
33 return data.update_instance(database.get(person_id))
34
35
36app = Litestar(route_handlers=[update_person])
Run it
> curl http://127.0.0.1:8000/person/f32ff2ce-e32f-4537-9dc0-26e7599f1380 -X PATCH -H Content-Type: application/json -d {"name":"Peter Pan"}
{"id":"f32ff2ce-e32f-4537-9dc0-26e7599f1380","name":"Peter Pan","age":40}
The PatchDTO
class is defined for the Person class. The config
attribute of PatchDTO
is set to exclude the
id field, preventing clients from setting it when updating a person, and the partial
attribute is set to True
,
which allows the DTO to accept a subset of the model attributes.
Inside the handler, the DTOData.update_instance
method is called
to update the instance of Person
before returning it.
In our request, we set only the name
property of the Person
, from "Peter"
to "Peter Pan"
and received
the full object - with the modified name - back in the response.
Implicit Private Fields#
Fields that are named with a leading underscore are considered “private” by default. This means that they will not be parsed from client data, and will not be serialized into return data.
1from dataclasses import dataclass
2
3from litestar import Litestar, post
4from litestar.dto import DataclassDTO
5
6
7@dataclass
8class Foo:
9 this_will: str
10 _this_will: str = "Mars"
11
12
13@post("/", dto=DataclassDTO[Foo], sync_to_thread=False)
14def handler(data: Foo) -> Foo:
15 return data
16
17
18app = Litestar(route_handlers=[handler])
Run it
> curl http://127.0.0.1:8000/ -H Content-Type: application/json -d {"bar":"stay","_baz":"go_away!"}
{"status_code":400,"detail":"Object missing required field `this_will`"}
This can be overridden by setting the
DTOConfig.leading_underscore_private
attribute to
False
.
1from dataclasses import dataclass
2
3from litestar import Litestar, post
4from litestar.dto import DataclassDTO, DTOConfig
5
6
7@dataclass
8class Foo:
9 this_will: str
10 _this_will: str = "not_go_away!"
11
12
13class DTO(DataclassDTO[Foo]):
14 config = DTOConfig(underscore_fields_private=False)
15
16
17@post("/", dto=DTO, sync_to_thread=False)
18def handler(data: Foo) -> Foo:
19 return data
20
21
22app = Litestar(route_handlers=[handler])
Run it
> curl http://127.0.0.1:8000/ -H Content-Type: application/json -d {"this_will":"stay","_this_will":"not_go_away!"}
{"this_will":"stay","_this_will":"not_go_away!"}
Wrapping Return Data#
Litestar’s DTO Factory types are versatile enough to manage your data, even when it’s nested within generic wrappers.
The following example demonstrates a route handler that returns DTO managed data wrapped in a generic type. The wrapper is used to deliver additional metadata about the response - in this case, a count of the number of items returned. Read on for an explanation of how to do this yourself.
1from dataclasses import dataclass
2from datetime import datetime
3from typing import Generic, List, TypeVar
4
5from sqlalchemy.orm import Mapped
6
7from litestar import Litestar, get
8from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
9from litestar.dto import DTOConfig
10
11from .my_lib import Base
12
13T = TypeVar("T")
14
15
16@dataclass
17class WithCount(Generic[T]):
18 count: int
19 data: List[T]
20
21
22class User(Base):
23 name: Mapped[str]
24 password: Mapped[str]
25 created_at: Mapped[datetime]
26
27
28class UserDTO(SQLAlchemyDTO[User]):
29 config = DTOConfig(exclude={"password", "created_at"})
30
31
32@get("/users", dto=UserDTO, sync_to_thread=False)
33def get_users() -> WithCount[User]:
34 return WithCount(
35 count=1,
36 data=[
37 User(
38 id=1,
39 name="Litestar User",
40 password="xyz",
41 created_at=datetime.now(),
42 ),
43 ],
44 )
45
46
47app = Litestar(route_handlers=[get_users])
1from dataclasses import dataclass
2from datetime import datetime
3from typing import Generic, TypeVar
4
5from sqlalchemy.orm import Mapped
6
7from litestar import Litestar, get
8from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
9from litestar.dto import DTOConfig
10
11from .my_lib import Base
12
13T = TypeVar("T")
14
15
16@dataclass
17class WithCount(Generic[T]):
18 count: int
19 data: list[T]
20
21
22class User(Base):
23 name: Mapped[str]
24 password: Mapped[str]
25 created_at: Mapped[datetime]
26
27
28class UserDTO(SQLAlchemyDTO[User]):
29 config = DTOConfig(exclude={"password", "created_at"})
30
31
32@get("/users", dto=UserDTO, sync_to_thread=False)
33def get_users() -> WithCount[User]:
34 return WithCount(
35 count=1,
36 data=[
37 User(
38 id=1,
39 name="Litestar User",
40 password="xyz",
41 created_at=datetime.now(),
42 ),
43 ],
44 )
45
46
47app = Litestar(route_handlers=[get_users])
Run it
> curl http://127.0.0.1:8000/users
{"count":1,"data":[{"id":1,"name":"Litestar User"}]}
First, create a generic dataclass to act as your wrapper. This type will contain your data and any additional
attributes you might need. In this example, we have a WithCount
dataclass which has a count
attribute.
The wrapper must be a python generic type with one or more type parameters, and at least one of those type parameters
should describe an instance attribute that will be populated with the data.
from dataclasses import dataclass
from typing import Generic, TypeVar
T = TypeVar("T")
@dataclass
class WithCount(Generic[T]):
count: int
data: List[T]
Now, create a DTO for your data object and configure it using DTOConfig
. In this example, we’re excluding
password
and created_at
from the final output.
from advanced_alchemy.dto import SQLAlchemyDTO
from litestar.dto import DTOConfig
class UserDTO(SQLAlchemyDTO[User]):
config = DTOConfig(exclude={"password", "created_at"})
Then, set up your route handler. This example sets up a /users
endpoint, where a list of User
objects is
returned, wrapped in the WithCount
dataclass.
from litestar import get
@get("/users", dto=UserDTO, sync_to_thread=False)
def get_users() -> WithCount[User]:
return WithCount(
count=1,
data=[
User(
id=1,
name="Litestar User",
password="xyz",
created_at=datetime.now(),
),
],
)
This setup allows the DTO to manage the rendering of User
objects into the response. The DTO Factory type will find
the attribute on the wrapper type that holds the data and perform its serialization operations upon it.
Returning enveloped data is subject to the following constraints:
The type returned from the handler must be a type that Litestar can natively encode.
There can be multiple type arguments to the generic wrapper type, but there must be exactly one type argument to the generic wrapper that is a type supported by the DTO.
Working with Litestar’s Pagination Types#
Litestar offers paginated response wrapper types, and DTO Factory types can handle this out of the box.
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped
4
5from litestar import Litestar, get
6from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
7from litestar.dto import DTOConfig
8from litestar.pagination import ClassicPagination
9
10from .my_lib import Base
11
12
13class User(Base):
14 name: Mapped[str]
15 password: Mapped[str]
16 created_at: Mapped[datetime]
17
18
19class UserDTO(SQLAlchemyDTO[User]):
20 config = DTOConfig(exclude={"password", "created_at"})
21
22
23@get("/users", dto=UserDTO, sync_to_thread=False)
24def get_users() -> ClassicPagination[User]:
25 return ClassicPagination(
26 page_size=10,
27 total_pages=1,
28 current_page=1,
29 items=[
30 User(
31 id=1,
32 name="Litestar User",
33 password="xyz",
34 created_at=datetime.now(),
35 ),
36 ],
37 )
38
39
40app = Litestar(route_handlers=[get_users])
Run it
> curl http://127.0.0.1:8000/users
{"items":[{"id":1,"name":"Litestar User"}],"page_size":10,"current_page":1,"total_pages":1}
The DTO is defined and configured, in our example, we’re excluding password
and created_at
fields from the final
representation of our users.
from advanced_alchemy.dto import SQLAlchemyDTO
from litestar.dto import DTOConfig
class UserDTO(SQLAlchemyDTO[User]):
config = DTOConfig(exclude={"password", "created_at"})
The example sets up a /users
endpoint, where a paginated list of User
objects is returned, wrapped in
ClassicPagination
.
from litestar import get
from litestar.pagination import ClassicPagination
@get("/users", dto=UserDTO, sync_to_thread=False)
def get_users() -> ClassicPagination[User]:
return ClassicPagination(
page_size=10,
total_pages=1,
current_page=1,
items=[
User(
id=1,
name="Litestar User",
password="xyz",
created_at=datetime.now(),
),
],
)
The ClassicPagination
class contains page_size
(number of items per page),
total_pages
(total number of pages), current_page
(current page number), and items
(items for the current
page).
The DTO operates on the data contained in the items
attribute, and the pagination wrapper is handled automatically
by Litestar’s serialization process.
Using Litestar’s Response Type with DTO Factory#
Litestar’s DTO (Data Transfer Object) Factory Types can handle data wrapped in a Response
type.
1from datetime import datetime
2
3from sqlalchemy.orm import Mapped
4
5from litestar import Litestar, Response, get
6from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
7from litestar.dto import DTOConfig
8
9from .my_lib import Base
10
11
12class User(Base):
13 name: Mapped[str]
14 password: Mapped[str]
15 created_at: Mapped[datetime]
16
17
18class UserDTO(SQLAlchemyDTO[User]):
19 config = DTOConfig(exclude={"password", "created_at"})
20
21
22@get("/users", dto=UserDTO, sync_to_thread=False)
23def get_users() -> Response[User]:
24 return Response(
25 content=User(
26 id=1,
27 name="Litestar User",
28 password="xyz",
29 created_at=datetime.now(),
30 ),
31 headers={"X-Total-Count": "1"},
32 )
33
34
35app = Litestar(route_handlers=[get_users])
Run it
> curl http://127.0.0.1:8000/users
{"id":1,"name":"Litestar User"}
We create a DTO for the User
type and configure it using DTOConfig
to exclude password
and created_at
from the serialized output.
from advanced_alchemy.dto import SQLAlchemyDTO
from litestar.dto import DTOConfig
class UserDTO(SQLAlchemyDTO[User]):
config = DTOConfig(exclude={"password", "created_at"})
The example sets up a /users
endpoint where a User
object is returned wrapped in a Response
type.
from litestar import get, Response
@get("/users", dto=UserDTO, sync_to_thread=False)
def get_users() -> Response[User]:
return Response(
content=User(
id=1,
name="Litestar User",
password="xyz",
created_at=datetime.now(),
),
headers={"X-Total-Count": "1"},
)
The Response
object encapsulates the User
object in its content
attribute and allows us to configure the
response received by the client. In this case, we add a custom header.