FastAPI#

SQLSpec provides a FastAPI extension that wires database sessions into the request lifecycle using standard dependency injection. The extension manages connection pools and ensures proper cleanup between requests.

Installation#

Install SQLSpec with the FastAPI extra:

uv add "sqlspec[fastapi]"

pip install "sqlspec[fastapi]"

poetry add "sqlspec[fastapi]"

pdm add "sqlspec[fastapi]"

Basic Setup#

Create a SQLSpec instance, register your database config, and attach the plugin to your FastAPI app. The plugin provides provide_session dependencies that yield a session for each request. Configure FastAPI-specific options under extension_config["fastapi"].

fastapi basic setup#
from fastapi import Depends, FastAPI

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

sqlspec = SQLSpec()
sqlspec.add_config(AiosqliteConfig(connection_config={"database": ":memory:"}))

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

@app.get("/teams")
async def list_teams(db: Annotated[AiosqliteDriver, Depends(db_ext.provide_session())]) -> dict[str, Any]:
    result = await db.execute("select 1 as ok")
    return result.one()

Transaction Modes#

FastAPI uses Starlette-compatible transaction middleware configured via extension_config["fastapi"]:

manual (default)

SQLSpec manages the connection lifecycle and closes connections at the end of the request, leaving transaction commit and rollback decisions to your handler logic.

autocommit

SQLSpec automatically commits transactions for 2xx responses and rolls back on 4xx/5xx responses or unhandled exceptions.

autocommit_include_redirect

Extends autocommit to also commit on redirect responses (2xx and 3xx).

from sqlspec.adapters.aiosqlite import AiosqliteConfig

config = AiosqliteConfig(
    connection_config={"database": "app.db"},
    extension_config={
        "fastapi": {
            "commit_mode": "autocommit",
            "extra_rollback_statuses": {409},
            "session_key": "db",
        }
    },
)

Multiple Databases#

Configure each database with unique session_key, connection_key, and pool_key values under extension_config["fastapi"]. Inject distinct sessions by passing the key to provide_session():

fastapi multi database#
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

sqlspec = SQLSpec()

# Primary async database
sqlspec.add_config(
    AiosqliteConfig(
        connection_config={"database": ":memory:"},
        extension_config={
            "fastapi": {"session_key": "db", "connection_key": "db_connection", "pool_key": "db_pool"}
        },
    )
)

# ETL sync database (e.g., DuckDB pattern)
sqlspec.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(sqlspec, 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]:
    # Async query to primary database
    users = await db.select("SELECT 1 as id, 'Alice' as name")
    # Sync query to ETL database
    metrics = etl_db.select("SELECT 'metric1' as name, 100 as value")
    return {"users": users, "metrics": metrics}

Dependency Injection Helpers#

The SQLSpecPlugin provides several dependency factories for FastAPI routes:

  • db_ext.provide_session(key=None): Injects a driver session (sync or async depending on config).

  • db_ext.provide_async_session(key=None): Type-narrowed dependency returning AsyncDriverAdapterBase.

  • db_ext.provide_sync_session(key=None): Type-narrowed dependency returning SyncDriverAdapterBase.

  • db_ext.provide_connection(key=None): Injects the raw database connection.

Filter Dependencies#

SQLSpec provides provide_filters to parse HTTP query parameters into typed StatementFilter objects (pagination, sorting, search, before/after):

from typing import Annotated
from fastapi import Depends, FastAPI
from sqlspec.core import FilterTypes
from sqlspec.extensions.fastapi import provide_filters

app = FastAPI()

filter_dep = provide_filters(
    {
        "pagination_type": "limit_offset",
        "pagination_size": 20,
        "sort_field": ["name", "created_at"],
        "search": "name,email",
    }
)

@app.get("/items")
async def list_items(
    filters: Annotated[list[FilterTypes], Depends(filter_dep)],
) -> dict[str, str]:
    return {"status": "ok"}

To page with cursors, pass keys and a signing secret to the provider:

import os

cursor_filter_dep = provide_filters({
    "pagination_type": "cursor",
    "cursor_keys": [("created_at", "desc"), ("id", "desc")],
    "cursor_secret": os.environ["CURSOR_SECRET"],
    "pagination_size": 20,
})

Inject cursor_filter_dep through Depends in place of filter_dep and pass *filters to SQLSpecAsyncService(db_session).paginate() (from sqlspec.service). Clients use cursor and pageSize to move through pages. Invalid cursor tokens yield an HTTP 422 validation response at the cursor query parameter. See Cursor Pagination for key selection, NULL values, and signed tokens.

Observability and SQLCommenter#

The FastAPI extension includes middleware for request correlation and SQL query commenting:

from sqlspec.adapters.aiosqlite import AiosqliteConfig

config = AiosqliteConfig(
    connection_config={"database": "app.db"},
    extension_config={
        "fastapi": {
            "enable_correlation_middleware": True,
            "correlation_header": "x-request-id",
            "enable_sqlcommenter_middleware": True,
            "sqlcommenter_framework": "fastapi",
        }
    },
)

  • Correlation Middleware: Extracts or generates a correlation ID from incoming request headers, attaches it to the response as X-Correlation-ID, and stores it in request.state.correlation_id and CorrelationContext.

  • SQLCommenter Middleware: Injects route and action information (e.g. /*route='/items',action='list_items',framework='fastapi'*/) into SQL query comments for downstream query telemetry.