Flask#

SQLSpec provides a Flask extension that manages database connections within the Flask request lifecycle. The extension registers connection pool setup and teardown with Flask's application context hooks, supporting both synchronous and asynchronous adapters.

Installation#

Install SQLSpec with the Flask extra:

uv add "sqlspec[flask]"

pip

pip install "sqlspec[flask]"

Poetry

poetry add "sqlspec[flask]"

PDM

pdm add "sqlspec[flask]"

Basic Setup#

Create a SQLSpec instance, register your database config, and attach the plugin to your Flask app. Use plugin.get_session() inside request handlers to obtain a session.

flask basic setup#

from flask import Flask

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig, SqliteDriver
from sqlspec.extensions.flask import SQLSpecPlugin

# Create SQLSpec and plugin at module level
sqlspec = SQLSpec()
sqlspec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
plugin = SQLSpecPlugin(sqlspec)

def create_app() -> Flask:
    """Application factory pattern."""
    app = Flask(__name__)
    plugin.init_app(app)

    @app.get("/health")
    def health() -> dict[str, int]:
        db: SqliteDriver = plugin.get_session()
        result = db.execute("select 1 as ok")
        return result.one()

    return app

app = create_app()

Transaction Modes#

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

manual (default): SQLSpec manages connection lifecycle within the request context. Your route handler commits or rolls back explicitly.
autocommit: SQLSpec automatically commits transactions for 2xx responses and rolls back for 4xx/5xx responses or unhandled exceptions.
autocommit_include_redirect: Extends autocommit to also commit on redirect responses (2xx and 3xx).

from sqlspec.adapters.sqlite import SqliteConfig

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

Multiple Databases#

For multiple databases, assign unique session_key, connection_key, and pool_key settings under extension_config["flask"]. Retrieve sessions by passing the key to plugin.get_session():

flask multi database#

from flask import Flask

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig, SqliteDriver
from sqlspec.extensions.flask import SQLSpecPlugin

sqlspec = SQLSpec()

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

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

plugin = SQLSpecPlugin(sqlspec)

def create_app() -> Flask:
    app = Flask(__name__)
    plugin.init_app(app)

    @app.get("/report")
    def report() -> dict[str, list]:
        db: SqliteDriver = plugin.get_session("db")
        etl_db: SqliteDriver = plugin.get_session("etl_db")

        users = db.select("SELECT 1 as id, 'Alice' as name")
        metrics = etl_db.select("SELECT 'metric1' as name, 100 as value")
        return {"users": users, "metrics": metrics}

    return app

app = create_app()

Async Adapter Support#

While Flask operates synchronously by default, SQLSpec provides seamless support for asynchronous database adapters (such as AsyncpgConfig or AiosqliteConfig) via an internal AnyIO portal runner. In async configurations, get_session() handles the event loop bridge transparently.

Observability and SQLCommenter#

The Flask extension integrates request correlation and query tagging:

from sqlspec.adapters.sqlite import SqliteConfig

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

Correlation Tracking: Extracts x-request-id on before_request, populates Flask's request context and CorrelationContext, and writes X-Correlation-ID on after_request.
SQLCommenter: Automatically attaches Flask endpoint and blueprint details as SQL comments.