Sanic Integration¶

sqlspec.extensions.sanic provides SQLSpecPlugin for Sanic applications, registering before_server_start and after_server_stop pool listeners, managing per-request sessions on request.ctx, integrating with sanic-ext constructor injection, and supporting status-aware transaction commit modes and observability middleware.

Installation¶

uv add "sqlspec[sanic]"

Basic Setup¶

Configure extension_config["sanic"] (SanicConfig), attach the plugin via db_plugin.init_app(app), and retrieve the session with db_plugin.get_session(request, "db"):

docs/examples/frameworks/sanic/basic_setup.py

"""Register SQLSpecPlugin with a Sanic application and query an async SQLite session."""

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

__all__ = ("main",)


def main() -> Sanic:
    """Create a Sanic application wired with SQLSpecPlugin."""
    spec = SQLSpec()
    spec.add_config(
        AiosqliteConfig(
            connection_config={"database": ":memory:"},
            extension_config={"sanic": {"commit_mode": "manual", "session_key": "db"}},
        )
    )

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

    @app.get("/health")
    async def health(request: Request) -> HTTPResponse:
        db: AiosqliteDriver = db_plugin.get_session(request, "db")
        return response.json(await db.select_one("SELECT 1 AS ok"))

    db_plugin.init_app(app)
    return app


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

Commit Modes and Custom Rollback Statuses¶

Set commit_mode ("manual", "autocommit", or "autocommit_include_redirect") and optional extra_commit_statuses / extra_rollback_statuses sets in extension_config["sanic"]:

docs/examples/frameworks/sanic/commit_modes.py

"""Configure Sanic autocommit mode with custom HTTP status codes triggering rollback."""

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

__all__ = ("main",)


def main() -> Sanic:
    """Create a Sanic application rolling back automatically on HTTP 409 responses."""
    spec = SQLSpec()
    spec.add_config(
        AiosqliteConfig(
            connection_config={"database": ":memory:"},
            extension_config={
                "sanic": {"commit_mode": "autocommit", "extra_rollback_statuses": {409}, "session_key": "db"}
            },
        )
    )

    db_plugin = SQLSpecPlugin(spec)
    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)
    return app


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

Multi-Database Routing¶

Assign distinct bind_key, session_key, connection_key, and pool_key values when connecting a Sanic service to multiple databases:

docs/examples/frameworks/sanic/multi_database.py

"""Query multiple async and sync SQLSpec database binds inside a Sanic route."""

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

__all__ = ("main",)


def main() -> Sanic:
    """Create a Sanic application wired to primary async and analytics sync databases."""
    spec = SQLSpec()
    spec.add_config(
        AiosqliteConfig(
            bind_key="primary",
            connection_config={"database": ":memory:"},
            extension_config={
                "sanic": {
                    "connection_key": "primary_connection",
                    "pool_key": "primary_pool",
                    "session_key": "primary_db",
                }
            },
        )
    )
    spec.add_config(
        SqliteConfig(
            bind_key="analytics",
            connection_config={"database": ":memory:"},
            extension_config={
                "sanic": {
                    "connection_key": "analytics_connection",
                    "pool_key": "analytics_pool",
                    "session_key": "analytics_db",
                }
            },
        )
    )

    db_plugin = SQLSpecPlugin(spec)
    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)
    return app


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

Pool-Only Mode (`disable_di=True`) and Observability Middleware¶

For high-throughput endpoints that only acquire connections on demand rather than opening a request-scoped session on every request, set disable_di: True and use request.app.ctx.<pool_key> directly:

Pool-Only Mode (disable_di)Correlation & SQLCommenter Middleware

docs/examples/frameworks/sanic/disable_di.py

"""Manage Sanic connection pools directly on app.ctx with per-request DI disabled."""

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

__all__ = ("main",)


def main() -> Sanic:
    """Create a Sanic application acquiring connections manually from app.ctx.db_pool."""
    spec = SQLSpec()
    config = AiosqliteConfig(
        connection_config={"database": ":memory:"},
        extension_config={"sanic": {"disable_di": True, "pool_key": "db_pool"}},
    )
    spec.add_config(config)

    db_plugin = SQLSpecPlugin(spec)
    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)
            return response.json(await db.select_one("SELECT 1 AS ok"))

    db_plugin.init_app(app)
    return app


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

docs/examples/frameworks/sanic/observability.py

"""Enable request correlation IDs and SQLCommenter trace context middleware in Sanic."""

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

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

__all__ = ("main",)


def main() -> Sanic:
    """Create a Sanic application with correlation and SQLCommenter middleware enabled."""
    spec = SQLSpec()
    spec.add_config(
        AiosqliteConfig(
            connection_config={"database": ":memory:"},
            statement_config=StatementConfig(enable_sqlcommenter=True),
            extension_config={
                "sanic": {
                    "enable_correlation_middleware": True,
                    "enable_sqlcommenter_middleware": True,
                    "session_key": "db",
                }
            },
        )
    )

    db_plugin = SQLSpecPlugin(spec)
    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)
    return app


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

Recap¶

Configure session_key, commit_mode, disable_di, and observability flags in extension_config["sanic"].
Retrieve sessions via db_plugin.get_session(request, key) or inject drivers directly when sanic-ext is installed.
See the Sanic Extension Reference for SQLSpecPlugin and SanicConfig signatures.