FastAPI¶

FastAPI integration with dependency injection helpers for FastAPI's Depends() system, including filter dependency builders.

Plugin¶

class sqlspec.extensions.fastapi.SQLSpecPlugin[source]¶

Bases: SQLSpecPlugin

SQLSpec integration for FastAPI applications.

Extends Starlette integration with dependency injection helpers for FastAPI's Depends() system.

Example

from fastapi import Depends, FastAPI from sqlspec import SQLSpec from sqlspec.adapters.asyncpg import AsyncpgConfig from sqlspec.extensions.fastapi import SQLSpecPlugin

sqlspec = SQLSpec() config = sqlspec.add_config(

AsyncpgConfig(
connection_config={"dsn": "postgresql://localhost/mydb"}, extension_config={

"fastapi": {
"commit_mode": "autocommit", "session_key": "db"

}

}

)

)

app = FastAPI() db_ext = SQLSpecPlugin(sqlspec, app)

@app.get("/users") async def list_users(db = Depends(db_ext.provide_session())):

result = await db.execute("SELECT * FROM users") return {"users": result.all()}

__init__(sqlspec, app=None)[source]¶

Initialize SQLSpec FastAPI extension.

Parameters:

sqlspec¶ (SQLSpec) -- Pre-configured SQLSpec instance with registered configs.
app¶ (FastAPI | None) -- Optional FastAPI application to initialize immediately.

provide_session(key=None)[source]¶

Overloads:

self, key (None) → Callable[[Request], AsyncDriverAdapterBase | SyncDriverAdapterBase]
self, key (str) → Callable[[Request], AsyncDriverAdapterBase | SyncDriverAdapterBase]
self, key (type[AsyncDatabaseConfig]) → Callable[[Request], AsyncDriverAdapterBase]
self, key (type[SyncDatabaseConfig]) → Callable[[Request], SyncDriverAdapterBase]
self, key (AsyncDatabaseConfig) → Callable[[Request], AsyncDriverAdapterBase]
self, key (SyncDatabaseConfig) → Callable[[Request], SyncDriverAdapterBase]

Create dependency factory for session injection.

Returns a callable that can be used with FastAPI's Depends() to inject a database session into route handlers.

Parameters:: key¶ -- Optional session key (str), config type for type narrowing, or None.
Returns:: Dependency callable for FastAPI Depends().

Example

# No args - returns union type @app.get("/users") async def get_users(db = Depends(db_ext.provide_session())):

return await db.execute("SELECT * FROM users")

# String key for multi-database @app.get("/products") async def get_products(db = Depends(db_ext.provide_session("products"))):

return await db.execute("SELECT * FROM products")

# Config instance for type narrowing config = AsyncpgConfig(...) @app.get("/typed") async def typed_query(db = Depends(db_ext.provide_session(config))):

# db is properly typed as AsyncDriverAdapterBase return await db.execute("SELECT 1")

# Config type/class for type narrowing @app.get("/typed2") async def typed_query2(db = Depends(db_ext.provide_session(AsyncpgConfig))):

# db is properly typed as AsyncDriverAdapterBase return await db.execute("SELECT 1")

provide_async_session(key=None)[source]¶

Create dependency factory for async session injection.

Type-narrowed version of provide_session() that returns AsyncDriverAdapterBase. Useful when using string keys and you know the config is async.

Parameters:: key¶ -- Optional session key for multi-database configurations.
Returns:: Dependency callable that returns AsyncDriverAdapterBase.

Example

@app.get("/users") async def get_users(db = Depends(db_ext.provide_async_session())):

# db is AsyncDriverAdapterBase return await db.execute("SELECT * FROM users")

@app.get("/products") async def get_products(db = Depends(db_ext.provide_async_session("products_db"))):

# db is AsyncDriverAdapterBase for "products_db" key return await db.execute("SELECT * FROM products")

provide_sync_session(key=None)[source]¶

Create dependency factory for sync session injection.

Type-narrowed version of provide_session() that returns SyncDriverAdapterBase. Useful when using string keys and you know the config is sync.

Parameters:: key¶ -- Optional session key for multi-database configurations.
Returns:: Dependency callable that returns SyncDriverAdapterBase.

Example

@app.get("/users") def get_users(db = Depends(db_ext.provide_sync_session())):

# db is SyncDriverAdapterBase return db.execute("SELECT * FROM users")

