Metadata-Version: 2.4
Name: pgcommon
Version: 0.1.0
Summary: Opinionated async PostgreSQL access layer for platform services — RLS tenancy, retries, and observability by default.
Project-URL: Homepage, https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
Project-URL: Repository, https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
Project-URL: Changelog, https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/blob/main/CHANGELOG.md
Author: BCBP Solutions FZC LLC
License: Apache-2.0
License-File: LICENSE
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: asyncpg>=0.29
Requires-Dist: opentelemetry-api>=1.24
Requires-Dist: prometheus-client>=0.20
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == 'fastapi'
Provides-Extra: structlog
Requires-Dist: structlog>=24.1; extra == 'structlog'
Description-Content-Type: text/markdown

# 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`](https://github.com/BCBP-SOLUTIONS-FZC-LLC/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](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.

```mermaid
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](#design-principles)
- [Mental model — request flow](#mental-model--request-flow)
- [TL;DR](#tldr)
- [When NOT to use this](#when-not-to-use-this)
- [Why pgcommon instead of raw asyncpg?](#why-pgcommon-instead-of-raw-asyncpg)
- [Installing](#installing)
- [Quick start](#quick-start)
- [Integration checklist](#integration-checklist)
- [Connection pool](#connection-pool)
- [RLS GUC injection](#rls-guc-injection)
- [FastAPI integration](#fastapi-integration)
- [Transactions](#transactions)
- [Retry helpers](#retry-helpers)
- [Health check and load shedding](#health-check-and-load-shedding)
- [PostgreSQL error helpers](#postgresql-error-helpers)
- [Slow-query logging](#slow-query-logging)
- [Migrations](#migrations)
- [Observability — Prometheus and OTel](#observability--prometheus-and-otel)
- [PgBouncer / RDS Proxy](#pgbouncer--rds-proxy)
- [Failure model](#failure-model)
- [Performance considerations](#performance-considerations)
- [Common pitfalls](#common-pitfalls)
- [Configuration reference](#configuration-reference)
- [Error reference](#error-reference)
- [Testing](#testing)
- [CI](#ci)
- [Versioning and releases](#versioning-and-releases)
- [Glossary](#glossary)
- [License / ownership](#license--ownership)
- [See also](#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](https://pypi.org/project/pgcommon/) — the source repository stays
private, but the released package is public (see [VERSIONING.md § Publishing to
PyPI](VERSIONING.md#publishing-to-pypi) for why).

```bash
pip install pgcommon
# or with uv:
uv add pgcommon
```

Optional extras:

```bash
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](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

```python
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](#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](#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](#connection-pool))
- [ ] **`pgcommon.metrics.init`** — call once at startup to register Prometheus counters and enable OTel spans ([Observability](#observability--prometheus-and-otel))
- [ ] **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](#retry-helpers))
- [ ] **Health endpoint** — wire `pool.health()` into your `/healthz` handler and shed load when `utilization` exceeds your threshold ([Health check and load shedding](#health-check-and-load-shedding))
- [ ] **`drain_and_close`** — call this in your shutdown path, not `pool.close()` ([Connection pool](#connection-pool))

---

## Connection pool

`Pool.create` wraps [`asyncpg.create_pool`](https://magicstack.github.io/asyncpg/current/api/index.html#connection-pools) and applies sensible defaults. A `ping` is performed at construction time so misconfiguration fails fast.

```python
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

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

### Pool statistics

```python
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.

```python
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](tests/integration/test_rls_guc_injection.py) and [tests/e2e/test_rls_pgbouncer_e2e.py](tests/e2e/test_rls_pgbouncer_e2e.py) for exactly this kind of test.

### Wiring (recommended — contextvars-based)

```python
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](#fastapi-integration).

### Manual injection

```python
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](#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.

```python
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

```python
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

```python
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:

```python
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:

```python
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

```python
@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](#fastapi-integration).)

### Utilization-based load shedding

```python
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:

```python
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
```

```python
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:

```bash
# 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):

```python
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/`](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):

```python
import pgcommon.metrics as pgmetrics

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

For isolated test registries:

```python
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](#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

```python
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):

```python
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](#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`.

```python
# ❌ 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
```

```python
# ✅ 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.

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

```python
# ✅ 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.

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

```python
# ✅ 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.

```python
# ❌ 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)
```

```python
# ✅ 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.

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

```python
# ✅ 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.

```python
# ❌ 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
```

```python
# ✅ 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.

```python
# ❌ 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)
```

```python
# ✅ 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.

```python
# ❌ 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
```

```python
# ✅ 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 | None]` — use `guc_set_from_context` |
| `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:
>
> ```python
> 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](#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

```bash
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:

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

Run a single test by name:

```bash
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:

```bash
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](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](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 |

```bash
# 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](./ARCHITECTURE.md) | Design decisions, layer diagrams, and the asyncio/contextvars translation notes |
| [docs/MigrationGuide.md](./docs/MigrationGuide.md) | Concept-by-concept mapping from the Go `platform-pgcommon` to this port |
| [docs/adr/](./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](./VERSIONING.md) | SemVer rules, supported versions, release process |
| [CHANGELOG.md](./CHANGELOG.md) | Per-version changes |
| [CONTRIBUTING.md](./CONTRIBUTING.md) | Development setup, adding middleware, PR checklist |
| [SECURITY.md](./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.
