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)