Skip to content

Query, routing & results

sql 

sql(statement: str) -> TextClause

Wrap a raw SQL string in a SQLAlchemy text() clause.

This is the only supported way to pass raw SQL. Parameters must be bound with :name placeholders — never interpolated (§18.4)::

sql("SELECT * FROM users WHERE id = :user_id")

Query dataclass 

Query(
    name: str,
    statement: Statement,
    operation: Operation = "read",
    timeout: float | None = None,
    idempotent: bool = False,
    expensive: bool = False,
    expected_cardinality: Cardinality | None = None,
    sensitive_parameters: frozenset[str] = frozenset(),
    settings: Mapping[str, str] | None = None,
)

A named, parameterized query with execution metadata (§8.5).

name is a stable logical label used for metrics, tracing, and logs — never the raw SQL text (§18.3). sensitive_parameters are redacted everywhere (§13.4).

is_write property 

is_write: bool

Whether this query mutates data or schema (operation is write/ddl).

QueryRegistry 

QueryRegistry()

In-process registry of named queries for CLI listing and duplicate detection (§8.5).

Create an empty registry.

register 

register(query: Query) -> Query

Register query and return it; raises if the name is already taken by a different :class:Query object (registering the same object twice is a no-op).

get 

get(name: str) -> Query | None

The registered :class:Query named name, or None if none was registered.

names 

names() -> list[str]

Every registered query name, sorted.

all 

all() -> list[Query]

Every registered :class:Query, ordered by name.

default_registry module-attribute 

default_registry = QueryRegistry()

DatabaseTarget dataclass 

DatabaseTarget(
    database: str,
    role: Role = "write",
    shard_key: object | None = None,
)

Where an operation should run (§8.2, §22.1).

role selects primary vs replica. shard_key (any hashable business value, e.g. an organization id) is resolved to a concrete shard by a :class:ShardResolver in Phase 4; in Phase 1 it is carried through and ignored by the single-shard resolver.

wants_replica property 

wants_replica: bool

Whether role prefers a replica (read/prefer_replica).

requires_primary property 

requires_primary: bool

Whether role must hit the primary (write/primary_only).

ShardResolver

Bases: Protocol

Resolve a (database, shard_key) pair to a concrete shard id (§22.2).

resolve 

resolve(database: str, shard_key: object) -> str

Return the shard id shard_key belongs to within database.

SingleShardResolver

Default resolver for single-shard deployments: every key maps to "default".

resolve 

resolve(database: str, shard_key: object) -> str

Always return "default", regardless of shard_key.

HashShardResolver 

HashShardResolver(
    num_shards: int, *, prefix: str = "shard"
)

Hash-based sharding (§22.2): shard_key -> {prefix}-NNN.

Uses SHA-256 rather than the builtin hash() — Python randomizes hash() for str/bytes per process (PYTHONHASHSEED), which would route the same key to different shards across restarts or app instances. Write routing must be deterministic (§22.3).

num_shards fixed buckets, named {prefix}-000, {prefix}-001, etc.

resolve 

resolve(database: str, shard_key: object) -> str

Hash shard_key (via SHA-256) into one of num_shards buckets.

RangeShardResolver 

RangeShardResolver(ranges: Sequence[ShardRange])

Range-based sharding (§22.2): an ordered set of upper bounds, e.g. by tenant-id range.

ranges need not be pre-sorted; they're sorted by upper_bound here.

resolve 

resolve(database: str, shard_key: object) -> str

The shard id of the first range whose upper_bound exceeds shard_key.

ShardRange dataclass 

ShardRange(upper_bound: int, shard_id: str)

A range boundary: keys < upper_bound (and >= the previous range's bound) map here.

DirectoryShardResolver 

DirectoryShardResolver(directory: Mapping[object, str])

Explicit shard_key -> shard_id directory lookup (§22.2).

Missing mappings fail closed (§22.3) rather than silently defaulting to some shard.

Seed the resolver with an initial {shard_key: shard_id} mapping.

resolve 

resolve(database: str, shard_key: object) -> str

Look up shard_key; raises :class:DatabaseRoutingError if unmapped.

set_mapping 

set_mapping(shard_key: object, shard_id: str) -> None

Update the directory, e.g. after an explicit tenant-migration workflow (§22.3).

CallableShardResolver 

CallableShardResolver(fn: Callable[[str, object], str])

Adapts a plain fn(database, shard_key) -> shard_id callback to :class:ShardResolver.

Wrap fn so it satisfies the :class:ShardResolver protocol.

resolve 

resolve(database: str, shard_key: object) -> str

Delegate to the wrapped callback.

ReplicaSelector

Bases: Protocol

Choose a replica name for a read, or None to use the primary (§23).

select 

select(database: str, shard_id: str) -> str | None

Return a replica name for this shard, or None to route to the primary.

RoundRobinReplicaSelector 

RoundRobinReplicaSelector(
    replicas: Mapping[str, Sequence[str]],
)

Cycles through a database's configured replica names in order (§23).

Constructed with an explicit {database: [replica_name, ...]} mapping rather than a config object, so it stays decoupled and independently testable.

replicas maps each database name to its ordered list of replica names.

select 

select(database: str, shard_id: str) -> str | None

The next replica in rotation for database, or None if it has none configured.

set_replicas 

set_replicas(database: str, names: Sequence[str]) -> None

Register/replace database's replica rotation at runtime.

Called by register_database/unregister_database so dynamically added databases participate in read routing; an empty names removes the entry.

WeightedReplicaSelector 

WeightedReplicaSelector(
    replicas: Mapping[str, Sequence[tuple[str, int]]],
    *,
    rand: Callable[[], float] | None = None,
)

Weighted-random replica selection using each replica's configured weight (§23).

replicas maps each database name to its (replica_name, weight) pairs. rand overrides the random source (e.g. for deterministic tests).

select 

select(database: str, shard_id: str) -> str | None

A replica for database chosen at random, weighted by its configured weight.

set_replicas 

set_replicas(
    database: str, specs: Sequence[tuple[str, int]]
) -> None

Register/replace database's weighted replicas at runtime (empty removes).

ExecutionResult dataclass 

ExecutionResult(
    row_count: int,
    query_name: str,
    database_name: str,
    duration_ms: float,
    inserted_primary_key: object | None = None,
    returned_rows: Sequence[RowMapping] = tuple(),
    shard_id: str | None = None,
    role: str | None = None,
    retry_attempt: int = 0,
)

Standard result returned by write methods (§8.3).