starlette¶

API Reference for the Starlette extensions module

Note

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

Starlette extension for Advanced Alchemy.

This module provides Starlette integration for Advanced Alchemy, including session management and service utilities.

class advanced_alchemy.extensions.starlette.AdvancedAlchemy[source]

Bases: object

AdvancedAlchemy integration for Starlette applications.

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

Parameters:

config¶ (advanced_alchemy.config.asyncio.SQLAlchemyAsyncConfig | advanced_alchemy.config.sync.SQLAlchemySyncConfig) – The SQLAlchemy configuration.
app¶ (starlette.applications.Starlette | None) – The Starlette application instance. Defaults to None.

__init__(config, app=None)[source]

Parameters:

config (SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig])
app (Starlette | None)

Return type:

None

property config: Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig]: Current Advanced Alchemy configuration.

init_app(app)[source]

Initializes the Starlette application with SQLAlchemy engine and sessionmaker.

Sets up middleware and shutdown handlers for managing the database engine.

Parameters:

app¶ (starlette.applications.Starlette) – The Starlette application instance.
app (Starlette)

Raises:

advanced_alchemy.exceptions.ImproperConfigurationError – If the application is not initialized.

Return type:

None

lifespan(app)[source]

Context manager for lifespan events.

Parameters:

app¶ (Starlette) – The starlette application.
app (Starlette)

Yields:

None

Return type:

AsyncGenerator[Any, None]

property app: Starlette

Returns the Starlette application instance.

Raises:: advanced_alchemy.exceptions.ImproperConfigurationError – If the application is not initialized.
Returns:: The Starlette application instance.
Return type:: starlette.applications.Starlette

async on_startup()[source]

Initializes the database.

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

map_configs()[source]

Maps the configs to the session bind keys.

Return type:: dict[str, Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]]
Returns:: A dictionary of config bind keys to configs.

get_config(key=None)[source]

Get the config for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the config for.
key (str | None)

Raises:

advanced_alchemy.exceptions.ImproperConfigurationError – If the config is not found.

Return type:

Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]

Returns:

The config for the given key.

get_async_config(key=None)[source]

Get the async config for the given key.

Raises:: advanced_alchemy.exceptions.ImproperConfigurationError – If the config is not found.
Return type:: SQLAlchemyAsyncConfig
Returns:: The async config for the given key.
Parameters:: key (str | None)

get_sync_config(key=None)[source]

Get the sync config for the given key.

Raises:: advanced_alchemy.exceptions.ImproperConfigurationError – If the config is not found.
Return type:: SQLAlchemySyncConfig
Returns:: The sync config for the given key.
Parameters:: key (str | None)

with_async_session(key=None)[source]

Context manager for getting an async session.

Yields:: The async session for the given key.
Return type:: AsyncGenerator[AsyncSession, None]
Parameters:: key (str | None)

with_sync_session(key=None)[source]

Context manager for getting a sync session.

Yields:: The sync session for the given key.
Return type:: Generator[Session, None]
Parameters:: key (str | None)

get_session(request, key=None)[source]

Get the session for the given key.

Parameters:

request¶ (Request) – The request object.
key¶ (Optional[str]) – The key to get the session for.
request (Request)
key (str | None)

Return type:

Union[Session, AsyncSession]

Returns:

The session for the given key.

get_async_session(request, key=None)[source]

Get the async session for the given key.

Parameters:

request¶ (Request) – The request object.
key¶ (Optional[str]) – The key to get the session for.
request (Request)
key (str | None)

Return type:

AsyncSession

Returns:

The async session for the given key.

get_sync_session(request, key=None)[source]

Get the sync session for the given key.

Parameters:

request¶ (Request) – The request object.
key¶ (Optional[str]) – The key to get the session for.
request (Request)
key (str | None)

Return type:

Session

Returns:

The sync session for the given key.

provide_session(key=None)[source]

Get the session for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the session for.
key (str | None)

Return type:

Callable[[Request], Union[Session, AsyncSession]]

Returns:

The session for the given key.

provide_async_session(key=None)[source]

Get the async session for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the session for.
key (str | None)

Return type:

Callable[[Request], AsyncSession]

Returns:

The async session for the given key.

provide_sync_session(key=None)[source]

Get the sync session for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the session for.
key (str | None)

Return type:

Callable[[Request], Session]

Returns:

The sync session for the given key.

get_engine(key=None)[source]

Get the engine for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the engine for.
key (str | None)

Return type:

Union[sqlalchemy.engine.Engine, sqlalchemy.ext.asyncio.AsyncEngine]

Returns:

The engine for the given key.

get_async_engine(key=None)[source]

Get the async engine for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the engine for.
key (str | None)

Return type:

sqlalchemy.ext.asyncio.AsyncEngine

Returns:

The async engine for the given key.

get_sync_engine(key=None)[source]

Get the sync engine for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the engine for.
key (str | None)

Return type:

sqlalchemy.engine.Engine

Returns:

The sync engine for the given key.

provide_engine(key=None)[source]

Get the engine for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the engine for.
key (str | None)

Return type:

Callable[[], Union[sqlalchemy.engine.Engine, sqlalchemy.ext.asyncio.AsyncEngine]]

Returns:

The engine for the given key.

provide_async_engine(key=None)[source]

Get the async engine for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the engine for.
key (str | None)

Return type:

Callable[[], sqlalchemy.ext.asyncio.AsyncEngine]

Returns:

The async engine for the given key.

provide_sync_engine(key=None)[source]

Get the sync engine for the given key.

Parameters:

key¶ (Optional[str]) – The key to get the engine for.
key (str | None)

Return type:

Callable[[], sqlalchemy.engine.Engine]

Returns:

The sync engine for the given key.

class advanced_alchemy.extensions.starlette.AlembicAsyncConfig[source]

Bases: GenericAlembicConfig

Configuration for an Async Alembic’s Config class.

__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.starlette.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', 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.starlette.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.starlette.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.

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.starlette.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.
app (Starlette)

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.
session (AsyncSession)
request (Request)
response (Response)

Return type:

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.

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

__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.starlette.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.

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()[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.
app (Starlette)

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.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)[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.

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

class advanced_alchemy.extensions.starlette.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'>)