SQLAlchemy Serialization Plugin#
The SQLAlchemy Serialization Plugin allows Litestar to do the work of transforming inbound and outbound data to and from SQLAlchemy models. The plugin takes no arguments, simply instantiate it and pass it to your application.
Example#
SQLAlchemy Async Serialization Plugin#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9
10if TYPE_CHECKING:
11 from typing import List
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/")
24async def add_item(data: TodoItem) -> List[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
SQLAlchemy Async Serialization Plugin#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9
10if TYPE_CHECKING:
11 pass
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/")
24async def add_item(data: TodoItem) -> list[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
SQLAlchemy Sync Serialization Plugin#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9
10if TYPE_CHECKING:
11 from typing import List
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/", sync_to_thread=False)
24def add_item(data: TodoItem) -> List[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
SQLAlchemy Sync Serialization Plugin#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9
10if TYPE_CHECKING:
11 pass
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/", sync_to_thread=False)
24def add_item(data: TodoItem) -> list[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
How it works#
The plugin works by defining a SQLAlchemyDTO class for each
handler data or return annotation that is a SQLAlchemy model, or collection of SQLAlchemy models, that isn’t
otherwise managed by an explicitly defined DTO class.
The following two examples are functionally equivalent:
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9
10if TYPE_CHECKING:
11 from typing import List
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/")
24async def add_item(data: TodoItem) -> List[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9
10if TYPE_CHECKING:
11 pass
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/")
24async def add_item(data: TodoItem) -> list[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
9
10if TYPE_CHECKING:
11 from typing import List
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/", dto=SQLAlchemyDTO[TodoItem])
24async def add_item(data: TodoItem) -> List[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item])
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
9
10if TYPE_CHECKING:
11 pass
12
13
14class Base(DeclarativeBase): ...
15
16
17class TodoItem(Base):
18 __tablename__ = "todo_item"
19 title: Mapped[str] = mapped_column(primary_key=True)
20 done: Mapped[bool]
21
22
23@post("/", dto=SQLAlchemyDTO[TodoItem])
24async def add_item(data: TodoItem) -> list[TodoItem]:
25 return [data]
26
27
28app = Litestar(route_handlers=[add_item])
During registration, the application recognizes that there is no DTO class explicitly defined and determines that the
handler annotations are supported by the SQLAlchemy Serialization Plugin. The plugin is then used to generate a DTO
class for both the data keyword argument and the return annotation.
Configuring data transfer#
As the serialization plugin merely defines DTOs for the handler, we can mark the model fields to control the data that we allow in and out of our application.
SQLAlchemy Async Marking Fields#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9from litestar.dto import dto_field
10
11if TYPE_CHECKING:
12 from typing import List
13
14
15class Base(DeclarativeBase): ...
16
17
18class TodoItem(Base):
19 __tablename__ = "todo_item"
20 title: Mapped[str] = mapped_column(primary_key=True)
21 done: Mapped[bool]
22 super_secret_value: Mapped[str] = mapped_column(info=dto_field("private"))
23
24
25@post("/")
26async def add_item(data: TodoItem) -> List[TodoItem]:
27 data.super_secret_value = "This is a secret"
28 return [data]
29
30
31app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
SQLAlchemy Async Marking Fields#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9from litestar.dto import dto_field
10
11if TYPE_CHECKING:
12 pass
13
14
15class Base(DeclarativeBase): ...
16
17
18class TodoItem(Base):
19 __tablename__ = "todo_item"
20 title: Mapped[str] = mapped_column(primary_key=True)
21 done: Mapped[bool]
22 super_secret_value: Mapped[str] = mapped_column(info=dto_field("private"))
23
24
25@post("/")
26async def add_item(data: TodoItem) -> list[TodoItem]:
27 data.super_secret_value = "This is a secret"
28 return [data]
29
30
31app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
SQLAlchemy Sync Marking Fields#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9from litestar.dto import dto_field
10
11if TYPE_CHECKING:
12 from typing import List
13
14
15class Base(DeclarativeBase): ...
16
17
18class TodoItem(Base):
19 __tablename__ = "todo_item"
20 title: Mapped[str] = mapped_column(primary_key=True)
21 done: Mapped[bool]
22 super_secret_value: Mapped[str] = mapped_column(info=dto_field("private"))
23
24
25@post("/", sync_to_thread=False)
26def add_item(data: TodoItem) -> List[TodoItem]:
27 data.super_secret_value = "This is a secret"
28 return [data]
29
30
31app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
SQLAlchemy Sync Marking Fields#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
6
7from litestar import Litestar, post
8from litestar.contrib.sqlalchemy.plugins import SQLAlchemySerializationPlugin
9from litestar.dto import dto_field
10
11if TYPE_CHECKING:
12 pass
13
14
15class Base(DeclarativeBase): ...
16
17
18class TodoItem(Base):
19 __tablename__ = "todo_item"
20 title: Mapped[str] = mapped_column(primary_key=True)
21 done: Mapped[bool]
22 super_secret_value: Mapped[str] = mapped_column(info=dto_field("private"))
23
24
25@post("/", sync_to_thread=False)
26def add_item(data: TodoItem) -> list[TodoItem]:
27 data.super_secret_value = "This is a secret"
28 return [data]
29
30
31app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])
In the above example, a new attribute called super_secret_value has been added to the model, and a value set for it
in the handler. However, due to “marking” the field as “private”, when the model is serialized, the value is not present
in the response.