fastapi

API Reference for the FastAPI extensions module

Note

Private methods and attributes are not included in the API reference.

---

FastAPI extension for Advanced Alchemy.

This module provides FastAPI integration for Advanced Alchemy, including session management, database migrations, and service utilities.

class advanced_alchemy.extensions.fastapi.AdvancedAlchemy

Bases: AdvancedAlchemy

AdvancedAlchemy integration for FastAPI applications.

This class manages SQLAlchemy sessions and engine lifecycle within a FastAPI application. It provides middleware for handling transactions based on commit strategies.

__init__(config, app=None)

provide_service(service_class, /, key=None, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, count_with_window_function=None)

Provides a service instance for dependency injection.

Parameters:

• service_class – The service class to provide.

• key – Optional key for the service.

• statement – Optional SQLAlchemy statement.

• error_messages – Optional error messages.

• load – Optional load specification.

• execution_options – Optional execution options.

• uniquify – Optional flag to uniquify the service.

• count_with_window_function – Optional flag to use window function for counting.

• service_class (type[AsyncServiceT_co | SyncServiceT_co])

• key (Optional[str])

• statement (Optional[Select[tuple[ModelT]]])

• error_messages (Optional[Union[ErrorMessages, EmptyType]])

• load (Optional[LoadSpec])

• execution_options (Optional[dict[str, Any]])

• uniquify (bool | None)

• count_with_window_function (bool | None)

Returns:

A callable that returns an async generator for async services or a generator for sync services.

Return type:

Callable[…, Union[AsyncGenerator[AsyncServiceT_co, None], Generator[SyncServiceT_co, None, None]]]

static provide_filters(config, /, dep_defaults=None)

Provides filters for dependency injection.

Parameters:

• config – The filters to provide.

• dep_defaults – Optional key for the filters.

• config (FilterConfig)

• dep_defaults (Optional[DependencyDefaults])

Returns:

A callable that returns an async generator for async filters or a generator for sync filters.

Return type:

Callable[…, list[filters.FilterTypes]]

class advanced_alchemy.extensions.fastapi.AlembicAsyncConfig

Bases: GenericAlembicConfig

Configuration for an Async Alembic’s Config class.

See also

https://alembic.sqlalchemy.org/en/latest/api/config.html

__init__(script_config='alembic.ini', version_table_name='alembic_versions', version_table_schema=None, script_location='migrations', toml_file=None, user_module_prefix='sa.', render_as_batch=True, compare_type=False, template_path='/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates')

class advanced_alchemy.extensions.fastapi.AlembicSyncConfig

Bases: GenericAlembicConfig

Configuration for Alembic’s synchronous migrations.

For details see: https://alembic.sqlalchemy.org/en/latest/api/config.html

__init__(script_config='alembic.ini', version_table_name='alembic_versions', version_table_schema=None, script_location='migrations', toml_file=None, user_module_prefix='sa.', render_as_batch=True, compare_type=False, template_path='/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates')

class advanced_alchemy.extensions.fastapi.AsyncSessionConfig

Bases: GenericSessionConfig[AsyncConnection, AsyncEngine, AsyncSession]

SQLAlchemy async session config.

sync_session_class

alias of Empty

__init__(autobegin=<class 'advanced_alchemy.utils.dataclass.Empty'>, autoflush=<class 'advanced_alchemy.utils.dataclass.Empty'>, bind=<class 'advanced_alchemy.utils.dataclass.Empty'>, binds=<class 'advanced_alchemy.utils.dataclass.Empty'>, class_=<class 'advanced_alchemy.utils.dataclass.Empty'>, expire_on_commit=<class 'advanced_alchemy.utils.dataclass.Empty'>, info=<class 'advanced_alchemy.utils.dataclass.Empty'>, join_transaction_mode=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cls=<class 'advanced_alchemy.utils.dataclass.Empty'>, twophase=<class 'advanced_alchemy.utils.dataclass.Empty'>, sync_session_class=<class 'advanced_alchemy.utils.dataclass.Empty'>)

class advanced_alchemy.extensions.fastapi.EngineConfig

Bases: EngineConfig

Configuration for SQLAlchemy’s Engine.

This class extends the base EngineConfig with Starlette-specific JSON serialization options.

For details see: https://docs.sqlalchemy.org/en/20/core/engines.html

json_deserializer

Callable for converting JSON strings to Python objects.

json_serializer

Callable for converting Python objects to JSON strings.

Parameters:

value (Any)

Return type:

str

json_deserializer(self, buf)

For dialects that support the JSON datatype, this is a Python callable that will convert a JSON string to a Python object. But default, this uses the built-in serializers.

