Skip to content

Connection management & health

dorm.db.connection.health_check(alias: str = 'default', timeout: float = 5.0, deep: bool = False) -> dict[str, Any]

Override-friendly version of the basic :func:health_check.

See :func:_health_check_basic for the always-on probe. Pass deep=True to additionally include :func:pool_stats so the same endpoint can serve both readiness and observability.

dorm.db.connection.ahealth_check(alias: str = 'default', timeout: float = 5.0, deep: bool = False) -> dict[str, Any] async

Async counterpart of :func:health_check.

dorm.db.connection.pool_stats(alias: str = 'default') -> dict[str, Any]

Return live pool statistics for alias.

Returned keys, when available:

  • alias — the database alias.
  • vendor — "postgresql" or "sqlite".
  • has_pool — whether this backend has a real connection pool.
  • For PostgreSQL with an open pool: pool_min, pool_max, pool_size (currently open connections), pool_available (idle / not in use), requests_waiting, requests_num (total checkouts), usage_ms, connections_ms and the rest of psycopg's get_stats() <https://www.psycopg.org/psycopg3/docs/advanced/pool.html>_ output, all under the same key names.
  • For SQLite (no pool): a minimal dict with the in-flight atomic-block depth so dashboards have something to graph.

Use this in a Prometheus / OpenTelemetry exporter or a debug endpoint. Never raises — returns {"status": "uninitialised"} if the alias has no live connection yet (calling this in a healthz handler before the first query is fine).

dorm.db.connection.get_connection(alias: str = 'default')

dorm.db.connection.get_async_connection(alias: str = 'default')

dorm.db.connection.close_all()

dorm.db.connection.close_all_async() async

dorm.db.connection.router_db_for_read(model, *, default: str = 'default', **hints) -> str

Consult settings.DATABASE_ROUTERS for the alias to use when reading rows of model. First router that returns a truthy string wins; otherwise default.

When a write through :func:router_db_for_write happened on the current context within the last settings.READ_AFTER_WRITE_WINDOW seconds (default 3.0), the router is bypassed and default is returned — so the request that wrote the row sees its own change instead of a stale replica snapshot. Pass sticky=False in hints to opt out of this behaviour for a specific call (analytics queries that explicitly want the replica even right after a write).

dorm.db.connection.router_db_for_write(model, *, default: str = 'default', **hints) -> str

Mirror of :func:router_db_for_read for writes. Records the write so subsequent reads stay sticky to the primary for the configured window.

dorm.db.utils.with_transient_retry(func, *, in_transaction: bool = False, attempts: int | None = None, backoff: float | None = None)

Run func() with simple exponential-backoff retry on transient DB errors. Skips retries while inside a transaction (would re-apply already-committed work).

dorm.db.utils.awith_transient_retry(coro_factory, *, in_transaction: bool = False, attempts: int | None = None, backoff: float | None = None) async

Async counterpart of :func:with_transient_retry. coro_factory is a 0-arg callable that returns a fresh coroutine on each retry — coroutines can only be awaited once.

health_check(deep=True) — combined readiness + observability

Both health_check and ahealth_check accept a deep=True flag that adds the live pool snapshot under the pool key — handy when the same /healthz endpoint must serve both readiness probes and observability scrapers:

import dorm

@app.get("/healthz")
async def healthz():
    return await dorm.ahealth_check(deep=True)
# {
#   "status": "ok", "alias": "default", "elapsed_ms": 0.42,
#   "pool": {
#     "alias": "default", "vendor": "postgresql", "has_pool": True,
#     "pool_min": 1, "pool_max": 10,
#     "pool_size": 7, "pool_available": 4, "requests_waiting": 0,
#     "requests_num": 18234, "usage_ms": 412.3, "connections_ms": 1.1,
#   }
# }

pool_stats(alias) is also exposed standalone for Prometheus / OTel exporters that want only the pool view:

from dorm import pool_stats
metrics.gauge("db.pool.in_use", pool_stats("default")["pool_size"])

pool_stats never raises — for a never-used alias it returns {"alias": ..., "status": "uninitialised"}, so calling it from a healthz handler before the first query is safe.