Litestar Integration¶

sqlspec.extensions.litestar provides SQLSpecPlugin, wiring SQLSpec connection pools into Litestar's application lifespan, registering request-scoped session and connection dependencies, mounting litestar database CLI migration commands, providing declarative query filter dependencies, and supplying server-side Litestar Store implementations across 13 adapters.

Installation¶

uv add "sqlspec[litestar]"

Plugin Setup and Session Injection¶

Register one or more database configurations on a SQLSpec instance, configure extension_config["litestar"], and pass SQLSpecPlugin(sqlspec=spec) to Litestar(plugins=[...]):

docs/examples/extensions/litestar/plugin_setup.py

"""Register SQLSpecPlugin with a Litestar application and inject a named database session."""

from typing import Any

from litestar import Litestar, get
from litestar.di import NamedDependency

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig, SqliteDriver
from sqlspec.extensions.litestar import SQLSpecPlugin

__all__ = ("health_check", "main")


@get("/health", sync_to_thread=False)
def health_check(db_session: NamedDependency[SqliteDriver]) -> dict[str, Any]:
    """Execute a health-check query against the injected SQLite session."""
    return db_session.select_one("SELECT 'ok' AS status")


def main() -> Litestar:
    """Create a Litestar application wired with SQLSpecPlugin."""
    spec = SQLSpec()
    spec.add_config(
        SqliteConfig(
            connection_config={"database": ":memory:"}, extension_config={"litestar": {"session_key": "db_session"}}
        )
    )
    return Litestar(route_handlers=[health_check], plugins=[SQLSpecPlugin(sqlspec=spec)])


if __name__ == "__main__":
    print(main())

By default, a single configuration registers db_session, db_connection, and db_pool in Litestar's dependency graph. Override session_key, connection_key, or pool_key inside extension_config["litestar"] to customize parameter names.

Multi-Database Dependency Keys¶

When an application connects to multiple databases (for example, a transactional primary database and a read-heavy analytics database), assign a distinct bind_key and session_key to each configuration:

docs/examples/extensions/litestar/dependency_keys.py

"""Register distinct Litestar dependency injection keys for multiple database configurations."""

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.extensions.litestar import SQLSpecPlugin

__all__ = ("main",)


def main() -> SQLSpecPlugin:
    """Configure primary and analytics SQLite binds with explicit Litestar session_key values."""
    spec = SQLSpec()
    spec.add_config(
        SqliteConfig(
            bind_key="analytics",
            connection_config={"database": ":memory:"},
            extension_config={"litestar": {"session_key": "analytics"}},
        )
    )
    spec.add_config(
        SqliteConfig(
            bind_key="primary",
            connection_config={"database": ":memory:"},
            extension_config={"litestar": {"session_key": "primary"}},
        )
    )
    plugin = SQLSpecPlugin(sqlspec=spec)
    assert plugin.get_config("analytics").bind_key == "analytics"
    assert plugin.get_config("primary").bind_key == "primary"
    return plugin


if __name__ == "__main__":
    print(main())

Transaction Commit Modes¶

Configure commit_mode inside extension_config["litestar"] to control how request-scoped sessions finalize when a route handler returns:

Commit Mode	Commits On	Rolls Back On	Best For
`"manual"` (default)	Explicit `session.commit()` calls only	Unhandled exceptions or uncommitted changes at request exit	Explicit service-layer transaction boundaries
`"autocommit"`	HTTP `200`–`299` responses	HTTP `>= 300` responses (`3xx`, `4xx`, `5xx`) or unhandled exceptions	JSON REST APIs where every `2xx` handler mutates state
`"autocommit_include_redirect"`	HTTP `200`–`399` responses	HTTP `>= 400` responses or unhandled exceptions	HTML / HTMX form handlers that commit and return `302` / `303` redirects

docs/examples/frameworks/litestar/commit_modes.py

