FastAPI¶

FastAPI integration extending the Starlette plugin 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={

“starlette”: { # FastAPI uses the “starlette” key
“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]¶

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]¶

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”,

})

),

):

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¶

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 generating dynamic filters for FastAPI.

id_filter: NotRequired[type[UUID | int | str]]¶: Type of ID filter to enable (UUID, int, or str). When set, enables collection filtering by IDs.

id_field: NotRequired[str]¶: Field name for ID filtering. Defaults to ‘id’.

sort_field: NotRequired[str | set[str]]¶: Default field(s) to use for sorting.

sort_order: NotRequired[Literal['asc', 'desc']]¶: Default sort order (‘asc’ or ‘desc’). Defaults to ‘desc’.

pagination_type: NotRequired[Literal['limit_offset']]¶: When set to ‘limit_offset’, enables pagination with page size and current page parameters.

pagination_size: NotRequired[int]¶: Default pagination page size. Defaults to DEFAULT_PAGINATION_SIZE (20).

search: NotRequired[str | set[str]]¶: Field(s) to enable search filtering on. Can be comma-separated string or set of field names.

search_ignore_case: NotRequired[bool]¶: When True, search is case-insensitive. Defaults to False.

created_at: NotRequired[bool]¶: When True, enables created_at date range filtering. Uses ‘created_at’ field by default.

updated_at: NotRequired[bool]¶: When True, enables updated_at date range filtering. Uses ‘updated_at’ field by default.

not_in_fields: NotRequired[FieldNameType | set[FieldNameType]]¶: Fields that support not-in collection filtering. Can be single field or set of fields with type info.

in_fields: NotRequired[FieldNameType | set[FieldNameType]]¶: Fields that support in-collection filtering. Can be single field or set of fields with type info.

null_fields: NotRequired[str | set[str]]¶: Fields that support IS NULL filtering. Can be single field name or set of field names.

not_null_fields: NotRequired[str | set[str]]¶: Fields that support IS NOT NULL filtering. Can be single field name or set of field names.

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.