SQLAlchemy Init Plugin

The SQLAlchemyInitPlugin adds functionality to the application that supports using Litestar with SQLAlchemy.

The plugin:

• Makes the SQLAlchemy engine and session available via dependency injection.

• Manages the SQLAlchemy engine and session factory in the application’s state.

• Configures a before_send handler that is called before sending a response.

• Includes relevant names in the signature namespace to aid resolving annotated types.

Dependencies

The plugin makes the engine and session available for injection.

Async

SQLAlchemy Async Dependencies

from __future__ import annotations

from typing import TYPE_CHECKING

from sqlalchemy import select

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin

if TYPE_CHECKING:

    from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession


@post("/")
async def handler(db_session: AsyncSession, db_engine: AsyncEngine) -> tuple[int, int]:
    one = (await db_session.execute(select(1))).scalar()

    async with db_engine.begin() as conn:
        two = (await conn.execute(select(2))).scalar()

    return one, two


config = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///async.sqlite")
plugin = SQLAlchemyInitPlugin(config=config)
app = Litestar(route_handlers=[handler], plugins=[plugin])

Sync

SQLAlchemy Sync Dependencies

from __future__ import annotations

from typing import TYPE_CHECKING

from sqlalchemy import select

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemyInitPlugin, SQLAlchemySyncConfig

if TYPE_CHECKING:

    from sqlalchemy import Engine
    from sqlalchemy.orm import Session


@post("/", sync_to_thread=True)
def handler(db_session: Session, db_engine: Engine) -> tuple[int, int]:
    one = db_session.execute(select(1)).scalar()

    with db_engine.begin() as conn:
        two = conn.execute(select(2)).scalar()

    return one, two


config = SQLAlchemySyncConfig(connection_string="sqlite:///sync.sqlite")
plugin = SQLAlchemyInitPlugin(config=config)
app = Litestar(route_handlers=[handler], plugins=[plugin])

The above example illustrates how to access the engine and session in the handler, and like all other dependencies, they can also be injected into other dependency functions.

Renaming the dependencies

You can change the name that the engine and session are bound to by setting the engine_dependency_key and session_dependency_key attributes on the plugin configuration.

Configuring the before send handler

The plugin configures a before_send handler that is called before sending a response. The default handler closes the session and removes it from the connection scope.

You can change the handler by setting the before_send_handler attribute on the configuration object. For example, an alternate handler is available that will also commit the session on success and rollback upon failure.

Async

SQLAlchemy Async Before Send Handler

from __future__ import annotations

from litestar import Litestar
from litestar.plugins.sqlalchemy import (
    SQLAlchemyAsyncConfig,
    SQLAlchemyInitPlugin,
    async_autocommit_before_send_handler,
)

config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///:memory:", before_send_handler=async_autocommit_before_send_handler
)
plugin = SQLAlchemyInitPlugin(config=config)

app = Litestar(route_handlers=[], plugins=[plugin])

Sync

SQLAlchemy Sync Before Send Handler

from __future__ import annotations

from litestar import Litestar
from litestar.plugins.sqlalchemy import SQLAlchemyInitPlugin, SQLAlchemySyncConfig, sync_autocommit_before_send_handler

config = SQLAlchemySyncConfig(
    connection_string="sqlite:///:memory:",
    before_send_handler=sync_autocommit_before_send_handler,
)
plugin = SQLAlchemyInitPlugin(config=config)

app = Litestar(route_handlers=[], plugins=[plugin])

Configuring the plugins

Both the SQLAlchemyAsyncConfig and the SQLAlchemySyncConfig have an engine_config attribute that is used to configure the engine. The engine_config attribute is an instance of EngineConfig and exposes all of the configuration options available to the SQLAlchemy engine.

The SQLAlchemyAsyncConfig class and the SQLAlchemySyncConfig class also have a session_config attribute that is used to configure the session. This is either an instance of AsyncSessionConfig or SyncSessionConfig depending on the type of config object. These classes expose all of the configuration options available to the SQLAlchemy session.

Finally, the SQLAlchemyAsyncConfig class and the SQLAlchemySyncConfig class expose configuration options to control their behavior.

Consult the reference documentation for more information.

Example

The below example is a complete demonstration of use of the init plugin. Readers who are familiar with the prior section may note the additional complexity involved in managing the conversion to and from SQLAlchemy objects within the handlers. Read on to see how this increased complexity is efficiently handled by the SQLAlchemySerializationPlugin.

Async

SQLAlchemy Async Init Plugin Example

from __future__ import annotations

from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin

if TYPE_CHECKING:
    from typing import Any

    from sqlalchemy.ext.asyncio import AsyncSession


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: dict[str, Any], db_session: AsyncSession) -> list[dict[str, Any]]:
    todo_item = TodoItem(**data)
    async with db_session.begin():
        db_session.add(todo_item)
    return [
        {
            "title": item.title,
            "done": item.done,
        }
        for item in (await db_session.execute(select(TodoItem))).scalars()
    ]


async def init_db(app: Litestar) -> None:
    async with config.get_engine().begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


config = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///todo_async.sqlite")
plugin = SQLAlchemyInitPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin], on_startup=[init_db], debug=True)

Sync

SQLAlchemy Sync Init Plugin Example

from __future__ import annotations

from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from litestar import Litestar, post
from litestar.plugins.sqlalchemy import SQLAlchemyInitPlugin, SQLAlchemySyncConfig

if TYPE_CHECKING:
    from typing import Any

    from sqlalchemy.orm import Session


class Base(DeclarativeBase): ...


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


@post("/", sync_to_thread=True)
def add_item(data: dict[str, Any], db_session: Session) -> list[dict[str, Any]]:
    todo_item = TodoItem(**data)
    with db_session.begin():
        db_session.add(todo_item)
    return [
        {
            "title": item.title,
            "done": item.done,
        }
        for item in db_session.execute(select(TodoItem)).scalars()
    ]


def init_db(app: Litestar) -> None:
    with config.get_engine().begin() as conn:
        Base.metadata.create_all(conn)


config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite")
plugin = SQLAlchemyInitPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin], on_startup=[init_db])