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

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

get_async_config(key=None)[source]

Get the async config for the given key.

Return type:: SQLAlchemyAsyncConfig

get_async_engine(key=None)[source]

Get the async engine for the given key.

Return type:: sqlalchemy.ext.asyncio.AsyncEngine

get_async_session(request, key=None)[source]

Get the async session for the given key.

Return type:: AsyncSession

get_config(key=None)[source]

Get the config for the given key.

Return type:: Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]

get_engine(key=None)[source]

Get the engine for the given key.

Return type:: Union[sqlalchemy.engine.Engine, sqlalchemy.ext.asyncio.AsyncEngine]

get_session(request, key=None)[source]

Get the session for the given key.

Return type:: Union[Session, AsyncSession]

get_sync_config(key=None)[source]

Get the sync config for the given key.

Return type:: SQLAlchemySyncConfig

get_sync_engine(key=None)[source]

Get the sync engine for the given key.

Return type:: sqlalchemy.engine.Engine

get_sync_session(request, key=None)[source]

Get the sync session for the given key.

Return type:: Session

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

lifespan(app)[source]

Context manager for lifespan events.

Parameters:: app¶ (Starlette) – The starlette application.
Yields:: None
Return type:: AsyncGenerator[Any, None]

map_configs()[source]

Maps the configs to the session bind keys.

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

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

async on_startup()[source]

Initializes the database.

Return type:: None

provide_async_engine(key=None)[source]

Get the async engine for the given key.

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

provide_async_session(key=None)[source]

Get the async session for the given key.

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

provide_engine(key=None)[source]

Get the engine for the given key.

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

provide_session(key=None)[source]

Get the session for the given key.

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

provide_sync_engine(key=None)[source]

Get the sync engine for the given key.

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

provide_sync_session(key=None)[source]

Get the sync session for the given key.

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

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.

Return type:: Generator[Session, None]

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.

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

sync_session_class: alias of 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.

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

json_deserializer(self, buf)

Deserialize an object from JSON.

Parameters:: buf¶ (bytes-like or str) – The message to decode.
Returns:: obj – The deserialized object.
Return type:: Any

json_serializer()

Serialize JSON field values.

Parameters:: value¶ (Any) – Any JSON serializable value.
Returns:: JSON string representation of the value.
Return type:: str

class advanced_alchemy.extensions.starlette.SQLAlchemyAsyncConfig[source]

Bases: SQLAlchemyAsyncConfig

SQLAlchemy Async config for Starlette.

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

app: Optional[Starlette] = None: The Starlette application instance.

async close_engine()[source]

Close the engine.

Return type:: None

commit_mode: Literal['manual', 'autocommit', 'autocommit_include_redirect'] = 'manual': The commit mode to use for database sessions.

async create_all_metadata()[source]

Create all metadata tables in the database.

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]

engine_key: str = 'db_engine': Key to use for the dependency injection of database engines.

init_app(app)[source]

Initialize the Starlette application with this configuration.

Parameters:: app¶ (Starlette) – The Starlette application instance.
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.

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

async on_startup()[source]

Initialize the Starlette application with this configuration.

Return type:: None

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

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.

class advanced_alchemy.extensions.starlette.SQLAlchemySyncConfig[source]

Bases: SQLAlchemySyncConfig

SQLAlchemy Sync config for Starlette.

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

app: Optional[Starlette] = None: The Starlette application instance.

async close_engine()[source]

Close the engines.

Return type:: None

commit_mode: Literal['manual', 'autocommit', 'autocommit_include_redirect'] = 'manual': The commit mode to use for database sessions.

async create_all_metadata()[source]

Create all metadata tables in the database.

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]

engine_key: str = 'db_engine': Key to use for the dependency injection of database engines.

init_app(app)[source]

Initialize the Starlette application with this configuration.

Parameters:: app¶ (Starlette) – The Starlette application instance.
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.

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

async on_startup()[source]

Initialize the Starlette application with this configuration.

Return type:: None

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

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.

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