routing¶

API Reference for the routing module

Note

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

Read/Write Routing¶

The routing module provides automatic routing of read operations to read replicas while directing write operations to the primary database. This enables better scalability by distributing read load across multiple replica databases.

Key Features¶

Automatic Routing: SELECT queries route to replicas, INSERT/UPDATE/DELETE to primary
Sticky-After-Write: Ensures read-your-writes consistency by routing reads to primary after writes
FOR UPDATE Detection: Automatically routes SELECT ... FOR UPDATE to primary
Multiple Replica Support: Round-robin or random selection across multiple replicas
Agnostic Bind Group Routing: Define and route to arbitrary groups (e.g., “analytics”, “reporting”)
Context Managers: Explicit control with primary_context(), replica_context(), and use_bind_group()
Framework Integration: Built-in support for Litestar, FastAPI, Flask, Sanic, Starlette

Configuration Classes¶

class advanced_alchemy.config.routing.RoutingConfig[source]

Bases: object

Read/Write routing configuration.

This configuration enables automatic routing of database operations to different engine groups (e.g., writer, reader, analytics).

primary_connection_string: str | None = None

Connection string for the primary (write) database. Mapped to engines[default_group].

Type:: Legacy

read_replicas: list[str | EngineConfig]

Read replica connection strings or configs. Mapped to engines[read_group].

Type:: Legacy

engines: dict[str, list[str | EngineConfig]]

Dictionary mapping group names to lists of engine configs.

Example

{
    "writer": ["postgres://primary"],
    "reader": ["postgres://rep1", "postgres://rep2"],
    "analytics": ["postgres://warehouse"]
}

__init__(primary_connection_string=None, read_replicas=<factory>, engines=<factory>, default_group='default', read_group='read', routing_strategy=RoutingStrategy.ROUND_ROBIN, enabled=True, sticky_after_write=True, reset_stickiness_on_commit=True)

Parameters:

primary_connection_string (str | None)
read_replicas (list[str | EngineConfig])
engines (dict[str, list[str | EngineConfig]])
default_group (str)
read_group (str)
routing_strategy (RoutingStrategy)
enabled (bool)
sticky_after_write (bool)
reset_stickiness_on_commit (bool)

Return type:

None

default_group: str = 'default': Name of the group to use for write operations.

read_group: str = 'read': Name of the group to use for read operations.

routing_strategy: RoutingStrategy = 1: Strategy for selecting engines within a group.

enabled: bool = True: Enable/disable routing.

sticky_after_write: bool = True: Stick to writer after first write in context (read-your-writes).

reset_stickiness_on_commit: bool = True: Reset stickiness after commit.

__post_init__()[source]

Normalize configuration.

Return type:: None

get_engine_configs(group)[source]

Get engine configs for a specific group.

Parameters:

group¶ (str) – Name of the engine group.
group (str)

Return type:

list[EngineConfig]

Returns:

List of EngineConfig instances.

advanced_alchemy.config.routing.ReplicaConfig: alias of EngineConfig

class advanced_alchemy.config.routing.RoutingStrategy[source]

Bases: Enum

Strategy for selecting engines from a group.

Determines how the routing layer chooses which engine to use when multiple engines are configured for a routing group.

ROUND_ROBIN = 1: Cycle through engines in order.

RANDOM = 2: Select engines randomly.

Session Classes¶

class advanced_alchemy.routing.RoutingSyncSession[source]¶

Bases: Session

Synchronous session with read/write routing via get_bind().

This session class extends SQLAlchemy’s Session to provide automatic routing of operations to different engine groups (e.g. writer/reader).

The routing decision is made in get_bind() based on: 1. Execution options (bind_group) 2. Context variables (bind_group, force_primary) 3. Stickiness state 4. Operation type (Write vs Read)

_default_engine¶: The default (write) database engine.

_selectors¶: Map of group names to engine selectors.

_routing_config¶: Configuration for routing behavior.

__init__(routing_config, selectors, default_engine, **kwargs)[source]¶

Initialize the routing session.

Parameters:

routing_config¶ (RoutingConfig) – Configuration for routing behavior.
selectors¶ (dict[str, EngineSelector[sqlalchemy.engine.Engine]]) – Map of group names to engine selectors.
default_engine¶ (sqlalchemy.engine.Engine) – The default (fallback/write) engine.
**kwargs¶ (Any) – Additional arguments passed to the parent Session.
routing_config (RoutingConfig)
selectors (dict[str, EngineSelector[MyTypeAliasForwardRef('sqlalchemy.engine.Engine')]])
default_engine (sqlalchemy.engine.Engine)
kwargs (Any)

