pgcommon 0.1.0

Opinionated async PostgreSQL access layer for platform services — RLS tenancy, retries, and observability by default.

pgcommon

Opinionated async PostgreSQL access layer that enforces RLS tenancy, retries, and observability by making the safe path the default — and the unsafe path impossible by design.

Shared asyncpg-backed PostgreSQL library for platform services. Python port of platform-pgcommon (Go) with full feature parity, reimplemented idiomatically for asyncio. Published on PyPI (pip install pgcommon) — not deployed on its own.

Features: connection pooling · RLS GUC injection · a lightweight .sql migration runner · transaction helpers · slow-query logging · Prometheus metrics · OpenTelemetry query spans · PostgreSQL error helpers · health check · backoff retry with jitter · optional FastAPI glue.

See ARCHITECTURE.md for design decisions, layer diagrams, and — importantly — how this port's asyncio/contextvars model differs from the Go original's context.Context model.

---

Design principles

• Fail fast on misconfiguration — Pool.create pings the database at construction time. A wrong DSN, unreachable host, or oversized pool is a startup error, not a runtime surprise.
• Make safe usage the default — with_conn and run_in_tx acquire, inject GUCs, and release automatically. You cannot forget to release a connection or skip RLS injection when the defaults are wired correctly.
• Keep business logic free from database plumbing — services import pgcommon and work with an asyncpg connection. Retry backoff, slow-query logging, and metric recording are invisible to the caller.
• Centralise cross-cutting concerns — RLS GUC injection, retry strategy, observability hooks, and health checks live here, not scattered across every service. A single change propagates to all consumers on the next version bump.

---

Mental model — request flow

This diagram shows the life of a database call from the moment an HTTP request arrives to the moment PostgreSQL enforces Row-Level Security. Every box on this path is either owned by the library or governed by its configuration.

Think of pgcommon as a gatekeeper: every query must pass through it to gain tenancy, safety, and observability guarantees.

flowchart TD
    A[HTTP Request]
    B["Auth Middleware\nwith_validated_guc_set(GUCSet) via contextvars"]
    C[pgcommon.Pool]
    D[with_conn / run_in_tx]
    E["GUC Injection\napp.user_id · app.tenant_id · app.tenant_roles"]
    F["asyncpg Connection\nyour SQL runs here"]
    G[PgBouncer / RDS Proxy\nexternal infra layer — optional]
    H[PostgreSQL + RLS\ncurrent_setting scopes every query]

    A --> B --> C --> D --> E --> F --> G --> H

    style B fill:#fff3cd,stroke:#856404
    style C fill:#d1ecf1,stroke:#0c5460
    style D fill:#d1ecf1,stroke:#0c5460
    style E fill:#d1ecf1,stroke:#0c5460
    style G fill:#e2e3e5,stroke:#6c757d
    style H fill:#d4edda,stroke:#155724

Read top → bottom: each layer adds guarantees (identity → GUCs → retries → observability) before the query reaches PostgreSQL.

Detailed annotated view:

HTTP Request
    │
    ▼
