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[source]

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)[source]
class advanced_alchemy.extensions.fastapi.AlembicAsyncConfig[source]

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', 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[source]

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', 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[source]

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[source]

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.

json_deserializer(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

__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[source]

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()[source]

Create all metadata tables in the database.

Return type:

None

init_app(app)[source]

Initialize the Starlette application with this configuration.

Parameters:

app (Starlette) – The Starlette application instance.

Return type:

None

async on_startup()[source]

Initialize the Starlette application with this configuration.

Return type:

None

create_session_maker()[source]

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)[source]

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.

Return type:

None

Returns:

None

async middleware_dispatch(request, call_next)[source]

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.

Parameters:

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

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

Returns:

The HTTP response.

Return type:

starlette.responses.Response

async close_engine()[source]

Close the engine.

Return type:

None

async on_shutdown()[source]

Handles the shutdown event by disposing of the SQLAlchemy engine.

Ensures that all connections are properly closed during application shutdown.

Return type:

None

Returns:

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, enable_touch_updated_timestamp_listener=True, bind_key=None, alembic_config=<factory>, 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[source]

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.

__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, enable_touch_updated_timestamp_listener=True, bind_key=None, alembic_config=<factory>, app=None, commit_mode='manual', engine_key='db_engine', session_key='db_session', session_maker_key='session_maker_class')
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()[source]

Create all metadata tables in the database.

Return type:

None

init_app(app)[source]

Initialize the Starlette application with this configuration.

Parameters:

app (Starlette) – The Starlette application instance.

Return type:

None

async on_startup()[source]

Initialize the Starlette application with this configuration.

Return type:

None

create_session_maker()[source]

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)[source]

Handles the session after a request is processed.

Applies the commit strategy and ensures the session is closed.

Parameters:
Return type:

None

Returns:

None

async middleware_dispatch(request, call_next)[source]

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.

Parameters:

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

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

Returns:

The HTTP response.

Return type:

starlette.responses.Response

async close_engine()[source]

Close the engines.

Return type:

None

async on_shutdown()[source]

Handles the shutdown event by disposing of the SQLAlchemy engine.

Ensures that all connections are properly closed during application shutdown.

Return type:

None

Returns:

None

class advanced_alchemy.extensions.fastapi.SyncSessionConfig[source]

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)[source]

Retrieve the Advanced Alchemy extension from a FastAPI application instance.

Return type:

AdvancedAlchemy