"""Configure autocommit and manual transaction commit modes on Litestar database binds."""

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.extensions.litestar import SQLSpecPlugin

__all__ = ("main",)


def main() -> SQLSpecPlugin:
    """Create a Litestar SQLSpecPlugin with autocommit and manual binds."""
    spec = SQLSpec()
    spec.add_config(
        SqliteConfig(
            connection_config={"database": ":memory:"}, extension_config={"litestar": {"commit_mode": "autocommit"}}
        )
    )
    spec.add_config(
        SqliteConfig(
            bind_key="manual",
            connection_config={"database": ":memory:"},
            extension_config={"litestar": {"commit_mode": "manual"}},
        )
    )
    return SQLSpecPlugin(sqlspec=spec)


if __name__ == "__main__":
    print(main())

Add extra_commit_statuses or extra_rollback_statuses in LitestarConfig when specific status codes require custom commit or rollback treatment.

Declarative Query Filter Providers¶

Use create_filter_dependencies with FilterConfig to generate OpenAPI-documented Litestar dependencies that parse query parameters directly into StatementFilter lists (LimitOffsetFilter, OrderByFilter, SearchFilter, BeforeAfterFilter, CursorFilter, and InCollectionFilter):

from datetime import datetime
from typing import Annotated, Any
from uuid import UUID

from litestar import Litestar, get
from litestar.params import Dependency

from sqlspec.adapters.asyncpg import AsyncpgDriver
from sqlspec.core import FilterTypes
from sqlspec.extensions.litestar.providers import FilterConfig, create_filter_dependencies

user_filters = create_filter_dependencies(
    FilterConfig(
        id_filter=UUID,
        id_field="id",
        sort_field="created_at",
        sort_order="desc",
        pagination_type="limit_offset",
        pagination_size=25,
        pagination_max_size=100,
        search="name,email",
        search_ignore_case=True,
        created_at=True,
        updated_at=True,
    )
)


@get("/users", dependencies=user_filters)
async def list_users(
    db_session: AsyncpgDriver, filters: Annotated[list[FilterTypes], Dependency(skip_validation=True)]
) -> list[dict[str, Any]]:
    return await db_session.select("SELECT id, name, email, created_at FROM users", *filters)

For bidirectional keyset pagination (CursorPagination), set cursor_key="id" (or pass CursorKey specs) on FilterConfig; see Filtering and Pagination.

Server-Side Session Stores¶

Thirteen adapters expose a Litestar Store implementation under sqlspec.adapters.<adapter>.litestar (AsyncpgStore, PsycopgAsyncStore, PsqlpyStore, AiosqliteStore, SqliteStore, AsyncmyStore, AiomysqlStore, PymysqlStore, MysqlConnectorAsyncStore, OracleAsyncStore, SpannerSyncStore, SpannerAsyncStore, DuckDBStore, AdbcStore, ArrowOdbcStore, and Db2SyncStore):

docs/examples/frameworks/litestar/session_stores.py

"""Construct an asynchronous SQLSpec server-side session store for Litestar."""

from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.adapters.aiosqlite.litestar import AiosqliteStore

__all__ = ("main",)


def main() -> AiosqliteStore:
    """Create an AiosqliteStore backed by an in-memory SQLite database."""
    config = AiosqliteConfig(connection_config={"database": ":memory:"})
    return AiosqliteStore(config)


if __name__ == "__main__":
    print(main())

Call await store.create_table() during application startup (or include the extension migration via enable_extensions=["litestar"] in migration_config) and pass the store to Litestar's ServerSideSessionConfig or stores={"sessions": store} registry.

Recap¶

SQLSpecPlugin manages pool lifecycles, injects db_session / db_connection / db_pool, and registers litestar database CLI commands automatically.
Set commit_mode, session_key, enable_correlation_middleware, and enable_sqlcommenter_middleware in extension_config["litestar"].
See the Litestar Extension Reference for SQLSpecPlugin, LitestarConfig, and FilterConfig signatures.