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

Async

SQLAlchemy Async Serialization Plugin

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/")
async def add_item(data: TodoItem) -> list[TodoItem]:
    return [data]


app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])

Sync

SQLAlchemy Sync Serialization Plugin

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/", sync_to_thread=False)
def add_item(data: TodoItem) -> list[TodoItem]:
    return [data]


app = 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:

Serialization Plugin

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/")
async def add_item(data: TodoItem) -> list[TodoItem]:
    return [data]


app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])

Data Transfer Object

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemyDTO


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/", dto=SQLAlchemyDTO[TodoItem])
async def add_item(data: TodoItem) -> list[TodoItem]:
    return [data]


app = 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.

Async

SQLAlchemy Async Marking Fields

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.dto import dto_field
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]
    super_secret_value: Mapped[str] = mapped_column(info=dto_field("private"))


@post("/")
async def add_item(data: TodoItem) -> list[TodoItem]:
    data.super_secret_value = "This is a secret"
    return [data]


app = Litestar(route_handlers=[add_item], plugins=[SQLAlchemySerializationPlugin()])

Sync

SQLAlchemy Sync Marking Fields

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.dto import dto_field
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin


class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]
    super_secret_value: Mapped[str] = mapped_column(info=dto_field("private"))


@post("/", sync_to_thread=False)
def add_item(data: TodoItem) -> list[TodoItem]:
    data.super_secret_value = "This is a secret"
    return [data]


app = 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.