Return type:

None

get_bind(mapper=None, clause=None, **kwargs)[source]¶

Route to appropriate engine based on operation and context.

Parameters:

mapper¶ – Optional mapper for the operation.
clause¶ – The SQL clause being executed.
**kwargs¶ – Additional keyword arguments.
mapper (Mapper[Any] | type[Any] | None)
clause (Any | None)
kwargs (Any)

Returns:

The selected engine.

Return type:

Engine

commit()[source]¶

Commit the transaction and reset routing state.

Return type:: None

rollback()[source]¶

Rollback the transaction and reset routing state.

Return type:: None

class advanced_alchemy.routing.RoutingAsyncSession[source]¶

Bases: AsyncSession

Async session with read/write routing support.

Wraps RoutingSyncSession to provide async routing capabilities.

sync_session_class¶: alias of RoutingSyncSession

__init__(routing_config, selectors, default_engine, **kwargs)[source]¶

Initialize the async routing session.

Parameters:

routing_config¶ (RoutingConfig) – Configuration for routing behavior.
selectors¶ (dict[str, EngineSelector[sqlalchemy.ext.asyncio.AsyncEngine]]) – Map of group names to async engine selectors.
default_engine¶ (sqlalchemy.ext.asyncio.AsyncEngine) – The default (fallback/write) async engine.
**kwargs¶ (Any) – Additional arguments passed to the parent AsyncSession.
routing_config (RoutingConfig)
selectors (dict[str, EngineSelector[TypeAliasForwardRef('sqlalchemy.ext.asyncio.AsyncEngine')]])
default_engine (sqlalchemy.ext.asyncio.AsyncEngine)
kwargs (Any)

Return type:

None

property primary_engine: sqlalchemy.ext.asyncio.AsyncEngine¶

Get the primary (default) async engine.

Returns:: The default database engine.

property routing_config: RoutingConfig¶

Get the routing configuration.

Returns:: The routing configuration.

Session Makers¶

class advanced_alchemy.routing.RoutingAsyncSessionMaker[source]¶

Bases: object

Factory for creating async routing sessions.

This class creates RoutingAsyncSession instances with properly configured async engines and routing selectors.

Example

Creating an async routing session maker:

maker = RoutingAsyncSessionMaker(
    routing_config=RoutingConfig(
        engines={
            "writer": ["postgresql+asyncpg://primary"],
            "reader": ["postgresql+asyncpg://replica1"],
        }
    ),
    engine_config={"pool_size": 10},
)

async with maker() as session:
    result = await session.execute(select(User))

__init__(routing_config, engine_config=None, session_config=None, create_engine_callable=create_async_engine)[source]¶

Initialize the async session maker.

Parameters:

routing_config¶ (RoutingConfig) – Configuration for read/write routing.
engine_config¶ (Optional[dict[str, Any]]) – Configuration options for engine creation.
session_config¶ (Optional[dict[str, Any]]) – Configuration options for session creation.
create_engine_callable¶ (Callable[[str], AsyncEngine]) – Callable to create async engines (for testing).
routing_config (RoutingConfig)
engine_config (dict[str, Any] | None)
session_config (dict[str, Any] | None)
create_engine_callable (Callable[[str], AsyncEngine])

Return type:

None

__call__()[source]¶

Create a new async routing session.

Any bind passed in the session config is ignored because routing controls bind selection.

Return type:: RoutingAsyncSession
Returns:: A new RoutingAsyncSession instance.

property primary_engine: AsyncEngine¶

Get the primary (default) async engine.

Returns:: The primary database async engine.

property replica_engines: list[AsyncEngine]¶

Get the replica async engines (from read_group).

Returns:: List of replica database async engines.

async close_all()[source]¶

Close all engines and release connections.

Call this when shutting down to properly release database connections.

Return type:: None

class advanced_alchemy.routing.RoutingSyncSessionMaker[source]¶

Bases: object

Factory for creating sync routing sessions.

This class creates RoutingSyncSession instances with properly configured engines and routing selectors.

Example

Creating a routing session maker:

maker = RoutingSyncSessionMaker(
    routing_config=RoutingConfig(
        engines={
            "writer": ["postgresql://primary"],
            "reader": ["postgresql://replica1"],
        }
    ),
    engine_config={"pool_size": 10},
)

session = maker()

__init__(routing_config, engine_config=None, session_config=None, create_engine_callable=create_engine)[source]¶

Initialize the session maker.

Parameters:

routing_config¶ (RoutingConfig) – Configuration for read/write routing.
engine_config¶ (Optional[dict[str, Any]]) – Configuration options for engine creation.
session_config¶ (Optional[dict[str, Any]]) – Configuration options for session creation.
create_engine_callable¶ (Callable[[str], Engine]) – Callable to create engines (for testing).
routing_config (RoutingConfig)
engine_config (dict[str, Any] | None)
session_config (dict[str, Any] | None)
create_engine_callable (Callable[[str], Engine])

Return type:

None

__call__()[source]¶

Create a new routing session.

Any bind passed in the session config is ignored because routing controls bind selection.

Return type:: RoutingSyncSession
Returns:: A new RoutingSyncSession instance.

property primary_engine: Engine¶

Get the primary (default) engine.

Returns:: The primary database engine.

property replica_engines: list[Engine]¶

Get the replica engines (from read_group).

Returns:: List of replica database engines.

close_all()[source]¶

Close all engines and release connections.

Call this when shutting down to properly release database connections.

Return type:: None

Replica Selectors¶

advanced_alchemy.routing.ReplicaSelector¶: alias of EngineSelector

class advanced_alchemy.routing.RoundRobinSelector[source]¶

Bases: EngineSelector[EngineT]

Round-robin engine selection.

Cycles through engines in order, distributing load evenly across all available engines.

This selector is thread-safe.

Example

Creating a round-robin selector:

selector = RoundRobinSelector(engines)
engine1 = selector.next()
engine2 = selector.next()
engine3 = selector.next()

This cycles through engines in order and wraps back to the first.

__init__(engines)[source]¶

Initialize the round-robin selector.

Parameters:

engines¶ (list[TypeVar(EngineT, bound= Union[Engine, AsyncEngine])]) – List of database engines.
engines (list[EngineT])

Return type:

None

next()[source]¶

Select the next engine in round-robin order.

Return type:: TypeVar(EngineT, bound= Union[Engine, AsyncEngine])
Returns:: The next engine in the cycle.
Raises:: RuntimeError – If no engines are configured.

class advanced_alchemy.routing.RandomSelector[source]¶

Bases: EngineSelector[EngineT]

Random engine selection.

Selects engines randomly, which can help with load distribution when engines have varying capacity or when you want to avoid predictable patterns.

Example

Creating a random selector:

selector = RandomSelector(engines)
engine = selector.next()

next()[source]¶

Select a random engine.

Return type:: TypeVar(EngineT, bound= Union[Engine, AsyncEngine])
Returns:: A randomly selected engine.
Raises:: RuntimeError – If no engines are configured.

Context Managers¶

advanced_alchemy.routing.primary_context()[source]¶

Force all operations to use primary within this context.

Use this context manager when you need to ensure all database operations (including reads) go to the primary database.

Example

Force a specific query to use the primary database:

from advanced_alchemy.routing import primary_context

with primary_context():
    user = await repo.get(user_id)
    orders = await order_repo.list()

Yields:: None
Return type:: Generator[None, None, None]

advanced_alchemy.routing.replica_context()[source]¶

Force read operations to use replicas (temporarily disable stickiness).

Use this context manager when you want to explicitly allow reads to go to replicas, even if a previous write has set the sticky-to-primary state.

Warning

Use with caution! This can lead to read-after-write inconsistency if you’re reading data that was recently written.

Example

Allow reads to use replicas after a write:

from advanced_alchemy.routing import replica_context

await repo.add(user)

user = await repo.get(user_id)

with replica_context():
    users = await repo.list()

Yields:: None
Return type:: Generator[None, None, None]

advanced_alchemy.routing.use_bind_group(name)[source]¶

Force operations to use a specific bind group.

Use this context manager to route operations to a specific group of engines, such as “analytics” or “reporting”.

Example

Route a query to the analytics database:

from advanced_alchemy.routing import use_bind_group

with use_bind_group("analytics"):
    data = await repo.list()

Parameters:

name¶ (str) – Name of the bind group to use.
name (str)

Yields:

None

Return type:

Generator[None, None, None]

advanced_alchemy.routing.reset_routing_context()[source]¶

Reset all routing context variables to their defaults.

This is typically called after a commit or rollback to allow subsequent reads to use replicas again.

Example

Manual reset after transaction:

from advanced_alchemy.routing import reset_routing_context

await session.commit()
reset_routing_context()

Return type:: None

Context Variables¶

advanced_alchemy.routing.stick_to_primary_var¶

advanced_alchemy.routing.force_primary_var¶