Starlette#

SQLSpec provides a Starlette extension that hooks database sessions into the ASGI lifecycle. The extension uses Starlette's lifespan context to manage connection pools and provides middleware for request-scoped sessions and transactions.

Installation#

Install SQLSpec with the Starlette extra:

uv add "sqlspec[starlette]"

pip

pip install "sqlspec[starlette]"

Poetry

poetry add "sqlspec[starlette]"

PDM

pdm add "sqlspec[starlette]"

Basic Setup#

Create a SQLSpec instance, register your database config, and attach SQLSpecPlugin to your Starlette app. The plugin adds lifespan handlers to manage pools and middleware for request sessions.

starlette basic setup#

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

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

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

# Create plugin at module level
db_plugin = SQLSpecPlugin(sqlspec)

async def health(request: Request) -> JSONResponse:
    db: AiosqliteDriver = db_plugin.get_session(request)
    result = await db.execute("select 1 as ok")
    return JSONResponse(result.one())

app = Starlette(routes=[Route("/health", health)])
db_plugin.init_app(app)  # Initialize plugin with app

Transaction Modes#

Configure the commit mode under extension_config["starlette"]:

manual (default): SQLSpec manages the connection lifecycle and closes connections before completing the request, leaving commits and rollbacks to your route handlers.
autocommit: SQLSpec commits transactions for successful 2xx responses and rolls back on 4xx/5xx responses or unhandled exceptions.
autocommit_include_redirect: Extends autocommit to commit transactions on redirect responses (2xx and 3xx).

from sqlspec.adapters.aiosqlite import AiosqliteConfig

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

Multiple Databases#

For multiple databases, configure each config with unique session_key, connection_key, and pool_key settings. Retrieve sessions by key:

starlette multi database#

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

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

sqlspec = SQLSpec()

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

# ETL sync database
sqlspec.add_config(
    SqliteConfig(
        connection_config={"database": ":memory:"},
        extension_config={
            "starlette": {"session_key": "etl_db", "connection_key": "etl_connection", "pool_key": "etl_pool"}
        },
    )
)

db_plugin = SQLSpecPlugin(sqlspec)

async def report(request: Request) -> JSONResponse:
    db: AiosqliteDriver = db_plugin.get_session(request, "db")
    etl_db: SqliteDriver = db_plugin.get_session(request, "etl_db")

    # 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 JSONResponse({"users": users, "metrics": metrics})

app = Starlette(routes=[Route("/report", report)])
db_plugin.init_app(app)

Request and App Access#

The plugin provides methods to access sessions, connections, and pools:

db_plugin.get_session(request, key=None): Returns the request-scoped driver session.
db_plugin.get_connection(request, key=None): Returns the underlying database connection.
db_plugin.get_pool(app, key=None): Returns the application connection pool from app.state.

disable_di#

Set disable_di=True when an external dependency injection framework (such as Dishka) manages request-scoped connections. SQLSpec still initializes and cleans up pools via Starlette lifespan, but omits request-scoped session middleware.

Observability and SQLCommenter#

The Starlette extension supports correlation tracking and query commenting:

from sqlspec.adapters.aiosqlite import AiosqliteConfig

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

Correlation Middleware: Captures or assigns correlation IDs, injecting X-Correlation-ID into response headers and synchronizing with request.state.correlation_id and CorrelationContext.
SQLCommenter Middleware: Attaches request route and action context to outgoing SQL queries.