Dynamic Registration¶

Register databases whose DSNs are discovered at runtime — per-tenant shards resolved from a service registry, credentials rotated by a secret manager — without maintaining an application-side engine registry.

Bootstrap dynamic-first¶

An explicitly empty databases: {} mapping is valid; every database arrives later:

from dbkit import AsyncDatabase, DatabaseTarget, Query, sql

db = AsyncDatabase.from_config({
    "databases": {},                # dynamic-first
    "max_databases": 500,           # LRU cap on dynamic registrations (optional)
    "max_engines": 64,              # LRU cap on live engines/pools (optional)
    "evict_lru_engines": True,
    "defaults": {
        "pool": {"size": 10, "max_overflow": 0, "timeout_seconds": 2.0},
        "retry": {"attempts": 2},
        "circuit_breaker": {"enabled": True},
    },
})
await db.start()

`ensure_database` — the per-request pattern¶

Idempotent: a lock-free no-op when the config is unchanged (a few µs), so call it in front of every query routed by runtime topology:

shard = my_topology.resolve(tenant_id)          # -> name + DSN, however you store it

await db.ensure_database(shard.name, {
    "primary": {"url": shard.dsn, "required": False},
    "concurrency": {"reads": 20},
})
rows = await db.fetch_all(
    LIST_ORDERS, {"tenant": tenant_id},
    target=DatabaseTarget(database=shard.name, role="read"),
)

A changed config re-registers in place: old engines are disposed (idle-only — in-flight work finishes), and the next query connects with the new settings. This covers host moves and password-only rotations with no restart.

Explicit control¶

replaced = await db.register_database("tenant-42", config)   # register or replace
await db.unregister_database("tenant-42")                    # engines + resilience state freed

async with db.database_scope("migration-src", config) as target:   # scoped lifetime
    rows = await db.fetch_all(EXPORT, target=target)
# auto-unregistered — pools released even on error

Scope database_scope to a unit of work, never to a request

Engines/pools are created inside the block and disposed on exit — a per-request scope recreates connections every request and defeats pooling. Long-lived services use ensure_database/register_database and keep shards registered.

Guarantees¶

Copy-on-write config swap — readers never observe a partially updated database map; the query hot path takes no registration lock (~350 ns regardless of shard count).
Admission control — registering past connection_budget.maximum_per_process (with enforce_at_startup: true) raises before anything is swapped in.
max_databases LRU — least-recently-ensured dynamic databases are fully purged (engines, concurrency limiter, circuit breakers, replica-selector entry) beyond the cap. Statically configured databases are never evicted.
Replicas participate immediately: registered replicas update the built-in round-robin/weighted selectors (custom selectors may implement set_replicas).

API reference¶

ensure_database `async` ¶

ensure_database(
    name: str,
    config: DatabaseConfig | Mapping[str, Any],
    *,
    connect: bool | None = None,
) -> bool

Idempotent :meth:register_database: act only when name is missing or its config changed. Returns True when a (re)registration actually happened.

The unchanged path is lock-free and cheap (build + compare one frozen config, a few µs), so callers routing by runtime topology can call this in front of every query::

await db.ensure_database(shard_id, {"primary": {"url": dsn}})
rows = await db.fetch_all(query, params,
                          target=DatabaseTarget(database=shard_id, role="read"))

A changed config (rotated host, credentials, pool settings) re-registers in place: old engines are disposed and the next query connects with the new settings — no restart needed, including password-only rotations.

register_database `async` ¶

register_database(
    name: str,
    config: DatabaseConfig | Mapping[str, Any],
    *,
    connect: bool | None = None,
) -> bool

Register (or replace) a named database at runtime.

For topologies discovered after :meth:start — e.g. shard DSNs resolved per tenant from a service registry — so callers do not need to maintain their own engine registries. config is a :class:DatabaseConfig or the same dict shape as one entry under the top-level databases mapping.

The database map is swapped copy-on-write, so concurrent readers always see a complete mapping. Replacing an existing name first disposes its engines and resets its limiter/breakers, so the new settings fully apply. Returns True when an existing entry was replaced.

connect forces (or suppresses) eager engine creation; the default follows primary.required when the facade is already started. Engine creation is lazy in SQLAlchemy, so this never blocks on the network unless the pool is warmed explicitly.

unregister_database `async` ¶

unregister_database(name: str) -> bool

Remove a dynamically (or statically) registered database and dispose its engines, limiter, and breakers. Returns False if name is unknown.

In-flight operations against the database finish on their checked-out connections (idle-only disposal); new calls raise DatabaseRoutingError.

database_scope `async` ¶

database_scope(
    name: str,
    config: DatabaseConfig | Mapping[str, Any],
    *,
    connect: bool | None = None,
) -> AsyncIterator[DatabaseTarget]

Register name for the duration of the block, then unregister it.

Yields a primary-routed :class:DatabaseTarget for convenience::

async with db.database_scope("tenant-42", {"primary": {"url": dsn}}) as target:
    rows = await db.fetch_all(query, params, target=target)

Scope this to a unit of work, never to a request. Engines/pools are created on first use inside the block and disposed on exit — wrapping every HTTP request in a scope recreates connections each time and defeats pooling entirely. Intended uses: tests, migrations, one-off tenant batch jobs, or an ad-hoc query against a shard the process does not normally serve. Long-lived services should call :meth:register_database once per discovered shard and keep it registered.

from_dict `classmethod` ¶

from_dict(
    data: Mapping[str, Any], *, name: str = "database"
) -> DatabaseConfig

Build one database config from a plain dict — the same shape as an entry under the top-level databases mapping. Used by dynamic registration (§22.4).

Dynamic Registration¶

Bootstrap dynamic-first¶

ensure_database — the per-request pattern¶

Explicit control¶

Guarantees¶

API reference¶

ensure_database async ¶

register_database async ¶

unregister_database async ¶

database_scope async ¶

from_dict classmethod ¶