FastAPI Integration¶

sqlspec.extensions.fastapi provides SQLSpecPlugin for FastAPI (>= 0.162.0), wiring database connection pools into FastAPI's lifespan context manager and exposing Depends(...) callables for request-scoped driver sessions, connections, and declarative query filters.

Installation¶

uv add "sqlspec[fastapi]"

Basic Setup and Session Injection¶

Attach SQLSpecPlugin(spec, app) to your FastAPI instance and inject the session into route handlers using Depends(db_ext.provide_session()):

docs/examples/frameworks/fastapi/basic_setup.py

"""Register SQLSpecPlugin with FastAPI and inject an async database session."""

from typing import Annotated, Any

from fastapi import Depends, FastAPI

from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig, AiosqliteDriver
from sqlspec.extensions.fastapi import SQLSpecPlugin

__all__ = ("main",)


def main() -> FastAPI:
    """Create a FastAPI application with an injected AiosqliteDriver route."""
    spec = SQLSpec()
    spec.add_config(AiosqliteConfig(connection_config={"database": ":memory:"}))

    app = FastAPI()
    db_ext = SQLSpecPlugin(spec, app)

    @app.get("/teams")
    async def list_teams(db: Annotated[AiosqliteDriver, Depends(db_ext.provide_session())]) -> dict[str, Any]:
        return await db.select_one("SELECT 1 AS ok")

    return app


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

SQLSpecPlugin automatically opens configured connection pools when FastAPI starts and closes them when the application shuts down.

Multi-Database Configuration¶

Register multiple synchronous or asynchronous adapters on SQLSpec with distinct session_key names under extension_config["fastapi"], then pass the key to db_plugin.provide_session("<key>"):

docs/examples/frameworks/fastapi/multi_database.py

"""Inject multiple async and sync SQLSpec database sessions into a FastAPI route."""

from typing import Annotated, Any

from fastapi import Depends, FastAPI

from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig, AiosqliteDriver
from sqlspec.adapters.sqlite import SqliteConfig, SqliteDriver
from sqlspec.extensions.fastapi import SQLSpecPlugin

__all__ = ("main",)


def main() -> FastAPI:
    """Create a FastAPI application wired to both primary async and ETL sync databases."""
    spec = SQLSpec()
    spec.add_config(
        AiosqliteConfig(
            connection_config={"database": ":memory:"},
            extension_config={
                "fastapi": {"session_key": "db", "connection_key": "db_connection", "pool_key": "db_pool"}
            },
        )
    )
    spec.add_config(
        SqliteConfig(
            connection_config={"database": ":memory:"},
            extension_config={
                "fastapi": {"session_key": "etl_db", "connection_key": "etl_connection", "pool_key": "etl_pool"}
            },
        )
    )

    app = FastAPI()
    db_plugin = SQLSpecPlugin(spec, app)

    @app.get("/report")
    async def report(
        db: Annotated[AiosqliteDriver, Depends(db_plugin.provide_session("db"))],
        etl_db: Annotated[SqliteDriver, Depends(db_plugin.provide_session("etl_db"))],
    ) -> dict[str, list[dict[str, Any]]]:
        users = await db.select("SELECT 1 AS id, 'Alice' AS name")
        metrics = etl_db.select("SELECT 'metric1' AS name, 100 AS value")
        return {"users": users, "metrics": metrics}

    return app


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

Transaction Commit Modes and Options¶

Configure extension_config["fastapi"] (FastAPIConfig, which shares StarletteConfig options) per database bind:

Key	Type	Default	Description
`session_key`	`str`	`"db_session"`	Identifier passed to `provide_session(key)` and stored on `request.state`.
`connection_key`	`str`	`"db_connection"`	Identifier used for raw connection dependencies.
`pool_key`	`str`	`"db_pool"`	Attribute key on `app.state` storing the adapter connection pool.
`commit_mode`	`"manual" \| "autocommit" \| "autocommit_include_redirect"`	`"manual"`	Controls automatic transaction commit or rollback based on response status codes.
`extra_commit_statuses`	`set[int]`	`set()`	Additional HTTP status codes that trigger an automatic commit.
`extra_rollback_statuses`	`set[int]`	`set()`	Additional HTTP status codes that trigger an automatic rollback.
`enable_correlation_middleware`	`bool`	`False`	Extracts or generates `X-Request-ID` correlation IDs into `CorrelationContext`.
`enable_sqlcommenter_middleware`	`bool`	`False`	Populates `SQLCommenterContext` with FastAPI route, method, and trace attributes.

Declarative Filter Dependencies¶

FastAPI shares the same FilterConfig builder as Litestar via sqlspec.extensions.fastapi.providers.create_filter_dependencies, generating typed Depends callables for LimitOffsetFilter, CursorFilter (returning HTTP 422 on malformed cursor tokens), OrderByFilter, SearchFilter, and BeforeAfterFilter.

Recap¶

Instantiate db_ext = SQLSpecPlugin(spec, app) to bind pool lifecycles and request sessions to FastAPI.
Use Depends(db_ext.provide_session()) for single-database routes or Depends(db_ext.provide_session("key")) for multi-database setups.
See the FastAPI Extension Reference for SQLSpecPlugin and provider signatures.