json_serializer()

For dialects that support the JSON datatype, this is a Python callable that will render a given object as JSON. By default, By default, the built-in serializer is used.

Return type:

str

Parameters:

value (Any)

__init__(connect_args=<class 'advanced_alchemy.utils.dataclass.Empty'>, echo=<class 'advanced_alchemy.utils.dataclass.Empty'>, echo_pool=<class 'advanced_alchemy.utils.dataclass.Empty'>, enable_from_linting=<class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options=<class 'advanced_alchemy.utils.dataclass.Empty'>, hide_parameters=<class 'advanced_alchemy.utils.dataclass.Empty'>, insertmanyvalues_page_size=<class 'advanced_alchemy.utils.dataclass.Empty'>, isolation_level=<class 'advanced_alchemy.utils.dataclass.Empty'>, json_deserializer=<built-in method decode of msgspec.json.Decoder object>, json_serializer=<function serializer>, label_length=<class 'advanced_alchemy.utils.dataclass.Empty'>, logging_name=<class 'advanced_alchemy.utils.dataclass.Empty'>, max_identifier_length=<class 'advanced_alchemy.utils.dataclass.Empty'>, max_overflow=<class 'advanced_alchemy.utils.dataclass.Empty'>, module=<class 'advanced_alchemy.utils.dataclass.Empty'>, paramstyle=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool=<class 'advanced_alchemy.utils.dataclass.Empty'>, poolclass=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_logging_name=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_pre_ping=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_size=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_recycle=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_reset_on_return=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_timeout=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_use_lifo=<class 'advanced_alchemy.utils.dataclass.Empty'>, plugins=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cache_size=<class 'advanced_alchemy.utils.dataclass.Empty'>, use_insertmanyvalues=<class 'advanced_alchemy.utils.dataclass.Empty'>)

class advanced_alchemy.extensions.fastapi.SQLAlchemyAsyncConfig

Bases: SQLAlchemyAsyncConfig

SQLAlchemy Async config for Starlette.

app: Starlette | None = None

The Starlette application instance.

commit_mode: Literal['manual', 'autocommit', 'autocommit_include_redirect'] = 'manual'

The commit mode to use for database sessions.

engine_key: str = 'db_engine'

Key to use for the dependency injection of database engines.

session_key: str = 'db_session'

Key to use for the dependency injection of database sessions.

session_maker_key: str = 'session_maker_class'

Key under which to store the SQLAlchemy sessionmaker in the application state instance.

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

async create_all_metadata()

Create all metadata tables in the database.

Return type:

None

init_app(app)

Initialize the Starlette application with this configuration.

Parameters:

• app (Starlette) – The Starlette application instance.

• app (Starlette)

Return type:

None

async on_startup()

Initialize the Starlette application with this configuration.

Return type:

None

create_session_maker()

Get a session maker. If none exists yet, create one.

Returns:

Session factory used by the plugin.

Return type:

Callable[[], Session]

async session_handler(session, request, response)

Handles the session after a request is processed.

Applies the commit strategy and ensures the session is closed.

Parameters:

• session (sqlalchemy.ext.asyncio.AsyncSession) – The database session.

• request (starlette.requests.Request) – The incoming HTTP request.

• response (starlette.responses.Response) – The outgoing HTTP response.

• session (AsyncSession)

• request (Request)

• response (Response)

Return type:

None

async middleware_dispatch(request, call_next)

Middleware dispatch function to handle requests and responses.

Processes the request, invokes the next middleware or route handler, and applies the session handler after the response is generated.

For generator-managed sessions (e.g., from provide_service()), the middleware stores the response status but skips cleanup, allowing the generator to handle commit/rollback/close operations properly.

Parameters:

• request (starlette.requests.Request) – The incoming HTTP request.

• call_next (starlette.middleware.base.RequestResponseEndpoint) – The next middleware or route handler.

• request (Request)

• call_next (Callable[[Request], Awaitable[Response]])

Returns:

The HTTP response.

Return type:

starlette.responses.Response

async close_engine()

Close the engine.

Return type:

None

async on_shutdown()

Handles the shutdown event by disposing of the SQLAlchemy engine.

Ensures that all connections are properly closed during application shutdown.

Return type:

None

__init__(create_engine_callable=<function create_async_engine>, session_config=<factory>, session_maker_class=<class 'sqlalchemy.ext.asyncio.session.async_sessionmaker'>, connection_string=None, engine_config=<factory>, session_maker=None, engine_instance=None, create_all=False, metadata=None, bind_key=None, enable_touch_updated_timestamp_listener=True, enable_file_object_listener=True, file_object_raise_on_error=True, alembic_config=<factory>, routing_config=None, app=None, commit_mode='manual', engine_key='db_engine', session_key='db_session', session_maker_key='session_maker_class')

