routing#

Configuration classes for read/write replica routing.

RoutingConfig#

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.

ReplicaConfig#

advanced_alchemy.config.routing.ReplicaConfig#: alias of EngineConfig

RoutingStrategy#

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.