Skip to content

Flask Integration

sqlspec.extensions.flask provides SQLSpecPlugin for Flask applications, managing synchronous (and portal-bridged asynchronous) connection pools across Flask's application factory lifecycle (init_app), caching request-scoped sessions on flask.g, and finalizing transactions in teardown_appcontext.

Installation

uv add "sqlspec[flask]"

Application Factory Setup

Create a SQLSpecPlugin(spec) instance, initialize it inside your create_app factory with plugin.init_app(app), and retrieve the active request session via plugin.get_session():

docs/examples/frameworks/flask/basic_setup.py
"""Configure SQLSpecPlugin with a Flask application factory."""

from typing import Any

from flask import Flask

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

__all__ = ("create_app", "main")


def create_app(plugin: SQLSpecPlugin) -> Flask:
    """Create a Flask application initialized with SQLSpecPlugin."""
    app = Flask(__name__)
    plugin.init_app(app)

    @app.get("/health")
    def health() -> dict[str, Any]:
        db: SqliteDriver = plugin.get_session()
        return db.select_one("SELECT 1 AS ok")

    return app


def main() -> Flask:
    """Build the SQLSpec plugin and return the configured Flask application."""
    spec = SQLSpec()
    spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
    plugin = SQLSpecPlugin(spec)
    return create_app(plugin)


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

Calling plugin.get_session() multiple times within the same HTTP request returns the same flask.g-cached driver session.

Multi-Database Binds

Configure distinct session_key, connection_key, and pool_key values under extension_config["flask"] (FlaskConfig) and pass the key to plugin.get_session("<key>"):

docs/examples/frameworks/flask/multi_database.py
"""Access multiple named SQLSpec database sessions inside a Flask route."""

from typing import Any

from flask import Flask

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

__all__ = ("create_app", "main")


def create_app(plugin: SQLSpecPlugin) -> Flask:
    """Create a Flask application exposing a multi-database report endpoint."""
    app = Flask(__name__)
    plugin.init_app(app)

    @app.get("/report")
    def report() -> dict[str, list[dict[str, Any]]]:
        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


def main() -> Flask:
    """Configure primary and ETL SQLite binds for Flask and return the app."""
    spec = SQLSpec()
    spec.add_config(
        SqliteConfig(
            connection_config={"database": ":memory:"},
            extension_config={"flask": {"session_key": "db", "connection_key": "db_connection", "pool_key": "db_pool"}},
        )
    )
    spec.add_config(
        SqliteConfig(
            connection_config={"database": ":memory:"},
            extension_config={
                "flask": {"session_key": "etl_db", "connection_key": "etl_connection", "pool_key": "etl_pool"}
            },
        )
    )
    plugin = SQLSpecPlugin(spec)
    return create_app(plugin)


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

FlaskConfig Options

Key Type Default Description
session_key str "db_session" Key passed to plugin.get_session(key) and cached on flask.g.
connection_key str "db_connection" Key passed to plugin.get_connection(key) on flask.g.
pool_key str "db_pool" Key stored in app.extensions["sqlspec"] for the connection pool.
commit_mode "manual" | "autocommit" | "autocommit_include_redirect" "manual" Commits on 2xx (autocommit) or 2xx/3xx (autocommit_include_redirect) in after_request and rolls back on errors.
extra_commit_statuses set[int] set() Additional HTTP status codes that trigger an automatic commit.
extra_rollback_statuses set[int] set() Additional HTTP status codes that trigger an automatic rollback.
enable_correlation_middleware bool False Binds request correlation headers into CorrelationContext.
enable_sqlcommenter_middleware bool False Populates SQLCommenterContext (framework="flask", route endpoint, method) for wire-level SQL comments.

Recap

  • Initialize SQLSpecPlugin with plugin.init_app(app) and access request-scoped sessions via plugin.get_session().
  • Configure commit_mode and multi-database session_key names in extension_config["flask"].
  • See the Flask Extension Reference for SQLSpecPlugin and FlaskConfig signatures.