class advanced_alchemy.extensions.fastapi.SQLAlchemySyncConfig

Bases: SQLAlchemySyncConfig

SQLAlchemy Sync config for Starlette.

app: Starlette | None = None

The Starlette application instance.

commit_mode: Literal['manual', 'autocommit', 'autocommit_include_redirect'] = 'manual'

The commit mode to use for database sessions.

engine_key: str = 'db_engine'

Key to use for the dependency injection of database engines.

__init__(create_engine_callable=<function create_engine>, session_config=<factory>, session_maker_class=<class 'sqlalchemy.orm.session.sessionmaker'>, connection_string=None, engine_config=<factory>, session_maker=None, engine_instance=None, create_all=False, metadata=None, bind_key=None, enable_touch_updated_timestamp_listener=True, enable_file_object_listener=True, file_object_raise_on_error=True, alembic_config=<factory>, routing_config=None, app=None, commit_mode='manual', engine_key='db_engine', session_key='db_session', session_maker_key='session_maker_class')

session_key: str = 'db_session'

Key to use for the dependency injection of database sessions.

session_maker_key: str = 'session_maker_class'

Key under which to store the SQLAlchemy sessionmaker in the application state instance.

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

async create_all_metadata()

Create all metadata tables in the database.

Return type:

None

init_app(app)

Initialize the Starlette application with this configuration.

Parameters:

• app (Starlette) – The Starlette application instance.

• app (Starlette)

Return type:

None

async on_startup()

Initialize the Starlette application with this configuration.

Return type:

None

create_session_maker()

Get a session maker. If none exists yet, create one.

Returns:

Session factory used by the plugin.

Return type:

Callable[[], Session]

async session_handler(session, request, response)

Handles the session after a request is processed.

Applies the commit strategy and ensures the session is closed.

Parameters:

• session (sqlalchemy.orm.Session | sqlalchemy.ext.asyncio.AsyncSession) – The database session.

• request (starlette.requests.Request) – The incoming HTTP request.

• response (starlette.responses.Response) – The outgoing HTTP response.

• session (Session)

• request (Request)

• response (Response)

Return type:

None

async middleware_dispatch(request, call_next)

Middleware dispatch function to handle requests and responses.

Processes the request, invokes the next middleware or route handler, and applies the session handler after the response is generated.

For generator-managed sessions (e.g., from provide_service()), the middleware stores the response status but skips cleanup, allowing the generator to handle commit/rollback/close operations properly.

Parameters:

• request (starlette.requests.Request) – The incoming HTTP request.

• call_next (starlette.middleware.base.RequestResponseEndpoint) – The next middleware or route handler.

• request (Request)

• call_next (Callable[[Request], Awaitable[Response]])

Returns:

The HTTP response.

Return type:

starlette.responses.Response

async close_engine()

Close the engines.

Return type:

None

async on_shutdown()

Handles the shutdown event by disposing of the SQLAlchemy engine.

Ensures that all connections are properly closed during application shutdown.

Return type:

None

class advanced_alchemy.extensions.fastapi.SyncSessionConfig

Bases: GenericSessionConfig[Connection, Engine, Session]

Configuration for synchronous SQLAlchemy sessions.

__init__(autobegin=<class 'advanced_alchemy.utils.dataclass.Empty'>, autoflush=<class 'advanced_alchemy.utils.dataclass.Empty'>, bind=<class 'advanced_alchemy.utils.dataclass.Empty'>, binds=<class 'advanced_alchemy.utils.dataclass.Empty'>, class_=<class 'advanced_alchemy.utils.dataclass.Empty'>, expire_on_commit=<class 'advanced_alchemy.utils.dataclass.Empty'>, info=<class 'advanced_alchemy.utils.dataclass.Empty'>, join_transaction_mode=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cls=<class 'advanced_alchemy.utils.dataclass.Empty'>, twophase=<class 'advanced_alchemy.utils.dataclass.Empty'>)

advanced_alchemy.extensions.fastapi.get_database_migration_plugin(app)

Retrieve the Advanced Alchemy extension from a FastAPI application instance.

Parameters:

• app (FastAPI) – The FastAPI application instance.

• app (FastAPI)

Raises:

ImproperConfigurationError – If the Advanced Alchemy extension is not properly configured.

Return type:

AdvancedAlchemy

Returns:

The Advanced Alchemy extension instance.