Sanic¶

SQLSpec provides a Sanic extension that uses Sanic-native application and request context. Connection pools are stored on app.ctx during the worker lifecycle, and request-scoped connections and sessions are stored on request.ctx.

Installation¶

Install SQLSpec with the Sanic extra:

uv add "sqlspec[sanic]"

pip

pip install "sqlspec[sanic]"

Poetry

poetry add "sqlspec[sanic]"

PDM

pdm add "sqlspec[sanic]"

Basic Setup¶

Create a SQLSpec instance, register your database config, and attach SQLSpecPlugin to your Sanic app. Configure Sanic-specific options under extension_config["sanic"].

sanic basic setup¶

from sanic import Request, Sanic, response
from sanic.response import HTTPResponse

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

sqlspec = SQLSpec()
sqlspec.add_config(
    AiosqliteConfig(
        connection_config={"database": ":memory:"},
        extension_config={"sanic": {"commit_mode": "manual", "session_key": "db"}},
    )
)

db_plugin = SQLSpecPlugin(sqlspec)
app = Sanic("SQLSpecSanicBasicExample")

@app.get("/health")
async def health(request: Request) -> HTTPResponse:
    db: AiosqliteDriver = db_plugin.get_session(request, "db")
    result = await db.execute("select 1 as ok")
    return response.json(result.one())

db_plugin.init_app(app)

Transaction Modes¶

Sanic supports the same transaction modes as the other request-oriented framework integrations:

manual: SQLSpec manages connection scope only. Your handler commits or rolls back.
autocommit: SQLSpec commits 2xx responses and rolls back other responses.
autocommit_include_redirect: SQLSpec commits 2xx and 3xx responses and rolls back other responses.

Extra commit and rollback statuses are configured per database config.

sanic commit modes¶

from sanic import Request, Sanic, response
from sanic.response import HTTPResponse

from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.extensions.sanic import SQLSpecPlugin

sqlspec = SQLSpec()
sqlspec.add_config(
    AiosqliteConfig(
        connection_config={"database": "app.db"},
        extension_config={
            "sanic": {"commit_mode": "autocommit", "extra_rollback_statuses": {409}, "session_key": "db"}
        },
    )
)

db_plugin = SQLSpecPlugin(sqlspec)
app = Sanic("SQLSpecSanicCommitModesExample")

@app.post("/users")
async def create_user(request: Request) -> HTTPResponse:
    db = db_plugin.get_session(request, "db")
    await db.execute("insert into users (name) values (:name)", {"name": request.json["name"]})
    return response.json({"created": True}, status=201)

db_plugin.init_app(app)

Multiple Databases¶

For multiple databases, give each config unique connection_key, pool_key, and session_key values. The plugin can then look up sessions by session key.

sanic multi database¶

from sanic import Request, Sanic, response
from sanic.response import HTTPResponse

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

sqlspec = SQLSpec()
sqlspec.add_config(
    AiosqliteConfig(
        bind_key="primary",
        connection_config={"database": "primary.db"},
        extension_config={
            "sanic": {
                "connection_key": "primary_connection",
                "pool_key": "primary_pool",
                "session_key": "primary_db",
            }
        },
    )
)
sqlspec.add_config(
    SqliteConfig(
        bind_key="analytics",
        connection_config={"database": "analytics.db"},
        extension_config={
            "sanic": {
                "connection_key": "analytics_connection",
                "pool_key": "analytics_pool",
                "session_key": "analytics_db",
            }
        },
    )
)

db_plugin = SQLSpecPlugin(sqlspec)
app = Sanic("SQLSpecSanicMultiDatabaseExample")

@app.get("/report")
async def report(request: Request) -> HTTPResponse:
    primary: AiosqliteDriver = db_plugin.get_session(request, "primary_db")
    analytics: SqliteDriver = db_plugin.get_session(request, "analytics_db")
    users = await primary.select("select 1 as id, 'Alice' as name")
    metrics = analytics.select("select 'active_users' as name, 100 as value")
    return response.json({"users": users, "metrics": metrics})

db_plugin.init_app(app)

Request Access¶

Use plugin.get_session(request, key=None) for driver sessions and plugin.get_connection(request, key=None) for raw connection access. The default key is the first registered config's session_key; passing a key is recommended when an app has multiple configs.

get_connection_from_request(request, config_state) and get_or_create_session(request, config_state) are lower-level helpers for custom integrations.

disable_di¶

Set disable_di=True when another dependency injection system owns request-scoped connection management. SQLSpec still creates and closes pools on Sanic startup and shutdown, but it does not put connections or sessions on request.ctx.

sanic disable di¶

from sanic import Request, Sanic, response
from sanic.response import HTTPResponse

from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.extensions.sanic import SQLSpecPlugin

sqlspec = SQLSpec()
config = AiosqliteConfig(
    connection_config={"database": "app.db"},
    extension_config={"sanic": {"disable_di": True, "pool_key": "db_pool"}},
)
sqlspec.add_config(config)

db_plugin = SQLSpecPlugin(sqlspec)
app = Sanic("SQLSpecSanicDisableDIExample")

@app.get("/health")
async def health(request: Request) -> HTTPResponse:
    async with config.provide_connection(request.app.ctx.db_pool) as connection:
        db = config.driver_type(connection=connection, statement_config=config.statement_config)
        result = await db.execute("select 1 as ok")
        return response.json(result.one())

db_plugin.init_app(app)

Correlation IDs and SQLCommenter¶

Set enable_correlation_middleware=True to extract request correlation IDs from headers and return X-Correlation-ID on the response. The active value is also available from request.ctx.correlation_id and CorrelationContext during the request.

SQLCommenter is enabled when the driver StatementConfig has enable_sqlcommenter=True and the Sanic config keeps enable_sqlcommenter_middleware=True. Request attributes include framework="sanic", route, and action.

sanic observability¶

from sanic import Request, Sanic, response
from sanic.response import HTTPResponse

from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig
from sqlspec.core import StatementConfig
from sqlspec.extensions.sanic import SQLSpecPlugin

sqlspec = SQLSpec()
sqlspec.add_config(
    AiosqliteConfig(
        connection_config={"database": "app.db"},
        statement_config=StatementConfig(enable_sqlcommenter=True),
        extension_config={
            "sanic": {
                "enable_correlation_middleware": True,
                "enable_sqlcommenter_middleware": True,
                "session_key": "db",
            }
        },
    )
)

db_plugin = SQLSpecPlugin(sqlspec)
app = Sanic("SQLSpecSanicObservabilityExample")

@app.get("/health")
async def health(request: Request) -> HTTPResponse:
    return response.json({"correlation_id": request.ctx.correlation_id})

db_plugin.init_app(app)