provide_connection(key=None)[source]¶

Overloads:

self, key (None) → Callable[[Request], Any]
self, key (str) → Callable[[Request], Any]
self, key (type[AsyncDatabaseConfig]) → Callable[[Request], Any]
self, key (type[SyncDatabaseConfig]) → Callable[[Request], Any]
self, key (AsyncDatabaseConfig) → Callable[[Request], Any]
self, key (SyncDatabaseConfig) → Callable[[Request], Any]

Create dependency factory for connection injection.

Returns a callable that can be used with FastAPI's Depends() to inject a database connection into route handlers.

Parameters:: key¶ -- Optional session key (str), config type for type narrowing, or None.
Returns:: Dependency callable for FastAPI Depends().

Example

# No args @app.get("/raw") async def raw_query(conn = Depends(db_ext.provide_connection())):

cursor = await conn.cursor() await cursor.execute("SELECT 1") return await cursor.fetchone()

# With config instance config = AsyncpgConfig(...) @app.get("/typed") async def typed_query(conn = Depends(db_ext.provide_connection(config))):

cursor = await conn.cursor() await cursor.execute("SELECT 1") return await cursor.fetchone()

# With config type/class @app.get("/typed2") async def typed_query2(conn = Depends(db_ext.provide_connection(AsyncpgConfig))):

cursor = await conn.cursor() await cursor.execute("SELECT 1") return await cursor.fetchone()

provide_async_connection(key=None)[source]¶

Create dependency factory for async connection injection.

Type-narrowed version of provide_connection() for async connections. Useful when using string keys and you know the config is async.

Parameters:: key¶ -- Optional session key for multi-database configurations.
Returns:: Dependency callable for async connection.

Example

@app.get("/raw") async def raw_query(conn = Depends(db_ext.provide_async_connection())):

cursor = await conn.cursor() await cursor.execute("SELECT 1") return await cursor.fetchone()

provide_sync_connection(key=None)[source]¶

Create dependency factory for sync connection injection.

Type-narrowed version of provide_connection() for sync connections. Useful when using string keys and you know the config is sync.

Parameters:: key¶ -- Optional session key for multi-database configurations.
Returns:: Dependency callable for sync connection.

Example

@app.get("/raw") def raw_query(conn = Depends(db_ext.provide_sync_connection())):

cursor = conn.cursor() cursor.execute("SELECT 1") return cursor.fetchone()

static provide_filters(config, dep_defaults=None)[source]¶

Create filter dependency for FastAPI routes.

Dynamically generates a FastAPI dependency function that parses query parameters into SQLSpec filter objects. The returned callable can be used with FastAPI's Depends() for automatic filter injection.

Parameters:

config¶ -- Filter configuration specifying which filters to enable.
dep_defaults¶ -- Optional dependency defaults for customization.

Returns:

Callable for use with Depends() that returns list of filters.

Example

from fastapi import Depends from sqlspec.extensions.fastapi import FilterConfig

@app.get("/users") async def list_users(

db = Depends(db_ext.provide_session()), filters = Depends(

db_ext.provide_filters({
"id_filter": UUID, "search": "name,email", "search_ignore_case": True, "pagination_type": "limit_offset", "sort_field": ["created_at", "uploaded_collections"],

})

),

):

stmt = sql("SELECT * FROM users") for filter in filters:

stmt = filter.append_to_statement(stmt)

result = await db.execute(stmt) return result.all()

Dependency Helpers¶

provide_filters() supports the same orderBy alias contract as the Litestar provider. Camel-case query values are accepted by default for configured sort_field values. Use sort_field_aliases for explicit API names, or set sort_field_camelize=False to accept only raw configured values. Alias values normalize to fields from sort_field before OrderByFilter is created, so the SQL-facing sort allowlist remains strict.

sqlspec.extensions.fastapi.provide_filters(config, dep_defaults=<sqlspec.extensions.fastapi.providers.DependencyDefaults object>)[source]¶

Create FastAPI dependency provider for filters based on configuration.

This function dynamically generates a FastAPI dependency function that parses query parameters into SQLSpec filter objects.

Parameters:

config¶ -- Filter configuration specifying which filters to enable.
dep_defaults¶ -- Dependency defaults for filter configuration.

Returns:

A FastAPI dependency callable that returns list of filters.

