asyncio

class advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.SQLAlchemyAsyncConfig

Bases: SQLAlchemyAsyncConfig

Async SQLAlchemy Configuration.

before_send_handler: BeforeMessageSendHookHandler | None | Literal['autocommit', 'autocommit_include_redirects'] = None

Handler to call before the ASGI message is sent.

The handler should handle closing the session stored in the ASGI scope, if it’s still open, and committing and uncommitted data.

engine_dependency_key: str = 'db_engine'

Key to use for the dependency injection of database engines.

session_dependency_key: str = 'db_session'

Key to use for the dependency injection of database sessions.

engine_app_state_key: str = 'db_engine'

Key under which to store the SQLAlchemy engine in the application State instance.

session_maker_app_state_key: str = 'session_maker_class'

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

session_scope_key: str = '_sqlalchemy_db_session'

Key under which to store the SQLAlchemy scope in the application.

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

create_session_maker() → Callable[[], AsyncSession]

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

Returns:

Session factory used by the plugin.

__init__(create_engine_callable: Callable[[str], AsyncEngine] = <function create_async_engine>, session_config: AsyncSessionConfig = <factory>, session_maker_class: type[async_sessionmaker[AsyncSession]] = <class 'sqlalchemy.ext.asyncio.session.async_sessionmaker'>, connection_string: str | None = None, engine_config: EngineConfig = <factory>, session_maker: Callable[[], SessionT] | None = None, engine_instance: EngineT | None = None, create_all: bool = False, metadata: MetaData | None = None, enable_touch_updated_timestamp_listener: bool = True, alembic_config: AlembicAsyncConfig = <factory>, before_send_handler: BeforeMessageSendHookHandler | None | Literal['autocommit', 'autocommit_include_redirects'] = None, engine_dependency_key: str = 'db_engine', session_dependency_key: str = 'db_session', engine_app_state_key: str = 'db_engine', session_maker_app_state_key: str = 'session_maker_class', session_scope_key: str = '_sqlalchemy_db_session') → None

provide_engine(state: State) → AsyncEngine

Create an engine instance.

Parameters:

state – The Litestar.state instance.

Returns:

An engine instance.

provide_session(state: State, scope: Scope) → AsyncSession

Create a session instance.

Parameters:

• state – The Litestar.state instance.

• scope – The current connection’s scope.

Returns:

A session instance.

property signature_namespace: dict[str, Any]

Return the plugin’s signature namespace.

Returns:

A string keyed dict of names to be added to the namespace for signature forward reference resolution.

async create_all_metadata(app: Litestar) → None

Create all metadata

Parameters:

app (Litestar) – The Litestar instance

create_app_state_items() → dict[str, Any]

Key/value pairs to be stored in application state.

update_app_state(app: Litestar) → None

Set the app state with engine and session.

Parameters:

app – The Litestar instance.

async advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.default_before_send_handler(message: Message, scope: Scope) → None

Handle commit/rollback, closing and cleaning up sessions before sending.

Parameters:

• message – ASGI-Message

• scope – An ASGI-Scope

Returns:

None

async advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.autocommit_before_send_handler(message: Message, scope: Scope) → None

Handle commit/rollback, closing and cleaning up sessions before sending.

Parameters:

• message – ASGI-Message

• scope – An ASGI-Scope

Returns:

None

advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.default_handler_maker(session_scope_key: str = '_sqlalchemy_db_session') → Callable[[Message, Scope], Coroutine[Any, Any, None]]

Set up the handler to issue a transaction commit or rollback based on specified status codes :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.default_handler_maker.session_scope_key: The key to use within the application state

Returns:

The handler callable

advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.autocommit_handler_maker(commit_on_redirect: bool = False, extra_commit_statuses: set[int] | None = None, extra_rollback_statuses: set[int] | None = None, session_scope_key: str = '_sqlalchemy_db_session') → Callable[[Message, Scope], Coroutine[Any, Any, None]]

Set up the handler to issue a transaction commit or rollback based on specified status codes :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.autocommit_handler_maker.commit_on_redirect: Issue a commit when the response status is a redirect (3XX) :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.autocommit_handler_maker.session_scope_key: The key to use within the application state

Returns:

The handler callable