asyncio

class advanced_alchemy.config.asyncio.AlembicAsyncConfig

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', 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.config.asyncio.AsyncSessionConfig

Bases: GenericSessionConfig[AsyncConnection, AsyncEngine, AsyncSession]

SQLAlchemy async session config.

sync_session_class

A Session subclass or other callable which will be used to construct the Session which will be proxied. This parameter may be used to provide custom Session subclasses. Defaults to the AsyncSession.sync_session_class class-level attribute.

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.config.asyncio.SQLAlchemyAsyncConfig

Bases: GenericSQLAlchemyConfig[AsyncEngine, AsyncSession, async_sessionmaker[AsyncSession]]

Async SQLAlchemy Configuration.

Note

The alembic configuration options are documented in the Alembic documentation.

Example

Basic async configuration:

config = SQLAlchemyAsyncConfig(
    connection_string="postgresql+asyncpg://user:pass@localhost/db",
)

Configuration with read/write routing:

from advanced_alchemy.config.routing import RoutingConfig

config = SQLAlchemyAsyncConfig(
    routing_config=RoutingConfig(
        primary_connection_string="postgresql+asyncpg://user:pass@primary/db",
        read_replicas=[
            "postgresql+asyncpg://user:pass@replica/db"
        ],
    ),
)

create_engine_callable(**kw)

Callable that creates an AsyncEngine instance or instance of its subclass.

Parameters:

• url (Union[str, URL])

• kw (Any)

Return type:

AsyncEngine

session_config: AsyncSessionConfig

Configuration options for the async_sessionmaker.

session_maker_class

Sessionmaker class to use.

alias of async_sessionmaker

alembic_config: AlembicAsyncConfig

Configuration for the SQLAlchemy Alembic migrations.

The configuration options are documented in the Alembic documentation.

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

routing_config: RoutingConfig | None = None

Optional read/write routing configuration.

When provided, enables automatic routing of read operations to replicas and write operations to the primary database.

Note

When using routing_config, do not set connection_string. The primary connection is specified in the routing config.

create_session_maker()

Get a session maker.

If routing is configured, returns a routing-aware session maker. Otherwise, returns a standard session maker.

Returns:

A callable that creates session instances.

Return type:

Callable[[], AsyncSession]

get_session()

Get a session from the session maker.

Yields:

AsyncGenerator[AsyncSession, None] – An async context manager that yields an AsyncSession.

Return type:

AsyncGenerator[AsyncSession, None]