Example

from fastapi import Depends, FastAPI from sqlspec.extensions.fastapi import SQLSpecPlugin, FilterConfig

app = FastAPI() db_ext = SQLSpecPlugin(sql, app)

@app.get("/users") async def list_users(

filters = Depends(

db_ext.provide_filters({
"id_filter": UUID, "search": "name,email", "pagination_type": "limit_offset",

})

),

):

stmt = sql("SELECT * FROM users") for filter in filters:

stmt = filter.append_to_statement(stmt)

result = await db.execute(stmt) return result.all()

class sqlspec.extensions.fastapi.DependencyDefaults[source]¶

Bases: object

Default values for dependency generation.

class sqlspec.extensions.fastapi.FilterConfig[source]¶

Bases: TypedDict

Configuration for generated FastAPI filter dependencies.

All keys are optional. A filter dependency is created only for each enabled key. Field names are SQL-facing allowlist values; generated query parameter names and order-by aliases remain API-facing.

id_filter: NotRequired[type[UUID | int | str]]¶: Type of ID filter to enable. When set, creates an ids collection filter.

id_field: NotRequired[str]¶: SQL-facing field name for ID filtering. Defaults to "id".

sort_field: NotRequired[str | set[str] | list[str]]¶: Allowed SQL-facing field or fields for orderBy sorting.

sort_field_aliases: NotRequired[dict[str, str]]¶: Additional API-facing orderBy aliases mapped to configured sort_field values.

sort_field_camelize: NotRequired[bool]¶: Whether to accept camel-case aliases for configured sort fields. Defaults to True.

sort_order: NotRequired[Literal['asc', 'desc']]¶: Default sort order. Defaults to "desc".

pagination_type: NotRequired[Literal['limit_offset']]¶: Pagination strategy to enable. Currently supports "limit_offset".

pagination_size: NotRequired[int]¶: Default page size for limit/offset pagination.

search: NotRequired[str | set[str]]¶: SQL-facing field or fields to search. Strings may be comma-separated.

search_ignore_case: NotRequired[bool]¶: Whether search filtering is case-insensitive. Defaults to False.

created_at: NotRequired[bool]¶: Whether to enable created_at before/after range filtering.

updated_at: NotRequired[bool]¶: Whether to enable updated_at before/after range filtering.

not_in_fields: NotRequired[FieldNameType | set[FieldNameType] | list[str | FieldNameType]]¶: Field or fields that support NOT IN collection filtering.

in_fields: NotRequired[FieldNameType | set[FieldNameType] | list[str | FieldNameType]]¶: Field or fields that support IN collection filtering.

null_fields: NotRequired[str | set[str]]¶: Field or fields that support IS NULL filtering.

not_null_fields: NotRequired[str | set[str]]¶: Field or fields that support IS NOT NULL filtering.

Middleware¶

class sqlspec.extensions.fastapi.SQLSpecAutocommitMiddleware[source]

Bases: BaseHTTPMiddleware

Middleware for autocommit transaction mode.

Acquires connection, commits on success status codes, rollbacks on error status codes.

__init__(app, config_state, include_redirect=False)[source]

Initialize middleware.

Parameters:

app¶ (Any) -- Starlette application instance.
config_state¶ (SQLSpecConfigState) -- Configuration state for this database.
include_redirect¶ (bool) -- If True, commit on 3xx status codes as well.

async dispatch(request, call_next)[source]

Process request with autocommit transaction mode.

Parameters:

request¶ (Request) -- Incoming HTTP request.
call_next¶ (Any) -- Next middleware or route handler.

Return type:

Any

Returns:

HTTP response.

class sqlspec.extensions.fastapi.SQLSpecManualMiddleware[source]

Bases: BaseHTTPMiddleware

Middleware for manual transaction mode.

Acquires connection from pool, stores in request.state, releases after request. No automatic commit or rollback - user code must handle transactions.

__init__(app, config_state)[source]

Initialize middleware.

Parameters:

app¶ (Any) -- Starlette application instance.
config_state¶ (SQLSpecConfigState) -- Configuration state for this database.

async dispatch(request, call_next)[source]

Process request with manual transaction mode.

Parameters:

request¶ (Request) -- Incoming HTTP request.
call_next¶ (Any) -- Next middleware or route handler.

Return type:

Any

Returns:

HTTP response.