Auth Middleware                        ← caller's responsibility
    │  extracts identity, builds a GUCSet(user_id, tenant_id, tenant_roles)
    │  stores it via pgcommon.with_validated_guc_set(guc_set) — a contextvars.ContextVar,
    │  scoped to the current asyncio Task (see ARCHITECTURE.md's Cancellation/contextvars notes)
    │
    ▼
pgcommon.Pool                          ← await Pool.create(Config(...))
    │  an asyncpg.Pool wrapper with added safety and observability
    │
    ▼
with_conn(fn) / run_in_tx(pool, fn)    ← the only safe entry points
    │  acquires a connection from the pool
    │  never exposes the raw asyncpg.Pool or Connection outside fn
    │
    ▼
guc_provider → inject_gucs             ← automatic; wired by Config.guc_provider
    │  sets app.user_id / app.tenant_id / app.tenant_roles
    │  on the connection before fn is called
    │  (session-scoped in standard mode; transaction-local in PgBouncer mode)
    │
    ▼
asyncpg Connection                     ← what fn receives
    │  execute your SQL here; RLS GUCs are already in place
    │
    ▼
PgBouncer / RDS Proxy    (external infra layer, optional)  ← transparent to the library
    │  set Config.pgbouncer_mode = True when present
    │
    ▼
PostgreSQL                             ← enforces Row-Level Security
    │  current_setting('app.tenant_id') scopes every query
    │
    ▼
    ├── SlowQueryTracer                ← side-channel; logs WARN if elapsed > threshold
    ├── OTelQueryRecorder              ← side-channel; emits a span per query
    └── Prometheus counters            ← pgcommon_query_total · pool_acquire_total · retry_total

Invariant: pgcommon.Pool is the only allowed entry point to PostgreSQL. Any code path that bypasses it silently disables tenancy, retries, and observability — and is considered a correctness bug.

Key invariants:

• with_conn/run_in_tx are the only paths to the database — there is no escape hatch to a raw pool or connection.
• GUC injection happens before fn is called, not inside it — your business logic never needs to know about RLS.
• Observability (metrics, traces, slow-query logs) is wired at pool construction time via pgcommon.metrics.init and Config.logger/Config.tracer — zero per-query code in callers.
• In PgBouncer mode, GUCs are transaction-local (is_local=true) and re-injected on every with_conn/run_in_tx call to survive connection multiplexing.

In standard mode, GUCs are re-injected on every connection acquire (cheap — one batched round trip), reused for every query on that acquire. In PgBouncer mode, GUCs are additionally wrapped in an implicit transaction so is_local=true settings clear themselves on commit/rollback.

Ownership summary:

• Service: auth middleware + business logic
• pgcommon: safety, GUC injection, retries, observability
• Infra: PgBouncer, PostgreSQL

---

Contents

• Design principles
• Mental model — request flow
• TL;DR
• When NOT to use this
• Why pgcommon instead of raw asyncpg?
• Installing
• Quick start
• Integration checklist
• Connection pool
• RLS GUC injection
• FastAPI integration
• Transactions
• Retry helpers
• Health check and load shedding
• PostgreSQL error helpers
• Slow-query logging
• Migrations
• Observability — Prometheus and OTel
• PgBouncer / RDS Proxy
• Failure model
• Performance considerations
• Common pitfalls
• Configuration reference
• Error reference
• Testing
• CI
• Versioning and releases
• Glossary
• License / ownership
• See also

---

TL;DR

• Use await Pool.create(config) to create a connection pool — never call asyncpg.create_pool directly.
• Always access the database through pool.with_conn(fn) or run_in_tx(pool, fn) — never hold a raw connection.
• Never reach for asyncpg directly — bypassing pgcommon.Pool silently disables GUC injection, retry logic, and all observability.
• Set Config.guc_provider = pgcommon.guc_set_from_context for multi-tenant services that use Row-Level Security.
• Use run_in_tx_with_retry / run_in_tx_with_retry_opts for write workflows that may encounter deadlocks or serialization failures.
• Call pgcommon.metrics.init(...) once at startup to enable Prometheus metrics and OTel spans.

---

When NOT to use this

This library is designed for long-running PostgreSQL-backed platform services. It may be more than you need if:

• Lightweight scripts or CLIs — a single asyncpg.connect() without a pool is simpler and has less startup overhead.
• Single-query services — if your service issues one query per process invocation (e.g. a Lambda with a cold-start pool), the pool's background maintenance tasks add unnecessary complexity.
• Non-PostgreSQL databases — the library is PostgreSQL-specific. MySQL, SQLite, and other backends are not supported.
• Read-only reporting jobs — if you don't need RLS GUC injection, retry logic, or migrations, a plain asyncpg.Pool is sufficient.
• Services that already own an asyncpg pool — do not wrap an existing pool in a second pool; inject the shared pgcommon.Pool as a dependency instead.

---

Why pgcommon instead of raw asyncpg?

Concern	Raw asyncpg.Pool	pgcommon.Pool
Connection lifecycle	Manual acquire()/release via async with; easy to leak connections on unusual control flow	with_conn/run_in_tx acquire and release automatically; leaks are structurally impossible
RLS GUC injection	Must SET LOCAL before every query; forgotten injection silently breaks multi-tenant isolation	guc_provider hook injects app.user_id, app.tenant_id, app.tenant_roles on every connection acquire — zero per-query code
Transaction commit/rollback	Manual async with conn.transaction(); easy to get isolation/read-only options wrong	run_in_tx commits on success, rolls back on exception and on cancellation
Retry on deadlock / serialization failure	Manual retry loop with backoff; most teams either skip it or implement it inconsistently	run_in_tx_with_retry/run_in_tx_with_retry_opts with exponential backoff and ±jitter, retrying only on SQLSTATE 40P01/40001
Prometheus metrics	Manual counter increments around every query	pgcommon_pool_acquire_total/pgcommon_slow_query_total/pgcommon_retry_total update automatically after pgmetrics.init; pgcommon_query_total/pgcommon_query_duration_seconds additionally require Config.tracer (see below)
OTel tracing	Manual span start/end per query	One db.query span per query with db.system, db.statement (redacted by default), net.peer.name, net.peer.port — wired at pool construction
Slow-query detection	Not built into asyncpg	SlowQueryTracer logs WARN with query, args_count, duration_ms, rate-limited to 1/s; Prometheus counter fires on every detection
PgBouncer / RDS Proxy compatibility	Statement-cache disabling and transaction-local GUCs must be configured and remembered	Config.pgbouncer_mode = True disables asyncpg's statement cache and switches GUC injection to is_local=true transaction scope automatically
Health check endpoint	Manual ping + stats aggregation	pool.health() returns healthy, acquired_conns, max_conns, utilization with a built-in 5-second deadline
Graceful shutdown	Manual wait-loop over in-flight connections	pool.drain_and_close() marks the pool closed, waits for in-flight callbacks, then force-closes on a deadline
Safety defaults	None enforced — consumers must remember every step	Default Config fields are safe; SQL is redacted from OTel by default; tenant IDs are not logged by default
Migration runner	A separate tool integration	pgcommon.migrate.Runner — advisory-locked, dirty-state-tracked, structured-logged per step

The single-sentence justification: every row above represents a bug we've seen in production services that used raw asyncpg — leaked connections, missing rollbacks, forgotten GUCs, silent RLS bypass, missing metrics. pgcommon makes the correct path the only path.

This is not an abstraction over asyncpg — it is a guardrail layer that enforces safety and correctness by construction.

---

Installing

Published on PyPI — the source repository stays private, but the released package is public (see VERSIONING.md § Publishing to PyPI for why).

pip install pgcommon
# or with uv:
uv add pgcommon

Optional extras:

pip install "pgcommon[fastapi]"    # FastAPI dependency + middleware glue
pip install "pgcommon[structlog]"  # structlog Logger adapter

# or with uv:
uv add "pgcommon[fastapi]"
uv add "pgcommon[structlog]"

Pin an exact version in production — see VERSIONING.md § Consume a release for the recommended pin styles.

Import only pgcommon.* (and, for FastAPI services, pgcommon.integrations.fastapi). Never import pgcommon._internal — it is enforced as private by lint rule and is not part of the supported API.

---

Quick start

import asyncio
import os

from pgcommon import Config, Pool, guc_set_from_context
from pgcommon.migrate import Runner


async def main() -> None:
    # 1. Run any pending schema migrations.
    runner = Runner(os.environ["DATABASE_URL"], "migrations")
    await runner.up()

    # 2. Open the connection pool.
    pool = await Pool.create(
        Config(
            dsn=os.environ["DATABASE_URL"],
            guc_provider=guc_set_from_context,  # automatic RLS GUC injection
        )
    )

    # 3. Use the pool.
    try:
        count = await pool.with_conn(lambda conn: conn.fetchval("SELECT count(*) FROM pg_tables"))
        print(count)
    finally:
        await pool.drain_and_close()


asyncio.run(main())

---

Integration checklist

Run through this before wiring the library into a new service. Each item links to the relevant section.

• Pool.create — create the pool via await Pool.create(config); never call asyncpg.create_pool directly (Connection pool)
• guc_provider — set guc_provider=pgcommon.guc_set_from_context on every multi-tenant service; omit only for internal tooling with no RLS (RLS GUC injection)
• with_conn / run_in_tx — wrap every DB call in one of these helpers; never reach into the pool's internals (Connection pool)
• pgcommon.metrics.init — call once at startup to register Prometheus counters and enable OTel spans (Observability)
• Retry helpers — use run_in_tx_with_retry/run_in_tx_with_retry_opts on write paths that run under repeatable_read or serializable isolation (Retry helpers)
• Health endpoint — wire pool.health() into your /healthz handler and shed load when utilization exceeds your threshold (Health check and load shedding)
• drain_and_close — call this in your shutdown path, not pool.close() (Connection pool)

---

Connection pool

Pool.create wraps asyncpg.create_pool and applies sensible defaults. A ping is performed at construction time so misconfiguration fails fast.

from datetime import timedelta

from pgcommon import Config, Pool

pool = await Pool.create(
    Config(
        dsn=os.environ["DATABASE_URL"],
        max_conns=20,
        min_conns=2,
        max_conn_lifetime=timedelta(hours=1),
        max_conn_idle_time=timedelta(minutes=30),
        slow_query_threshold=timedelta(milliseconds=300),
        logger=my_logger,  # implements the pgcommon.Logger protocol
        tracer=my_otel_tracer,  # an opentelemetry.trace.Tracer
    )
)

Concurrency

Pool is safe for concurrent use by multiple asyncio tasks on the same event loop. All helpers (with_conn, run_in_tx, run_in_tx_with_retry, health) are designed to be called from concurrent request handlers without additional synchronisation. The underlying asyncpg.Pool serialises connection acquisition internally; your code never needs a lock around pool calls.

This assumes single-event-loop use. Running multiple worker processes (e.g. Gunicorn/Uvicorn workers) is fine — each process gets its own Pool — but do not share one Pool instance across processes or manually spun-up event loops.

There is intentionally no public API to access the underlying asyncpg.Pool. This is a design choice to enforce invariants — bypassing the helpers would silently disable GUC injection, retry logic, and observability.

The terms "Pool helpers" and "entry points" used throughout this document both refer to with_conn and run_in_tx.

Acquiring connections

# Single connection — auto-released when fn returns.
result = await pool.with_conn(lambda conn: conn.fetchval("SELECT $1::int", 1))

Pool statistics

stat = pool.stat()
print(
    f"conns: {stat.acquired_size} acquired / {stat.idle_size} idle / {stat.size} total (max {stat.max_size})"
)
print(f"acquire count: {stat.acquire_count}  empty waits: {stat.empty_acquire_count}")

Stat fields: size, idle_size, min_size, max_size, acquired_size, acquire_count, acquire_duration (cumulative timedelta), empty_acquire_count, canceled_acquire_count.

Graceful shutdown

Always use drain_and_close — never close — when shutting down. close immediately terminates the pool, leaving in-flight with_conn callbacks mid-execution. drain_and_close marks the pool closed first (so new callers get PoolClosedError immediately), waits for in-flight callbacks to finish, then force-closes when the timeout is reached.

try:
    await pool.drain_and_close(timeout=30)
except Exception:
    logger.warning("pool drain timed out")
    # the pool is force-closed at this point; safe to exit

---

RLS GUC injection

PostgreSQL Row-Level Security policies read current_setting('app.user_id', true) etc. to scope rows. This library sets those GUCs automatically — no query needs SET LOCAL.

Injection lifecycle — when and how GUCs are set

Injection timing is critical: too early risks leaking GUCs across tenants; too late risks queries executing without tenant context.

The mode you run in determines when and how injection happens:

	Standard mode (pgbouncer_mode=False)	PgBouncer mode (pgbouncer_mode=True)
Trigger	Every connection acquisition (asyncpg's per-acquire setup hook)	Each with_conn/run_in_tx call, inside a wrapping transaction
Scope	Session-level (is_local=false) — persists on the connection until the next acquire re-injects it	Transaction-local (is_local=true) — automatically cleared on COMMIT/ROLLBACK
Cost	One batched round trip per acquire, zero per query after that	One extra round trip (implicit BEGIN + set_config) per with_conn/run_in_tx call
Risk if wrong mode	Session GUCs may leak to the next client in PgBouncer transaction pooling	Harmless but wasteful against a direct connection — is_local=true still works, just with unneeded overhead
When to use	Direct Postgres, PgBouncer session pooling, RDS direct	PgBouncer transaction pooling, RDS Proxy, any connection multiplexer

Standard mode timing:

pool.acquire() called
    │
    ▼
asyncpg's per-acquire `setup` hook fires ──── SELECT set_config('app.user_id', $1, false),
    │                                                 set_config('app.tenant_id', $2, false),
    │                                                 set_config('app.tenant_roles', $3, false)
    ▼
fn(conn) — GUCs already in place for the whole call
    │
    ▼
connection released — GUCs remain until the next acquire re-injects them

PgBouncer mode timing:

pool.acquire() called (no GUC injection here — pgbouncer_mode disables the acquire-time hook)
    │
    ▼
with_conn/run_in_tx opens an implicit transaction (or your own, for run_in_tx)
    │
    ▼
SELECT set_config('app.user_id', $1, true), set_config(..., true), set_config(..., true)
    │
    ▼
fn(conn) — GUCs in place for this transaction only
    │
    ▼
COMMIT / ROLLBACK — GUCs automatically cleared; connection safely returned to PgBouncer

Dependency: RLS policies must be written to read these GUCs (e.g. current_setting('app.tenant_id', true)) or GUC injection has no effect. Setting the GUC without a corresponding policy expression leaves rows unfiltered. Verify your policies in an integration test — a missing CREATE POLICY or a typo'd setting name silently returns all rows. See tests/integration/test_rls_guc_injection.py and tests/e2e/test_rls_pgbouncer_e2e.py for exactly this kind of test.

Wiring (recommended — contextvars-based)

from pgcommon import Config, GUCSet, Pool, guc_set_from_context, with_validated_guc_set

# 1. Wire once at startup:
pool = await Pool.create(Config(dsn=dsn, guc_provider=guc_set_from_context))


# 2. In ASGI middleware — store the GUCSet for the duration of the request.
#    with_validated_guc_set raises on an inconsistent identity (catch it and respond 401).
async def auth_middleware(request, call_next):
    claims = extract_claims(request)
    guc_set = GUCSet(
        user_id=claims.user_id, tenant_id=claims.tenant_id, tenant_roles=tuple(claims.roles)
    )
    try:
        with with_validated_guc_set(guc_set):
            return await call_next(request)
    except (InconsistentGUCSetError, InvalidTenantRoleError, EmptyTenantRoleError):
        return Response("invalid identity", status_code=401)


# 3. Every pool.with_conn / run_in_tx call downstream now injects GUCs automatically.

pgcommon.integrations.fastapi.GUCAuthMiddleware implements exactly this pattern — see FastAPI integration.

Manual injection

from pgcommon import GUCSet, inject_gucs


async def fn(conn):
    await inject_gucs(conn, GUCSet(user_id="user-123", tenant_id="tenant-456"))
    # queries now run under the injected GUC context
    return await conn.fetchval("SELECT id FROM orders LIMIT 1")


await pool.with_conn(fn)

inject_gucs always uses is_local=false (session-scoped), regardless of Config.pgbouncer_mode. Manual injection is for one-off cases; the pool's automatic injection respects the configured mode. See PgBouncer / RDS Proxy for the full PgBouncer configuration guide.

---

FastAPI integration

Optional — requires pgcommon[fastapi]. The core library never imports fastapi; pgcommon.integrations.fastapi is an explicit opt-in import.

from fastapi import FastAPI, Depends
from pgcommon import Config, GUCSet, guc_set_from_context
from pgcommon.integrations.fastapi import (
    GUCAuthMiddleware,
    create_lifespan,
    get_pool,
    healthz,
)

# guc_provider must still be set — GUCAuthMiddleware only stores the GUCSet on the
# contextvar; guc_set_from_context is what the pool reads it back with on every acquire.
config = Config(dsn=os.environ["DATABASE_URL"], guc_provider=guc_set_from_context)


def extract_identity(request) -> GUCSet | None:
    claims = decode_jwt(request.headers.get("authorization"))
    if claims is None:
        return None
    return GUCSet(user_id=claims.sub, tenant_id=claims.tenant_id, tenant_roles=tuple(claims.roles))


app = FastAPI(lifespan=create_lifespan(config))
app.add_middleware(GUCAuthMiddleware, identity_extractor=extract_identity)
app.add_api_route("/healthz", healthz(utilization_threshold=0.9), methods=["GET"])


@app.get("/orders")
async def list_orders(pool=Depends(get_pool)):
    return await pool.with_conn(lambda conn: conn.fetch("SELECT id, amount FROM orders"))

create_lifespan opens the Pool once at startup (stored on app.state.pool) and calls drain_and_close at shutdown — never reconstructed per request. get_pool is the corresponding Depends() provider.

contextvars and request isolation. Go threads an explicit context.Context through every call; Python instead relies on contextvars.ContextVar, which is copied per asyncio.Task at task-creation time. Each inbound ASGI request runs on its own task, so GUCAuthMiddleware's with_validated_guc_set(...) — entered around call_next(request) and exited (resetting the token) in a finally block — cannot leak into a concurrently-running request. The failure mode to avoid: never call the underlying ContextVar.set() without capturing and later reset()-ing its token in the same middleware invocation. See pgcommon/integrations/fastapi.py's module docstring and ARCHITECTURE.md for the full explanation.

---

Transactions

Basic transaction

from pgcommon import run_in_tx


async def fn(conn):
    await conn.execute("INSERT INTO orders (user_id) VALUES ($1)", user_id)
    await conn.execute("UPDATE inventory SET qty = qty - 1 WHERE id = $1", item_id)


await run_in_tx(pool, fn)
# or: await pool.run_in_tx(fn)

run_in_tx behaviour:

• fn returns normally → commit
• fn raises → rollback, original exception re-raised
• the surrounding task is cancelled while fn is running → rollback, asyncio.CancelledError propagates

Nested savepoints

from pgcommon import run_in_savepoint


async def fn(conn):
    await conn.execute("INSERT INTO orders (user_id) VALUES ($1)", user_id)

    # Optional audit — failure rolls back only to this savepoint; the outer tx is unaffected.
    async def audit(c):
        await c.execute("INSERT INTO audit_log (msg) VALUES ($1)", "order created")

    await run_in_savepoint(conn, "sp_audit", audit)


await run_in_tx(pool, fn)

Savepoint names must match ^[A-Za-z_][A-Za-z0-9_]*$; otherwise InvalidSavepointNameError is raised.

---

Retry helpers

⚠️ Retry does NOT make an operation safe — it re-runs it

The retry helpers re-execute your entire transaction function from the top. They are a concurrency-control tool for serialization failures and deadlocks — not a correctness guarantee. Three rules are non-negotiable:

1. Your fn must be idempotent. It will run 1 to max_attempts times. If running it twice produces two rows, two charges, or two emails, retry has corrupted your data. The transaction rolls back between attempts, so database writes are undone — but anything outside the transaction is not.

2. Side effects inside fn are dangerous. Any non-database action — payment charge, email send, webhook POST, queue publish, in-memory counter increment — may fire on attempt 1, then the DB transaction fails and retries, firing the side effect again. The database rollback cannot undo an HTTP call that already happened. Do all external I/O outside the transaction, or make it idempotent with a key.

3. Retries amplify load under contention. Exactly when the database is struggling (deadlocks, serialization conflicts), retry adds more attempts on top of the existing pressure. Always set jitter_fraction >= 0.2 and a bounded max_attempts (3-5). A retry storm can exhaust the connection pool and take down the database for everyone.

If you cannot satisfy rule 1, do not use the retry helpers — use a single run_in_tx and surface the error to the caller, who can decide whether replaying the whole request is safe.

Immediate retry

run_in_tx_with_retry retries on serialization failures (SQLSTATE 40001) and deadlocks (SQLSTATE 40P01). Non-retryable errors are raised immediately. Retries are immediate with no delay:

from pgcommon import run_in_tx_with_retry


async def fn(conn):
    await conn.execute(
        "UPDATE accounts SET balance = balance - $1 WHERE id = $2", amount, account_id
    )


await run_in_tx_with_retry(pool, fn, 5, isolation="serializable")

Exponential backoff with jitter

run_in_tx_with_retry_opts adds configurable backoff and jitter_fraction to spread retries under high concurrency. asyncio.CancelledError during the backoff sleep is honoured immediately:

from datetime import timedelta

from pgcommon import RetryOptions, run_in_tx_with_retry_opts


async def fn(conn):
    await conn.execute(
        "UPDATE accounts SET balance = balance - $1 WHERE id = $2", amount, account_id
    )


await run_in_tx_with_retry_opts(
    pool,
    fn,
    RetryOptions(
        max_attempts=5,
        initial_wait=timedelta(milliseconds=10),
        max_wait=timedelta(milliseconds=500),
        multiplier=2.0,
        jitter_fraction=0.25,  # ±25% per attempt — prevents thundering herd
    ),
    isolation="serializable",
)

Delay schedule per failed attempt:

delay = initial_wait * multiplier ** (attempt - 1)
delay = min(delay, max_wait)                     # when max_wait > 0
delay += random(-jitter_fraction, +jitter_fraction) * delay

RetryOptions field	Default	Purpose
max_attempts	3	Maximum attempts before giving up
initial_wait	timedelta(0) (immediate)	Delay before the 2nd attempt
max_wait	timedelta(0) (no cap)	Ceiling for exponential growth
multiplier	1.0 (constant wait)	Set 2.0 for standard doubling backoff
jitter_fraction	0.0 (no jitter)	Varies delay by ±fraction; range [0, 1]

When to use retry helpers

Use retry when the operation is idempotent and the error is transient:

• Financial transfers / balance updates — serializable reads prevent phantom reads; retrying is safe because the transaction either fully commits or fully rolls back.
• Idempotent inventory adjustments — UPDATE stock SET qty = qty - 1 WHERE id = $1 AND qty > 0 is safe to retry; the WHERE guard makes it a no-op if already decremented.
• Upserts with ON CONFLICT — the conflict resolution is deterministic; retrying on a serialization failure produces the correct final state.
• Audit log inserts — idempotency keys prevent duplicate rows on retry.

Do not use retry when the operation has observable side effects beyond the database:

• Non-idempotent writes — e.g. inserting a row without a uniqueness constraint creates duplicates on retry.
• Operations that trigger external calls inside the transaction function — the side effect may have already fired before the serialization failure was raised.
• User-facing mutations where the caller already handles retries — layering two retry loops amplifies request latency.

Suggested retry options (production)

Scenario	max_attempts	initial_wait	multiplier	jitter_fraction
High-throughput OLTP	3	10ms	2.0	0.25
Financial transactions	5	50ms	2.0	0.20
Background jobs	5	100ms	1.5	0.10

---

Health check and load shedding

pool.health() returns a HealthStatus snapshot with a liveness ping and current pool statistics. It never raises — a ping failure is reported as healthy=False.

/healthz endpoint

@app.get("/healthz")
async def healthz(pool: Pool = Depends(get_pool)):
    status = await pool.health()
    if not status.healthy:
        return Response("database unavailable", status_code=503)
    return {
        "conns": f"{status.acquired_conns}/{status.max_conns}",
        "utilization": status.utilization,
    }

(pgcommon.integrations.fastapi.healthz() builds exactly this handler, with utilization-based shedding included — see FastAPI integration.)

Utilization-based load shedding

async def db_health_middleware(request, call_next):
    status = await pool.health()

    # Fail fast when the database is unreachable.
    if not status.healthy:
        return Response("database unavailable", status_code=503)

    # Shed load when the pool is 90%+ saturated to protect the last connections
    # for in-flight requests — prevents cascading timeouts under traffic spikes.
    if status.utilization >= 0.9:
        return Response("service overloaded", status_code=503)

    return await call_next(request)

HealthStatus fields:

Field	Type	Description
healthy	bool	True when the database ping succeeded
total_conns	int	Acquired + idle
idle_conns	int	Waiting in the pool
acquired_conns	int	Currently held by callers
max_conns	int	Pool maximum (Config.max_conns)
utilization	float	acquired_conns / max_conns, in [0, 1]

pool.health() is a point-in-time snapshot — call it at the start of each request, not once globally. A 90% threshold is a reasonable starting point; tune it to your service's observed acquired_conns baseline.

---

PostgreSQL error helpers

Inspect asyncpg.PostgresError SQLSTATE codes without matching on strings:

from pgcommon import (
    constraint_name,
    is_check_violation,
    is_foreign_key_violation,
    is_not_null_violation,
    is_unique_violation,
)


async def create_user(pool, email):
    try:
        await pool.with_conn(
            lambda conn: conn.execute("INSERT INTO users (email) VALUES ($1)", email)
        )
    except Exception as exc:
        if is_unique_violation(exc):
            raise ValueError(
                f"email already registered (constraint: {constraint_name(exc)})"
            ) from exc
        if is_foreign_key_violation(exc):
            raise ValueError("referenced record does not exist") from exc
        if is_not_null_violation(exc):
            raise ValueError("required field is missing") from exc
        if is_check_violation(exc):
            raise ValueError(
                f"value failed validation (constraint: {constraint_name(exc)})"
            ) from exc
        raise

Helper	SQLSTATE	Trigger
is_unique_violation	23505	Duplicate value in a UNIQUE / PRIMARY KEY column
is_foreign_key_violation	23503	Referenced row does not exist
is_not_null_violation	23502	NULL inserted into NOT NULL column
is_check_violation	23514	Value failed a CHECK constraint
is_deadlock	40P01	Deadlock detected between transactions
is_serialization_failure	40001	SERIALIZABLE / REPEATABLE READ write conflict
constraint_name	—	Returns the offending constraint name, or ""

---

Slow-query logging

When Config.logger is set and Config.slow_query_threshold > 0, Pool.create attaches a SlowQueryTracer that logs any query exceeding the threshold at WARN level:

slow query query=SELECT args_count=3 duration_ms=312.4

from datetime import timedelta

pool = await Pool.create(
    Config(
        dsn=dsn,
        logger=logger,
        slow_query_threshold=timedelta(milliseconds=300),
        log_tenant_id=False,  # default — tenant IDs are sensitive; set True to opt in
    )
)

Rate limiting: consecutive slow-query log entries are throttled to at most one per second to prevent log flooding under a pathological slow-query pattern. The pgcommon_slow_query_total Prometheus counter is incremented on every slow query regardless of the rate limit.

Tenant correlation: by default tenant_id is omitted from log entries. Set Config.log_tenant_id = True when your logging pipeline is appropriately secured and you need to correlate slow queries to specific tenants.

Set slow_query_threshold=timedelta(0) or omit logger to disable. Construct a standalone SlowQueryTracer (from pgcommon.tracer) to attach it outside the pool.

SlowQueryTracer configuration methods:

Method	Default	Effect
set_log_tenant_id(bool)	False	Include tenant_id in log entries (privacy-sensitive — opt-in)
set_allow_full_statements(bool)	False	Log full SQL text (default: SQL verb only — prevents PII leakage)
set_slow_query_hook(callable)	None	Called on every slow-query detection regardless of log rate-limiting

---

Migrations

⚠️ Run migrations to completion before the application starts

Never start the application before migrations finish, and never run migrations concurrently with a running application instance. A service that boots against a half-migrated schema will fail with missing-column or missing-table errors — or worse, silently read stale data if a partially-applied migration left the schema inconsistent.

Correct deployment patterns (pick one):

Pattern	How	When to use
Init container	A Kubernetes initContainer runs pgcommon-migrate up and must exit 0 before the app container starts	Kubernetes deployments — the standard choice
Pre-boot job	A CI/CD pipeline step (or release job) runs migrations against the target DB before rolling out the new app version	ECS, Lambda, VM-based deploys, or any non-Kubernetes platform
Leader-only on boot	One designated instance runs runner.up() at startup; others wait for it to finish	Single-leader systems where an init container is not available

What NOT to do:

• ❌ Do not call runner.up() in every replica's startup path simultaneously. The runner takes a Postgres advisory lock so only one migration run proceeds at a time, but the losing replicas block on the lock and may time out their readiness probe. Worse, with rolling deploys the old app version can be running against the new schema during the window — design migrations to be backward-compatible (expand/contract pattern) if you cannot stop traffic.
• ❌ Do not assume the app and migrations share a deploy step. They are separate concerns: migrations change the schema, the app consumes it. Couple them in ordering (migrations first), not in process (same process, same time).

Backward-compatibility note: during a rolling deploy both schema versions are briefly live. Prefer additive migrations (add a nullable column, backfill, then later make it NOT NULL in a separate release) so the old app version keeps working until every replica is replaced. This is the expand/contract (a.k.a. parallel-change) pattern.

The pgcommon-migrate console script is built for exactly this — use it as your init container or pre-boot job entrypoint:

# In an init container or pre-deploy job:
DATABASE_URL=postgres://… pgcommon-migrate up

# Equivalently, runnable without the console script installed on PATH:
python -m pgcommon.migrate up

Or call the runner directly from a dedicated migration entrypoint (not your main service process):

from pgcommon.errors import MigrationDirtyError
from pgcommon.migrate import Runner

runner = Runner(os.environ["DATABASE_URL"], "migrations", logger=logger)  # logs each step

try:
    await runner.up()  # apply all pending migrations
except MigrationDirtyError:
    # A previous migration failed mid-run. Fix the migration and manually
    # clear the schema_migrations.dirty flag before retrying.
    raise

version, dirty = await runner.version()  # report current schema version
await runner.down(1)  # roll back one step
await runner.steps(2)  # apply exactly two steps forward (or -2 to roll back two)

Migration files follow the NNNN_description.up.sql / NNNN_description.down.sql naming convention (see migrations/ for the shipped example pair). Applied versions are tracked in a schema_migrations(version bigint, dirty boolean) table, created automatically if absent.

Custom migration source: pass any directory path as Runner's second argument — it is read fresh on every up/down/steps call, so no code changes are needed to add migration files.

---

Observability — Prometheus and OTel

Both Prometheus metrics and OTel tracing are optional. The pool, transactions, and GUC injection work without ever calling pgcommon.metrics.init or supplying Config.tracer.

Prometheus metrics

Call once at service startup (idempotent — first caller wins):

import pgcommon.metrics as pgmetrics

pgmetrics.init(os.environ["APP_NAME"], os.environ["BUILD_VERSION"])

For isolated test registries:

from prometheus_client import CollectorRegistry

pgmetrics.init_with_registry("test-svc", "v0.0.0", CollectorRegistry())

Registered metrics:

Metric	Type	Labels	Description
pgcommon_query_total	Counter	operation, status	Queries executed; operation is the SQL verb (SELECT, INSERT, UPDATE, DELETE, other, …) extracted from the query text — never the raw SQL
pgcommon_query_duration_seconds	Histogram	operation	Query duration by SQL verb
pgcommon_pool_acquire_total	Counter	status	Connection acquire attempts (success/error)
pgcommon_pool_acquire_duration_seconds	Histogram	—	Time waiting to acquire a connection; recorded for every attempt including failures
pgcommon_slow_query_total	Counter	—	Queries that exceeded slow_query_threshold; incremented on every detection regardless of log rate-limiting
pgcommon_retry_total	Counter	reason	Retried transaction attempts; reason is deadlock or serialization
pgcommon_retry_exhausted_total	Counter	reason	Retry sequences that exhausted all attempts without success; a rising value indicates persistent contention that retries are not resolving
pgcommon_build_info	Gauge	service, version	Always 1; a standard Prometheus build-info metric for correlating deploys

pgcommon_query_total / pgcommon_query_duration_seconds require Config.tracer to be set, not just pgmetrics.init(). Both are recorded inside OTelQueryRecorder, which Pool.create only constructs when Config.tracer is set (see OpenTelemetry query spans below) — it is wired with whatever Metrics handle pgmetrics.init() registered, if any, at the time the pool was created. If you want per-query Prometheus metrics without exporting traces anywhere, set Config.tracer to an OTel Tracer from a TracerProvider with no configured exporter — the spans are simply dropped, but the metric side effect still fires. Also note the OTel-metrics link is snapshotted once, at Pool.create time: call pgmetrics.init() before Pool.create, not after, or query_total/query_duration will never be recorded even with Config.tracer set.

OpenTelemetry query spans

from opentelemetry import trace

tracer = trace.get_tracer("my-service")

pool = await Pool.create(Config(dsn=dsn, tracer=tracer))

Each query produces a db.query span with attributes: db.system=postgresql, db.statement, db.user, net.peer.name, net.peer.port. Query errors are recorded on the span via span.record_exception + codes.Error.

SQL privacy: by default db.statement contains only the SQL verb (SELECT, INSERT, etc.) to prevent PII leakage to tracing backends. Set Config.allow_full_statements = True to include the full SQL — only do this when query text contains no sensitive data.

A note on how spans are produced. pgx's QueryTracer gets a pre-query and a post-query hook; asyncpg's closest analog, Connection.add_query_logger, fires only once — after the query has already completed. pgcommon.metrics.OTelQueryRecorder reconstructs a span with the correct start/end timestamps from the reported elapsed duration, and it still nests correctly under your request-level span because asyncio's scheduling of the logger callback preserves the contextvars context that OTel uses to track the "current span." See ARCHITECTURE.md for the detail.

Metrics and traces are correlated through the caller's active OTel context, the same way request-level spans nest child spans in any other OTel-instrumented Python code — no manual propagation needed. In a distributed trace viewer (Jaeger, Tempo, Honeycomb) the database spans appear nested under the service span.

---

PgBouncer / RDS Proxy

PgBouncer breaks connection stickiness — pgcommon compensates by re-injecting GUCs per transaction.

PgBouncer is infrastructure-owned (DB/platform team), not service-owned. Services only toggle compatibility via Config.pgbouncer_mode.

Set pgbouncer_mode=True when connecting through PgBouncer in transaction pooling mode (the default for RDS Proxy and most PgBouncer deployments):

pool = await Pool.create(
    Config(
        dsn=dsn,
        pgbouncer_mode=True,
        min_conns=0,  # let PgBouncer manage the backend pool size
        max_conns=20,  # match your PgBouncer pool_size for this db/user pair
        guc_provider=guc_set_from_context,
    )
)

pgbouncer_mode=True makes two changes:

1. Disables asyncpg's client-side statement cache (statement_cache_size=0). PgBouncer transaction pooling can hand a different backend session to the same client connection between transactions; a cached "prepared" statement from a previous backend is not guaranteed to still be valid.

2. Transaction-local GUCs — when guc_provider is set, GUC injection moves from the per-acquire hook (session-scoped) to the start of each with_conn/run_in_tx call using is_local=true. In transaction pooling mode the backend is returned to PgBouncer after each transaction; session-level GUCs would either be lost or leak to the next client. If guc_provider is not set, this change has no effect.

Leave pgbouncer_mode=False when connecting directly to RDS PostgreSQL without a pooler.

with_conn in PgBouncer mode: when guc_provider is set and pgbouncer_mode is True, the connection passed to fn is already inside an open transaction (used to scope GUCs as is_local=true). Do not call conn.transaction() inside fn in this mode — asyncpg treats a nested transaction as a SAVEPOINT, which is usually harmless but not what you want if you were expecting a fresh top-level transaction. Use run_in_tx instead if you need explicit transaction control.

---

Failure model

Failure	Behaviour
Wrong DSN / unreachable host	Pool.create raises ConnectionError immediately — the process should not start
Connection error during with_conn	The asyncpg error is raised to the caller; the connection is not returned to the idle pool
Retryable error (40001, 40P01)	run_in_tx_with_retry/run_in_tx_with_retry_opts retry up to max_attempts; non-retryable errors bypass retry and are raised unchanged
Non-retryable database error	Raised to the caller as-is; use is_unique_violation/is_foreign_key_violation/etc. to distinguish error classes
Cancellation (asyncio.CancelledError)	Aborts connection acquisition, in-flight queries, and the retry sleep — propagates immediately
Pool exhaustion	with_conn blocks until a connection is available
pool.close() / drain_and_close() called	All subsequent with_conn/ping calls raise PoolClosedError immediately; health() never raises — it catches the error and reports healthy=False
Slow query	Logged at WARN when execution exceeds slow_query_threshold; the query continues unless cancelled — logging is observability only, not a circuit breaker

Errors from asyncpg are never swallowed. Every path either propagates the original exception or wraps it with additional context via exception chaining (raise ... from exc).

---

Performance considerations

Overhead

pgcommon is a thin wrapper over asyncpg. The added cost per query is dominated by network and PostgreSQL execution time, not by the library:

Mechanism	When it runs	Approximate cost	Notes
GUC injection (standard mode)	Once per connection acquire, not per query	One batched 3-set_config round trip per acquire	Zero on subsequent queries within the same with_conn/run_in_tx call
GUC injection (PgBouncer mode)	Once per with_conn/run_in_tx	One inline set_config round trip per call	Higher than standard mode by design — required for pooler safety
Metrics recording	Every query	O(µs) — atomic counter + histogram observe, in-process	No allocation on the hot path; no network
OTel span	Every query (only if Config.tracer set)	O(µs) — span start/end, in-process	Export is async/batched by the OTel SDK, off the query path
Slow-query check	Every query	O(µs) — one time comparison + conditional log call	Logging only fires above threshold, rate-limited to 1/s
Retry logic	Only on 40001/40P01 failure	Zero on the success path	Backoff sleep is the only added latency, and only between failed attempts

For most workloads the library overhead is well under 1% of total query latency. A single-row indexed SELECT over a local network is ~100-500µs, and the per-query in-process work above is a few microseconds. The one cost worth budgeting for is PgBouncer-mode GUC injection, which adds one extra round trip per with_conn call.

What this means in practice:

• Do not micro-optimise by bypassing the pool — the overhead you would save is dwarfed by the correctness guarantees you would lose (see Common pitfalls).
• Metrics and tracing are not free, but they are cheap — both are optional and in-process; the network cost is the exporter's, not the query's.
• If GUC injection cost matters, prefer standard (session) mode with connection reuse over PgBouncer mode, where the architecture allows it.

Tuning

• Avoid blocking inside with_conn callbacks. A connection is held for the full duration of the callback. Performing non-database work inside it — HTTP calls, file I/O, expensive computation — reduces pool availability for other tasks. Fetch data, release the connection, then do the heavy lifting outside.
• Avoid long-running transactions. A transaction holds a connection for its entire duration. Long transactions starve the pool, delay VACUUM, and increase deadlock probability.
• Prefer with_conn over any lower-level access. There is no sanctioned way to acquire a raw connection and forget to release it — with_conn handles acquisition and release automatically.
• Use PgBouncer for high concurrency. asyncpg.Pool holds physical backend connections open. At high traffic the backend connection count can exceed PostgreSQL's max_connections. PgBouncer in transaction pooling mode multiplexes many application connections onto fewer backend connections.
• Set max_conns to match your service's concurrency, not "as high as possible". A good starting point is max_conns = concurrent_query_tasks * 1.2. Monitor pgcommon_pool_acquire_duration_seconds and Stat.empty_acquire_count to tune.
• Tune slow_query_threshold for your workload. The default 200ms is appropriate for transactional OLTP; analytical/reporting endpoints may legitimately take longer.

---

Common pitfalls

---

❌ Bypassing Pool helpers

Any code that reaches into the underlying asyncpg.Pool directly skips GUC injection, the retry safety net, slow-query logging, and all metrics. This is the most dangerous mistake because it silently disables multi-tenant isolation. There is no sanctioned way to do this — Pool never exposes the underlying asyncpg.Pool — but do not work around that by constructing your own asyncpg.create_pool alongside pgcommon.Pool.

# ❌ WRONG: a second, unmanaged asyncpg pool bypasses everything
raw_pool = await asyncpg.create_pool(dsn=dsn)
async with raw_pool.acquire() as conn:
    await conn.fetch("SELECT * FROM orders")  # GUCs not injected — RLS sees no tenant context

# ✅ CORRECT: GUCs injected automatically, connection released on return
await pool.with_conn(lambda conn: conn.fetch("SELECT * FROM orders"))

---

❌ Forgetting guc_provider on a multi-tenant service

If Config.guc_provider is None, GUCs are never injected and RLS policies fall through to their own default — often returning all rows or no rows without an error. There is no error message; the data is just wrong.

# ❌ WRONG: guc_provider is None — every query runs without tenant context
pool = await Pool.create(Config(dsn=dsn))

# ✅ CORRECT: wire once at startup; injection is automatic everywhere
pool = await Pool.create(Config(dsn=dsn, guc_provider=guc_set_from_context))

Always verify your RLS policy in an integration test — a missing CREATE POLICY or a typo in current_setting('app.tenant_id', true) silently returns all rows.

---

❌ Transactions without retry on serializable workloads

Serialization failures (40001) and deadlocks (40P01) are normal at higher concurrency. A single attempt that drops the error leaves callers with a permanent failure on what should have been a transient one.

# ❌ WRONG: single attempt; deadlocks and serialization failures surface as hard errors
await run_in_tx(pool, fn, isolation="serializable")

# ✅ CORRECT: retries automatically on 40001 / 40P01; non-retryable errors surface immediately
await run_in_tx_with_retry_opts(
    pool,
    fn,
    RetryOptions(
        max_attempts=5, initial_wait=timedelta(milliseconds=10), multiplier=2.0, jitter_fraction=0.2
    ),
    isolation="serializable",
)

---

❌ Opening your own transaction inside with_conn in PgBouncer mode

When pgbouncer_mode=True and a GUCSet is active, with_conn already wraps fn in a transaction to scope the injected GUCs. Calling conn.transaction() again inside fn creates a nested SAVEPOINT, not the fresh top-level transaction you may be expecting.

# ❌ WRONG: conn is already inside pgcommon's GUC-scoping transaction
async def fn(conn):
    async with conn.transaction():  # this is a SAVEPOINT, not a new top-level transaction
        await conn.execute(...)


await pool.with_conn(fn)

# ✅ CORRECT: use run_in_tx for explicit transaction control
await run_in_tx(pool, fn)

---

❌ pool.close() during graceful shutdown

pool.close() is not graceful — it immediately terminates the pool. In-flight with_conn callbacks will error mid-execution, leaving transactions in an unknown state.

# ❌ WRONG: kills in-flight connections immediately
pool.close()

# ✅ CORRECT: waits for in-flight callbacks to finish, then force-closes
try:
    await pool.drain_and_close(timeout=30)
except Exception:
    logger.warning("pool drain timed out")
    # pool is force-closed at this point; safe to exit

---

❌ Ignoring PoolClosedError

After pool.close() or drain_and_close(), every subsequent with_conn/ping call raises PoolClosedError (health() is the one exception — it never raises, it just reports healthy=False). Reconnect loops that do not check for this stop condition spin indefinitely.

# ❌ WRONG: blind retry loop never terminates after shutdown
while True:
    try:
        await pool.ping()
        break
    except Exception:
        await asyncio.sleep(1)  # spins forever if the pool is closed

# ✅ CORRECT: break on pool closed; retry only on transient connectivity errors
while True:
    try:
        await pool.ping()
        break
    except PoolClosedError:
        return  # pool is shutting down — stop retrying
    except Exception:
        await asyncio.sleep(1)

---

❌ Long transactions blocking VACUUM

PostgreSQL cannot reclaim dead rows while a transaction holds an older snapshot. One stuck long-running transaction in a high-write table causes table bloat and slow queries for every other session.

# ❌ WRONG: no external I/O boundary — a slow external call holds the snapshot indefinitely
async def fn(conn):
    rows = await conn.fetch("SELECT ...")
    result = await call_external_api()  # can block for seconds or minutes
    await conn.execute("UPDATE ... ", result.id)


await run_in_tx(pool, fn)

# ✅ CORRECT: do external I/O outside the transaction; keep the transaction short
result = await call_external_api()


async def fn(conn):
    await conn.execute("INSERT INTO events ... ", result.id)


await run_in_tx(pool, fn)

---

❌ Double-wrapping Pool

Passing a Pool into a constructor that calls Pool.create again creates two pools over the same DSN, doubling background maintenance tasks and connection overhead.

# ❌ WRONG: creates a second pool over the same DSN
class OrderRepo:
    def __init__(self, pool: Pool):
        self._inner_pool = None  # will be set by an async factory calling Pool.create again — wrong

    @classmethod
    async def create(cls, pool: Pool) -> "OrderRepo":
        self = cls(pool)
        self._inner_pool = await Pool.create(Config(dsn=os.environ["DATABASE_URL"]))  # wrong
        return self

# ✅ CORRECT: inject the shared pool directly
class OrderRepo:
    def __init__(self, pool: Pool) -> None:
        self._pool = pool

---

Configuration reference

Config field	Env var	Default	Notes
dsn	DATABASE_URL	—	Required. Full connection string
max_conns	PG_MAX_CONNS	10	Pool maximum connections
min_conns	PG_MIN_CONNS	2	Minimum idle connections
max_conn_lifetime	PG_MAX_CONN_LIFETIME	1h	Max age before recycling (whole-pool generation recycle — see ARCHITECTURE.md)
max_conn_lifetime_jitter	PG_MAX_CONN_LIFETIME_JITTER	0	Random spread on the recycle interval, to prevent thundering-herd recycling across service instances
max_conn_idle_time	PG_MAX_CONN_IDLE_TIME	30m	Max idle time before closing
health_check_period	PG_HEALTH_CHECK_PERIOD	1m	Interval for the background liveness probe
slow_query_threshold	PG_SLOW_QUERY_THRESHOLD	200ms	Log queries exceeding this duration
logger	—	None	Implements the pgcommon.Logger protocol — use StdlibLogger or StructlogLogger
tracer	—	None	An opentelemetry.trace.Tracer
guc_provider	—	None	`Callable[[], GUCSet
pgbouncer_mode	PG_BOUNCER_MODE	False	Enable for PgBouncer / RDS Proxy transaction pooling
log_tenant_id	—	False	Include tenant_id in slow-query log entries — opt-in only; tenant IDs are sensitive
allow_full_statements	—	False	Include full SQL in OTel span db.statement; default omits SQL to prevent PII leakage

Suggested defaults (production)

Setting	Recommended value	Rationale
max_conns	10-20 per service instance	Match your service's peak concurrent DB tasks; monitor Stat.empty_acquire_count to tune
min_conns	2	Keeps a warm connection ready without over-committing backends
slow_query_threshold	200-500ms	200ms for tight OLTP SLAs; 500ms for mixed-workload services
RetryOptions.max_attempts	3-5	3 for low-contention workloads; 5 for high-concurrency financial writes
max_conn_lifetime	1h	Prevents stale connections after RDS failovers or network resets
max_conn_idle_time	30m	Reclaims idle connections under low traffic without full pool teardown

For services deployed behind PgBouncer or RDS Proxy set max_conns to match pool_size for the relevant db/user pair in the PgBouncer config — there is no benefit to holding more application-side connections than PgBouncer can route to backends.

Individual PG_* variables used to build the DSN when DATABASE_URL is unset:

Env var	Default
PG_HOST	localhost
PG_PORT	5432
PG_USER	—
PG_PASSWORD	—
PG_DBNAME	—
PG_SSLMODE	require

PG_SSLMODE and PG_BOUNCER_MODE are not gated behind DATABASE_URL being unset — both are read unconditionally: PG_BOUNCER_MODE always sets Config.pgbouncer_mode, and PG_SSLMODE is always checked for the insecure-sslmode warning below, even when the DSN itself comes from DATABASE_URL (in which case PG_SSLMODE overrides checking DATABASE_URL's own sslmode= query parameter for that warning, but does not change the DSN itself).

Env var	Default
PG_BOUNCER_MODE	false

Copy .env-example to .env for local development (make setup).

Security warning: config_from_env() emits a ConfigWarning whenever an insecure sslmode (disable, allow, or prefer) is detected — whether set via PG_SSLMODE or embedded in DATABASE_URL. Log all warnings at startup so accidental TLS disablement is visible:

config, warnings = config_from_env()
for w in warnings:
    logger.warning("config warning", key=w.key, reason=w.reason)

---

Error reference

Error	Raised when
DSNRequiredError	Config.dsn is empty
PoolClosedError	with_conn or ping called after close()/drain_and_close() (health() never raises — see Health check and load shedding)
InvalidMaxConnsError	Config.max_conns is negative
InvalidMinConnsError	Config.min_conns is negative
MinConnsExceedsMaxError	Config.min_conns > Config.max_conns
InvalidSavepointNameError	Savepoint name is empty or fails ^[A-Za-z_][A-Za-z0-9_]*$
InconsistentGUCSetError	GUCSet.user_id is set but tenant_id is empty
InvalidTenantRoleError	A GUCSet.tenant_roles entry contains a comma (the GUC separator)
EmptyTenantRoleError	A GUCSet.tenant_roles entry is an empty string
MigrationDirtyError	Schema is in a dirty state from a previously failed migration run
NoMigrationsFoundError	The migration source directory has no matching *.up.sql/*.down.sql files

All sentinel errors subclass PgCommonError — check with isinstance(exc, PgCommonError), the Python equivalent of Go's errors.Is against the base error type. All are importable directly from pgcommon.

---

Testing

make test-unit   # unit tests — no Docker required
make test-int    # integration tests — spins up a real Postgres container (testcontainers)
make test-e2e    # e2e tests — spins up Postgres + PgBouncer containers
make cover       # full suite + HTML coverage report (threshold: ≥97%)

Skip Docker-dependent tests:

uv run pytest -m "not integration and not e2e"

Run a single test by name:

uv run pytest tests/unit/test_pool.py -k test_with_conn_raises_when_closed -v
uv run pytest tests/integration -k test_gucset_scopes_queries_to_own_tenant -v -m integration

Integration and e2e tests use testcontainers-python — Docker must be running locally. Unit tests have no external dependencies and always run in CI without Docker.

---

CI

The main pipeline (.github/workflows/ci.yml) runs on every push and pull request to main. validate and coverage/build/docker form one sequential chain; lint-dockerfile, build-image (and its dependents trivy-scan/smoke-tests) run independently and in parallel with that chain; pr-summary waits on everything and only runs for pull requests.

Job	Step	Gate
validate → Quality	ruff format --check, ruff check, mypy --strict, pip-audit (calls the reusable quality.yml)	✅ blocks coverage
validate → Test	Full test suite (unit + integration + e2e via testcontainers), ≥97% coverage gate (calls the reusable test.yml)	✅ blocks coverage
coverage	Re-runs the full suite across a Python 3.12/3.13/3.14 matrix; uploads the coverage report as an artifact	✅ blocks build
build	Build wheel/sdist; twine check; verify the pgcommon-migrate console script installs and runs --help	✅ blocks docker
lint-dockerfile	hadolint against the Dockerfile	— (independent)
build-image	Cached docker buildx build of the migrate-CLI image, no push	— (independent; blocks trivy-scan/smoke-tests)
trivy-scan	Trivy CVE scan of the built image (HIGH/CRITICAL, ignore-unfixed)	—
smoke-tests	Runs the built image: default CMD prints usage; a real up invocation against a bad migration source fails cleanly	—
docker	Push the migrate-CLI image to GHCR — push events only, not pull requests	—
pr-summary	Posts a single PR comment summarizing all of the above — pull requests only	—

.github/workflows/release.yml is a separate pipeline triggered by v* tags: it reruns the same validate → coverage → build chain at the tagged commit, then publishes to PyPI (Trusted Publishing) and pushes the versioned image to GHCR, before creating the GitHub Release.

Run the full local gate before pushing:

make ci   # format-check + lint + typecheck + full test suite + build

make ci covers the validate/coverage/build chain; the Docker-specific jobs (lint-dockerfile, build-image, trivy-scan, smoke-tests) have their own make targets — see the Makefile or make help.

---

Versioning and releases

See VERSIONING.md for the full release process, the public-API surface SemVer applies to, and how the Trusted Publishing (OIDC) flow to PyPI is set up — see VERSIONING.md § Publishing to PyPI. Summary:

Version bump	When
MAJOR	Breaking change in the public API (pgcommon.*)
MINOR	New backward-compatible capability
PATCH	Bug fix, performance improvement, documentation correction

# Tag and push triggers the release workflow automatically.
git tag -a v1.0.0 -m "v1.0.0"
git push origin v1.0.0

---

Glossary

Term	Meaning in this library
GUC	PostgreSQL runtime parameter set via set_config/SET. Used to pass app.user_id, app.tenant_id, and app.tenant_roles into every connection so RLS policies can read them with current_setting(...).
GUCSet	The frozen dataclass (pgcommon.GUCSet) that carries user_id, tenant_id, and tenant_roles through the request's contextvars.ContextVar. Stored with with_guc_set/with_validated_guc_set, retrieved with guc_set_from_context.
RLS	PostgreSQL Row-Level Security — policies attached to tables that filter or reject rows based on current_setting(...) values. Enforced server-side; cannot be bypassed from the application.
PgBouncer mode	Config.pgbouncer_mode = True. Switches GUC injection from session-level (per-acquire) to transaction-local (is_local=true on every with_conn/run_in_tx), required when a connection pooler reuses backend connections across sessions.
DSN	Data Source Name — the full Postgres connection string, e.g. postgres://user:pass@host:5432/db?sslmode=require. Passed as Config.dsn or read from DATABASE_URL.
Tx	A database transaction. Started and committed/rolled back automatically by run_in_tx; never open one manually inside a with_conn callback in PgBouncer mode.
Savepoint	A named marker inside a transaction that can be rolled back to without aborting the outer transaction. Created by run_in_savepoint.
SQLSTATE	PostgreSQL's five-character error code (e.g. 40001 = serialization failure, 23505 = unique violation). Checked by the is_* helpers in pgcommon.

---

License / ownership

BCBP Solutions FZC LLC — internal platform shared library.

---

See also

Document	Description
ARCHITECTURE.md	Design decisions, layer diagrams, and the asyncio/contextvars translation notes
docs/MigrationGuide.md	Concept-by-concept mapping from the Go platform-pgcommon to this port
docs/adr/	One-decision-per-file records for the riskiest translation choices (GUC propagation, cancellation/rollback safety, GUC injection timing, migration runner design)
VERSIONING.md	SemVer rules, supported versions, release process
CHANGELOG.md	Per-version changes
CONTRIBUTING.md	Development setup, adding middleware, PR checklist
SECURITY.md	Vulnerability reporting, trust model, supported versions

---

All services must treat pgcommon.Pool as the single entry point for database access to preserve the guarantees of GUC-based tenancy, consistent retry behaviour, and uniform observability.