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]

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.
Return type:: 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
Returns:: None

map_configs()[source]

Maps the configs to the session bind keys.

Return type:: dict[str, Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]]

get_config(key=None)[source]

Get the config for the given key.

Return type:: Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]

get_async_config(key=None)[source]

Get the async config for the given key.

Return type:: SQLAlchemyAsyncConfig

get_sync_config(key=None)[source]

Get the sync config for the given key.

Return type:: SQLAlchemySyncConfig

with_async_session(key=None)[source]

Context manager for getting an async session.

Return type:: AsyncGenerator[AsyncSession, None]

with_sync_session(key=None)[source]: Context manager for getting a sync session.

get_session(request, key=None)[source]

Get the session for the given key.

Return type:: Session | AsyncSession

get_async_session(request, key=None)[source]

Get the async session for the given key.

Return type:: AsyncSession

get_sync_session(request, key=None)[source]

Get the sync session for the given key.

Return type:: Session

provide_session(key=None)[source]

Get the session for the given key.

Return type:: Callable[[Request], Session | AsyncSession]

provide_async_session(key=None)[source]

Get the async session for the given key.

Return type:: Callable[[Request], AsyncSession]

provide_sync_session(key=None)[source]

Get the sync session for the given key.

Return type:: Callable[[Request], Session]

get_engine(key=None)[source]

Get the engine for the given key.

Return type:: Engine | AsyncEngine

get_async_engine(key=None)[source]

Get the async engine for the given key.

Return type:: sqlalchemy.ext.asyncio.AsyncEngine

get_sync_engine(key=None)[source]

Get the sync engine for the given key.

Return type:: sqlalchemy.engine.Engine

provide_engine(key=None)[source]

Get the engine for the given key.

Return type:: Callable[[], Engine | AsyncEngine]

provide_async_engine(key=None)[source]

Get the async engine for the given key.

Return type:: Callable[[], sqlalchemy.ext.asyncio.AsyncEngine]

provide_sync_engine(key=None)[source]

Get the sync engine for the given key.

Return type:: Callable[[], sqlalchemy.engine.Engine]

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

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

__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:

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.

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