Metadata-Version: 2.5
Name: pgcommon
Version: 0.3.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.32,>=0.30
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.

The platform's **PostgreSQL access layer for Python services**: an `asyncio` library on top of asyncpg that every Python platform service imports to reach its database. It is the Python port of [`platform-pgcommon`](https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon) (Go) with full feature parity — currently tracking Go **v2.0.1** — reimplemented idiomatically for `asyncio`. It owns the database access boundary: the connection pool, Row-Level Security identity injection (`app.user_id` / `app.tenant_id` / `app.tenant_roles`), transactions with deadlock and serialization retry, SQLSTATE classification, the migration runner, and the database-client telemetry every service emits (Tier 1 `platform_db_*` metrics, OTel spans and a slow-query log). The few places where asyncpg cannot match Go are listed in [ARCHITECTURE.md § Divergences from Go](ARCHITECTURE.md#divergences-from-go).

**Repository:** `github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py` (private)

**Package:** `pgcommon` 0.3.0 ([release status](#versioning-and-releases): released 2026-10-15, on [PyPI](https://pypi.org/project/pgcommon/)) · Apache-2.0 · Python ≥ 3.12 (CI runs 3.12, 3.13 and 3.14) · consumed as a dependency, never deployed on its own. It also ships one artefact, the `pgcommon-migrate` CLI, as a container image (`ghcr.io/bcbp-solutions-fzc-llc/platform-pgcommon-py`).

**Design:** one `src/pgcommon` package — public modules on top of a private `_internal/` layer, with the layer rules enforced by import-linter in CI. 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. This README is how to **use** it.

**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.

> **Reading this on PyPI?** Links to other files (ARCHITECTURE.md, CHANGELOG.md, docs/, …) are relative to the source repository, which is private, so they do not resolve here. Every linked document ships in the source distribution: `pip download --no-binary :all: --no-deps pgcommon` and unpack the `.tar.gz`.

---

## Contents

- [Mental model](#mental-model) — [Request flow](#request-flow) · [Design principles](#design-principles) · [TL;DR](#tldr)
- [Why this library exists](#why-this-library-exists) — [Why pgcommon instead of raw asyncpg?](#why-pgcommon-instead-of-raw-asyncpg) · [When NOT to use this](#when-not-to-use-this)
- [API overview](#api-overview)
- [Input validation](#input-validation) — [Error reference](#error-reference)
- [Architecture](#architecture)
- [Integrating with a consumer service](#integrating-with-a-consumer-service)
  - [Installing](#installing) · [Quick start](#quick-start) · [Integration checklist](#integration-checklist) · [Production checklist](#production-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) · [PgBouncer / RDS Proxy](#pgbouncer--rds-proxy)
  - [Failure model](#failure-model) · [Performance considerations](#performance-considerations) · [Common pitfalls](#common-pitfalls) · [Configuration reference](#configuration-reference)
- [Upgrading from 0.2.0](#upgrading-from-020)
- [Local development](#local-development)
- [Testing RLS locally](#testing-rls-locally)
- [Testing](#testing)
- [Environment variables](#environment-variables)
- [Security](#security)
- [Observability — Prometheus and OTel](#observability--prometheus-and-otel)
- [Deployment](#deployment) — [Versioning and releases](#versioning-and-releases)
- [CI](#ci)
- [Docker](#docker)
- [Dependencies](#dependencies)
- [Out of scope](#out-of-scope)
- [Contributing](#contributing)
- [Glossary](#glossary)
- [License / ownership](#license--ownership)
- [See also](#see-also)

---

## Mental model

| This library owns | It does NOT own |
|---|---|
| The connection pool (`asyncpg.Pool` underneath, with pgxpool's lifetime, idle, health-check and min-connection semantics rebuilt on top). Sizing, per-connection lifetimes with jitter, fail-fast construction, health, bounded drain | Your tables, schema and **RLS policies**. You write `CREATE POLICY`; pgcommon supplies the session variables it reads |
| **RLS identity injection.** Sets the three `app.*` GUCs from the request's `GUCSet` (held in a `contextvars.ContextVar`) on every acquire — per transaction in PgBouncer mode — and validates it before any connection is waited for. asyncpg's reset on release clears it | **Who the caller is.** Your auth middleware decides, and stores a `GUCSet` with `with_validated_guc_set` (or `GUCAuthMiddleware`) |
| Transactions: commit, rollback on exception or cancellation, savepoints, transaction-local timeouts, retry on `40P01` / `40001`, refusing to report non-atomic work as committed | Idempotency of your transaction function. Retry re-runs it |
| SQLSTATE classification (`is_unique_violation`, `is_deadlock`, `is_connection_lost`, …) | HTTP status mapping. You decide what an error means to your caller |
| The migration **runner** and CLI (`pgcommon.migrate.Runner`, `pgcommon-migrate`, the `pgcommon_migrations` tracking table), golang-migrate-compatible so Go and Python runners share state | The migrations themselves. The library ships **none**; you pass your own `*.sql` |
| Database-client metrics, OTel spans and the slow-query log, registered on **your** `CollectorRegistry`, tracer and logger | Your Prometheus exposition, OTel SDK and logging setup. You pass a `Logger` (`StdlibLogger`, `StructlogLogger` or your own) |
| PgBouncer / RDS Proxy compatibility (`pgbouncer_mode`) | PgBouncer / RDS Proxy themselves (infrastructure) |
| Optional FastAPI glue (`pgcommon.integrations.fastapi`) | The HTTP app factory, request middleware and service metrics (`platform-fastapicommon`) |
| — | The transactional outbox / inbox (`platform-eventcommon`, built on this library's `Pool` and `run_in_tx`) |
| — | Metric ratification and alert rules (pgcommon's `platform_db_*` names are **Proposed**; Go's reference rules apply unchanged) |

```
HTTP request
  → your auth middleware ── with pgcommon.with_validated_guc_set(GUCSet(user_id=…, tenant_id=…, tenant_roles=…))
  → your handler / repository ── await pool.with_conn(fn)  or  await run_in_tx(pool, fn)
      → pgcommon validates the identity, acquires a connection, sets app.* from the contextvar
      → fn(conn) runs your SQL ── metrics · span · slow-query log recorded around every statement
      → release: asyncpg's reset (RESET ALL) clears the session GUCs
  → PostgreSQL enforces RLS ── USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')::uuid)
```

### 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 metrics             ← Tier 1 platform_db_* only
```

> **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 via `pgcommon.metrics.init_with_identity` 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

### 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.

### 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_with_identity(...)` once at startup, before `Pool.create`, to register the Tier 1 `platform_db_*` Prometheus metrics (the identity is mandatory). Set `Config.tracer` for OTel spans.

---

## Why this library exists

Every platform service is multi-tenant, stores data in Postgres, and needs the same half-dozen things done exactly right. Done separately in each service, each of them is a production incident waiting to happen:

- **Tenant isolation at the database, not in every `WHERE` clause.** One forgotten `WHERE tenant_id = $1` leaks every tenant's rows. With a GUC-scoped RLS policy, a forgotten identity returns **no** rows. pgcommon makes setting and clearing that identity automatic, including the subtle parts: a pooled connection must never carry request A's tenant into request B, and in `asyncio` a `ContextVar` set without its reset must never carry it into a concurrent task.
- **Transactions that can't leak.** Rollback on exception **and** on cancellation, with every cleanup bounded to 5 s. A callback that ends its own transaction or swallows a failed statement is reported as a failure, never as a commit. Deadlock and serialization retry with backoff and jitter, retrying only errors that are actually safe to retry.
- **One error vocabulary.** `is_unique_violation(exc)` instead of string matching, working through `__cause__` chains, exception groups and pgcommon's own wrappers, so "conflict", "bad input" and "dependency down" are distinguishable in every service the same way.
- **Uniform telemetry across languages.** Every service exports the same database metrics with the same names, labels and semantics as the Go library. That is what lets one dashboard and one alert set cover Go and Python services alike (Tier 1 of the Enterprise Platform Observability Standard).
- **Migrations that can't race each other.** One runner, one tracking-table convention, and golang-migrate's advisory lock and state, so a Python runner, a Go runner and golang-migrate itself can share a database — plus a CLI image for deploy jobs.

A fix lands in every service with a version bump. This is enforced structurally: `Pool` never exposes the underlying `asyncpg.Pool` or an unmanaged connection, `pgcommon._internal` is private (leading underscore, never re-exported, outside the SemVer surface), and the import-linter contracts in `pyproject.toml` encode the dependency-direction rules CI checks on every PR.

### 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, and a swallowed error or a stray `COMMIT` in the callback silently loses atomicity | `run_in_tx` commits on success, rolls back on exception **and** on cancellation, and refuses to report success for work that was not committed atomically (`TxEndedInCallbackError`, `TxCommitRollbackError`) |
| 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 | Tier 1 `platform_db_*` (queries, acquires, slow queries, retries, pool gauges) update automatically after `pgmetrics.init_with_identity` — identical to the Go library's |
| OTel tracing | Manual span start/end per query | One `db.query` span per statement with the stable semconv attributes `db.system.name`, `db.operation.name` (the verb), `db.namespace`, `server.address`, `server.port` — full SQL (`db.query.text`) and error messages only on opt-in — 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`, `saturated`, `acquired_conns`, `max_conns`, `utilization` with a built-in 5-second deadline, and does not fail readiness just because the pool is busy |
| Graceful shutdown | Manual wait-loop over in-flight connections | `pool.drain_and_close()` marks the pool closed, waits for in-flight callbacks, then interrupts them and raises `DrainTimeoutError` at a bounded 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` and the `pgcommon-migrate` CLI — golang-migrate-compatible state, so Go and Python runners share a tracking table and advisory lock; 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.

### 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.

---

## API overview

Five public modules. Import only these — `pgcommon`, `pgcommon.migrate`, `pgcommon.metrics` (with `pgcommon.metrics_registry`) and `pgcommon.integrations.fastapi` — never `pgcommon._internal` ([VERSIONING.md § What counts as public API](VERSIONING.md#what-counts-as-public-api)). The source of truth is the docstrings and type hints (the package ships `py.typed` and is checked with `mypy --strict`); [docs/lld/platform-lld-pgcommon-py.md](docs/lld/platform-lld-pgcommon-py.md) § 5 has the full contract of every symbol.

### `pgcommon` — pool, connections and transactions

| Symbol | Purpose |
|---|---|
| `await Pool.create(Config) -> Pool` | Validate the config and the DSN, apply defaults, register the pool's metrics, **ping**. Every misconfiguration is an error here, not later |
| `config_from_env() -> (Config, list[ConfigWarning])` | Build a `Config` from `DATABASE_URL` / `PG_*`. Never raises; log every warning (insecure `sslmode`, bad values, missing credentials) |
| `Config.validate()` | The checks that need no DSN parsing, for startup validation without connecting |
| `await pool.with_conn(fn)` | Acquire → identity → `fn(conn)` → release. `fn` is `async def fn(conn: Conn)`; `Conn` is asyncpg's `PoolConnectionProxy` |
| `await run_in_tx(pool, fn, *, isolation, read_only, deferrable)` = `pool.run_in_tx(…)` = `pool.with_tx(…)` | Transaction: commit on return, rollback on exception (re-raised) or cancellation (propagates) |
| `await run_in_tx_with_retry(pool, fn, max_attempts, …)` | Retry the whole transaction on `40P01` / `40001`, immediately |
| `await run_in_tx_with_retry_opts(pool, fn, RetryOptions, …)` | Same, with exponential backoff and jitter; a cancellation during the sleep propagates at once |
| `await run_in_savepoint(conn, name, fn)` | Nested unit of work. On error, rolls back to the savepoint only |
| `await pool.health(*, timeout=5.0) -> HealthStatus` | Readiness snapshot: `healthy`, `saturated`, `utilization`, connection counts. Never raises |
| `await pool.drain_and_close(*, timeout=30.0)` · `await pool.aclose()` · `pool.close()` | Bounded graceful shutdown (`DrainTimeoutError` when it had to interrupt) · Go's `Close`, waits without a limit · synchronous hard stop |
| `await pool.ping()` · `pool.stat() -> Stat` · `pool.name` · `pool.metrics_auto_registered` | Connectivity check · counters snapshot · the pool's metric label · whether `create` registered the pool gauges |
| `migration_dsn_from_env()` | `MIGRATION_DATABASE_URL`, else the app DSN |
| `sql_operation(sql)` | The SQL verb used as the low-cardinality `operation` metric label |
| `Logger` · `StdlibLogger` · `StructlogLogger` | The logging protocol `Config.logger` takes, and adapters for `logging` and structlog (extra `structlog`) |
| `SlowQueryTracer` · `AcquireTracer` · `ReleaseTracer` | Standalone slow-query logger · protocols a `Config.query_logger` may implement for acquire / release events |

### `pgcommon` — tenant identity (RLS)

| Symbol | Purpose |
|---|---|
| `GUCSet(user_id, tenant_id, tenant_roles)` · `.validate()` · `.is_zero()` | The request identity (a frozen dataclass). A zero set injects nothing |
| `with with_validated_guc_set(GUCSet)` | **Use in middleware.** Validate the identity, then store it on the contextvar for the block |
| `with with_guc_set(GUCSet)` | Store without validating (validated later, at acquire) |
| `guc_set_from_context()` | Pass as `Config.guc_provider`. The pool calls it on every acquire |
| `await inject_gucs(conn, GUCSet)` | Manual, session-scoped injection on a direct connection |
| `IDENTITY_ERRORS` | The tuple of the five `GUCSet.validate()` errors — a caller error, not a database failure |

### `pgcommon` — errors

| Helper | Matches |
|---|---|
| `is_unique_violation` · `constraint_name` | `23505` · the offending constraint |
| `is_foreign_key_violation` · `is_not_null_violation` · `is_check_violation` | `23503` · `23502` · `23514` |
| `is_deadlock` · `is_serialization_failure` | `40P01` · `40001` (retried by the retry helpers) |
| `is_lock_not_available` · `is_query_canceled` | `55P03` · `57014` |
| `is_connection_exception` · `is_insufficient_resources` · `is_connection_lost` | class `08` · class `53` · a lost connection or a server restarting (Python-only) |
| `is_operator_intervention` · `is_system_error` | class `57` minus `57014` · class `58` |
| `is_pg_error` · `sql_state` · `is_pool_closed` | any PostgreSQL server error · the raw code · shutting down |
| `PgCommonError` | Base class of every pgcommon sentinel error ([Error reference](#error-reference)) |

Typical meanings and an example are in [PostgreSQL error helpers](#postgresql-error-helpers).

### `pgcommon.migrate` — migration runner

| Symbol | Purpose |
|---|---|
| `Runner(dsn, source, *, logger, table, lock_timeout)` | `source` is **required** (a directory or an `importlib.resources` `Traversable` holding your `*.up.sql` / `*.down.sql`); tracking table default `pgcommon_migrations`; `lock_timeout` default 30 s |
| `up()` · `down(n)` · `steps(±n)` · `version()` · `force(v)` | Apply · roll back · step · report `(version, dirty)` · recover a dirty schema. Each takes an optional `stop: asyncio.Event` |
| `DEFAULT_MIGRATIONS_TABLE` · `DEFAULT_LOCK_TIMEOUT` · `DEFAULT_CONNECT_TIMEOUT` | `pgcommon_migrations` · 30 s · 10 s |
| `MigrationDirtyError` · `VersionNotInSourceError` · `ShortLimitError` · `MigrationStoppedError` · … | A dirty schema · a recorded version with no migration in the source · fewer than `n` to run · `stop` was set ([Error reference](#error-reference)) |

### `pgcommon.metrics` — observability

| Symbol | Purpose |
|---|---|
| `init_with_identity(Identity(domain, service, environment), registry=None)` | **The only** init: registers the Tier 1 `platform_db_*` metrics (the identity is mandatory); returns `list[RegistrationWarning]` |
| `identity_from_labels(mapping)` · `identity_from_env(domain, service="")` · `environment_from_env()` | Build the identity. An empty `environment` is read from `APP_ENV` → `ENVIRONMENT` → `dev` |
| `OTelQueryRecorder(tracer, *, allow_full_statements=False)` | The span recorder `Config.tracer` wires in; usable as a standalone asyncpg query logger |
| `pgcommon.metrics_registry.registry()` · `lookup(name)` | Every metric's tier, status and ratification packet |
| `replace_current(None)` | Reset the process-wide metric set between tests |

### `pgcommon.integrations.fastapi` — optional (`pgcommon[fastapi]`)

| Symbol | Purpose |
|---|---|
| `create_lifespan(config)` | ASGI lifespan: `Pool.create` at startup (`app.state.pool`), `drain_and_close()` at shutdown |
| `get_pool` | `Depends()` provider returning `app.state.pool` |
| `GUCAuthMiddleware(app, identity_extractor=…)` | Validates the extracted `GUCSet`, answers `401` for `IDENTITY_ERRORS`, otherwise runs the request under `with_guc_set` |
| `healthz(*, utilization_threshold=0.9)` | A `/healthz` handler: `503` when unhealthy or at the threshold, else pool utilisation |

### `pgcommon-migrate` CLI (image `ghcr.io/bcbp-solutions-fzc-llc/platform-pgcommon-py`)

| Command | Purpose |
|---|---|
| `up` · `down [N]` | Apply all pending · roll back N (default 1) |
| `status` | Print the version and dirty flag; exits 3 when dirty, 4 when the version is not in `MIGRATIONS_DIR`. **Not read-only** |
| `force <V>` | Record version V and clear dirty (`-1` = none), after repairing a failed migration by hand |
| `version` / `--version` · `-h` / `--help` | Package version · usage |

Configuration is environment-only (`MIGRATION_DATABASE_URL`, `MIGRATIONS_DIR`, …). Exit codes: **0** success (including "no change"); **1** usage, configuration, connection or migration error; **3** dirty; **4** version not in `MIGRATIONS_DIR`. Details: [The `pgcommon-migrate` CLI](#the-pgcommon-migrate-cli). Also `python -m pgcommon.migrate`.

---

## Input validation

pgcommon validates everything it can **before** touching the database, so misconfiguration fails at startup and a bad identity fails at the request edge, before a connection is even waited for. Every error that could quote a DSN has its password redacted, throughout its exception chain. Environment variables are the exception to "fail fast": `config_from_env()` never raises, it falls back to the default and returns a `ConfigWarning` — log them.

| Input | Rule | Checked by | Raises |
|---|---|---|---|
| `GUCSet` | In this order: `user_id` needs `tenant_id`; `tenant_roles` need `tenant_id`; `user_id` / `tenant_id` contain no NUL and are UTF-8-encodable; each role is non-empty, contains no `,` (the GUC separator) and is a valid value. `tenant_id` alone is valid (tenant-only RLS); a zero `GUCSet` injects nothing | `GUCSet.validate()`, `with_validated_guc_set`, `GUCAuthMiddleware` (→ `401`), every acquire in both modes (before waiting for a connection), `inject_gucs` | `InconsistentGUCSetError` · `TenantRolesWithoutTenantError` · `InvalidGUCValueError` · `EmptyTenantRoleError` · `InvalidTenantRoleError` (`IDENTITY_ERRORS`); at acquire re-raised as the same class with the message `pgcommon: invalid GUC identity: …` |
| DSN | Non-empty; parseable as a URL or a libpq keyword/value string; `connect_timeout` (else `PGCONNECT_TIMEOUT`) whole seconds, not negative (a positive value is at least 2 s, as in libpq) | `Config.validate()` (empty only), `Pool.create` before connecting | `DSNRequiredError` · `InvalidDSNError` (never quotes the password) |
| Connection counts | `max_conns` / `min_conns` not negative (`None` or `0` = default; an explicit `min_conns=0` is kept); `min_conns ≤ max_conns` after the DSN's `pool_*` values and defaults | `Config.validate()` (when both are set), `Pool.create` | `InvalidMaxConnsError` · `InvalidMinConnsError` · `MinConnsExceedsMaxError` |
| Durations | `max_conn_lifetime`, `max_conn_idle_time`, `health_check_period`, `command_timeout` and the DSN's `pool_*` durations not negative. A negative `max_conn_lifetime_jitter`, `slow_query_threshold`, `acquire_timeout` or `saturated_alive_window` is the "off" value, not an error | `Config.validate()`, `Pool.create` | `NegativeDurationError` |
| Transaction and migration timeouts | `statement_timeout` / `lock_timeout` and `Runner(lock_timeout=)` at most INT_MAX ms (about 24.8 days); rounded up to whole milliseconds | `Config.validate()`, `Pool.create`, `Runner(...)` | `TimeoutOutOfRangeError` |
| `pool_name` | `^[a-z][a-z0-9_]{0,31}$` (it becomes the `pool` metric label); empty means `"default"`; `PG_POOL_NAME` is taken verbatim | `Config.validate()`, `Pool.create` | `InvalidPoolNameError` |
| Metric identity | `domain`, `service` and `environment` required, each `^[a-z][a-z0-9_-]{0,62}$`; the message names `APP_ENV` / `ENVIRONMENT` when the bad value came from there | `init_with_identity` (nothing changes on failure) | `InvalidIdentityError` |
| Savepoint name | `^[A-Za-z_][A-Za-z0-9_]*$`, checked before the name is interpolated into `SAVEPOINT` | `run_in_savepoint` | `InvalidSavepointNameError` |
| Migrations table | Not empty, no NUL (`Runner(table=…)`); a quoted `x-migrations-table` is `"table"` or `"schema"."table"`; a 0.2.0 mixed-case table that PostgreSQL folded to lower case is refused, not adopted | `Runner(...)` (the name), every runner operation (the database) | `InvalidMigrationsTableError` |
| Migration source | At least one `{version}_{title}.up.{ext}` / `.down.{ext}` file at the top level; one file per version and direction | Every runner operation | `NoMigrationsFoundError` · `DuplicateMigrationError` |
| Runner arguments | `down(n)` needs `n > 0`; `force(v)` needs `v ≥ -1`, and `v ≥ 0` must be a migration in the source | `Runner` | `ValueError` · `VersionNotInSourceError` |
| Migration DSN options | `x-statement-timeout` a non-negative integer (ms), `x-multi-statement` a boolean, `x-multi-statement-max-size` an integer; unknown `x-` options are ignored with a warning | `Runner` | `ValueError` (`migrate: invalid …`) |
| CLI environment | `MIGRATIONS_DIR` required; `MIGRATIONS_LOCK_TIMEOUT` must parse; the `DATABASE_URL` / `PG_*` fallback is refused when `PG_BOUNCER_MODE` is true | `pgcommon-migrate` | exit code 1 |

### Notable validation rules

| Rule | Enforcement |
|---|---|
| An invalid identity never reaches SQL, and never holds a pool slot | Validated before the acquire waits, in direct and PgBouncer mode alike (a deliberate divergence: Go validates PgBouncer mode's identity after `BEGIN`); recorded as an acquire wait, not an acquire error |
| A connection never carries a previous caller's identity | asyncpg's reset on release (`RESET ALL`) clears the session GUCs; a reset that fails or exceeds 5 s terminates the connection instead of returning it |
| GUC values and role names can't alter the SQL | Always bind parameters (`set_config($1, …)`); savepoint names are regex-checked before interpolation; the migrations table name is quoted |
| Metric labels are bounded | `operation` from a closed verb allowlist (`other` for anything else); `pool_name` and the identity validated; SQL text is never a label |
| Transaction timeouts stay in the transaction | `statement_timeout` / `lock_timeout` are applied with `set_config(…, true)` |
| `PG_BOUNCER_MODE=yes` is not silently `false` | `config_from_env()` accepts Go's `strconv.ParseBool` spellings and warns on anything else |
| An insecure connection is visible | An insecure `sslmode` warns, and so does a `DATABASE_URL` / `MIGRATION_DATABASE_URL` without one when asyncpg could fall back to plaintext |

### Error reference

| Error | Raised by | When |
|---|---|---|
| `DSNRequiredError` | `Pool.create`, `Config.validate()` | `Config.dsn` is empty |
| `InvalidDSNError` (a `ValueError`) | `Pool.create` | The DSN cannot be parsed (`pgcommon: invalid DSN: …`, never quoting the password) — raised before connecting |
| `InvalidMaxConnsError` · `InvalidMinConnsError` | `Pool.create`, `Config.validate()` | A negative connection count |
| `MinConnsExceedsMaxError` | `Pool.create` | `min_conns > max_conns` after the DSN's `pool_*` values and defaults |
| `NegativeDurationError` (a `ValueError`) | `Pool.create`, `Config.validate()` | A negative lifetime, idle time, health-check period or `command_timeout`, in the `Config` (or a negative `pool_*` duration in the DSN) (a negative jitter or slow-query threshold is the "disabled" value, not an error) |
| `TimeoutOutOfRangeError` (a `ValueError`) | `Pool.create`, `Runner` | `statement_timeout` / `lock_timeout` (or a migration `lock_timeout`) above INT_MAX ms |
| `InvalidPoolNameError` | `Pool.create`, `Config.validate()` | `pool_name` fails `^[a-z][a-z0-9_]{0,31}$` |
| `InvalidIdentityError` (a `ValueError`) | `pgcommon.metrics.init_with_identity` | An invalid metric identity |
| `PoolClosedError` | `with_conn`, `run_in_tx`, `ping` | The pool was closed (`health()` never raises — see [Health check and load shedding](#health-check-and-load-shedding)); match with `is_pool_closed` |
| `DrainTimeoutError` (a `TimeoutError`) | `drain_and_close` | In-flight callbacks did not finish in time; their connections were interrupted and the pool closed |
| `AcquireTimeoutError` (a `TimeoutError`) | `with_conn`, `run_in_tx`, `ping` | No connection could be acquired (and prepared) within `acquire_timeout` — or preparing one (GUC injection, `BEGIN`) timed out under `command_timeout` on every connection tried (the `TimeoutError` is its `__cause__`) |
| `TxEndedInCallbackError` | `run_in_tx` / `with_tx`, PgBouncer-mode `with_conn` | The callback ended the transaction pgcommon opened for it; `callback_error` holds the callback's own exception, if any. Never retried |
| `TxCommitRollbackError` | `run_in_tx` / `with_tx`, PgBouncer-mode `with_conn` | The callback swallowed a failed statement, so `COMMIT` rolled back |
| `SavepointCleanupError` | `run_in_savepoint` | `ROLLBACK TO` / `RELEASE SAVEPOINT` failed after `fn` failed — or was not attempted because a statement in `fn` was interrupted on the client side and the connection terminated (`fn`'s exception is `callback_error`) |
| `InvalidSavepointNameError` | `run_in_savepoint` | Savepoint name is empty or fails `^[A-Za-z_][A-Za-z0-9_]*$` |
| `InconsistentGUCSetError` · `TenantRolesWithoutTenantError` · `InvalidTenantRoleError` · `EmptyTenantRoleError` · `InvalidGUCValueError` (together: `IDENTITY_ERRORS`) | `GUCSet.validate()`, `with_validated_guc_set`, acquire, `inject_gucs` | `user_id` without `tenant_id` · roles without `tenant_id` · a role containing `,` · an empty role · a NUL or non-UTF-8-encodable value. At acquire the message is prefixed `pgcommon: invalid GUC identity:` |
| `MigrationDirtyError` | `Runner`, CLI exit 3 | The schema is dirty from a failed migration; repair, then `force` |
| `VersionNotInSourceError` | `Runner`, CLI exit 4 | The recorded version has no migration in the source (or `force` targets one); `.version` / `.dirty` carry the recorded state |
| `NoMigrationsFoundError` | `Runner` | No migration file at the top level of the source |
| `DuplicateMigrationError` (`pgcommon.migrate`) | `Runner` | Two files give one version the same direction |
| `ShortLimitError` | `Runner.down` / `steps` | Fewer than `n` migrations to run; `.short` is how many were missing |
| `MigrationStoppedError` | `Runner` | The run's `stop` event was set |
| `InvalidMigrationsTableError` | `Runner` | An empty tracking-table name, one containing NUL, or a malformed quoted `x-migrations-table` |

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

---

## Architecture

The Go module's Clean Architecture layers become one flat package: public modules on top, a private `_internal/` package of dependency sinks below, and two optional outer modules (`migrate`, `integrations`) that build on the core and are never imported by it. Full layer diagrams, flows, invariants and the asyncio translation notes are in **[ARCHITECTURE.md](ARCHITECTURE.md)**; the standalone Mermaid sources are in **[docs/architecture/](docs/architecture/README.md)**.

```
platform-pgcommon-py/
├── src/pgcommon/                      # The package (the wheel ships only this)
│   ├── __init__.py                    # Public re-exports: config, errors, guc, pool, tx, retry, tracer, logger, sql_operation, …
│   ├── pool.py                        # Pool (create, with_conn, run_in_tx / with_tx, ping, health, stat, drain_and_close, aclose, close), Stat, HealthStatus
│   ├── tx.py · retry.py               # run_in_tx, run_in_savepoint, transaction setup and escape detection · RetryOptions, the retry helpers
│   ├── guc.py                         # GUCSet, with_guc_set, with_validated_guc_set, guc_set_from_context, inject_gucs
│   ├── config.py                      # Config, config_from_env, migration_dsn_from_env, ConfigWarning, defaults
│   ├── errors.py                      # PgCommonError and every sentinel, IDENTITY_ERRORS, the is_* helpers, sql_state, constraint_name
│   ├── tracer.py · logger.py          # SlowQueryTracer, AcquireTracer / ReleaseTracer · Logger protocol, StdlibLogger, StructlogLogger
│   ├── metrics.py · metrics_registry.py  # init_with_identity, Identity, pool gauges, OTelQueryRecorder · the governed platform_db_* inventory
│   ├── migrate/                       # Runner and its errors; cli.py (pgcommon-migrate), __main__.py (python -m pgcommon.migrate)
│   ├── integrations/fastapi.py        # create_lifespan, get_pool, GUCAuthMiddleware, healthz (extra: pgcommon[fastapi])
│   └── _internal/                     # PRIVATE: DSN parsing and redaction, transaction-shape lexer, statement scopes, SQL verb, SQLSTATE, …
├── tests/                             # unit (no Docker) · integration (Postgres) · e2e (Postgres + PgBouncer), via testcontainers; fixtures/
├── migrations/                        # An example 0001 migration pair, for trying the CLI locally (not in the wheel)
├── docs/                              # adr/, architecture/ (Mermaid sources), lld/, observability/, MigrationGuide.md
├── .github/workflows/                 # ci, validate-test, validate-quality, release, docs, changelog-check
├── .github/scripts/                   # Job logic: coverage gate, arch-lint, api-compat, smoke tests, image fingerprint/tags, release checksums/notes, …
├── Dockerfile · build-constraints.txt # The migrate-CLI image (digest-pinned bases) · the hash-locked build backend
└── pyproject.toml · uv.lock · Makefile · docker-compose.yml · .env-example
```

### Dependency rules (enforced by import-linter in CI)

The contracts live in `pyproject.toml` (`[tool.importlinter]`); `make arch-lint` runs them, as validate-test's `test` job does.

| Component | May depend on |
|---|---|
| `pgcommon._internal.*` | Nothing public in `pgcommon` (a dependency sink) |
| `pgcommon.errors` | `pgcommon._internal.sqlstate` only |
| The core: the `pgcommon` package root, `config`, `errors`, `guc`, `logger`, `metrics`, `metrics_registry`, `pool`, `retry`, `tracer`, `tx`, `_internal` | Each other — **never** `pgcommon.migrate` or `pgcommon.integrations` |
| `pgcommon.migrate`, `pgcommon.integrations` | The core — **never** each other |

Plus the **RLS guard**, a grep in validate-quality: no session-level `SET app.tenant_id|user_id|tenant_roles` in `src/` (`*.py`, `*.sql`). Identity is set only through `set_config` (cleared by the reset on release), or `SET LOCAL` inside a transaction.

### Runtime dependencies (what consumers inherit)

| Library | Range (`pyproject.toml`) | Locked in `uv.lock` | Purpose |
|---|---|---|---|
| `asyncpg` | `>=0.30,<0.32` | 0.31.0 | Driver and pool. **Capped**: pgcommon relies on asyncpg behaviour that is not public API (see [Dependencies](#2-upstream-libraries)) |
| `prometheus-client` | `>=0.20` | 0.26.0 | Metrics |
| `opentelemetry-api` | `>=1.24` | 1.44.0 | Span API (the SDK is the service's) |
| `fastapi` (extra `fastapi`) | `>=0.110` | 0.140.0 | `pgcommon.integrations.fastapi` |
| `structlog` (extra `structlog`) | `>=24.1` | 26.1.0 | `StructlogLogger` |

No logging framework, no OTel SDK, no web framework outside the `fastapi` extra (importing `pgcommon` never imports `fastapi` or `starlette`), and no test tool: those live in the `dev` dependency group, precisely so they never enter a consumer's environment.

---

## Integrating with a consumer service

Everything a consuming service does with pgcommon, in roughly the order it does it: install, wire the pool and metrics once at startup, set the tenant identity per request, run queries and transactions, map errors, expose health, drain at shutdown, and run migrations as a separate step. The two checklists are the short version; the subsections after them are the reference.

### 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, migration_dsn_from_env
from pgcommon.migrate import Runner


async def main() -> None:
    # 1. Run any pending schema migrations (in production: an init container or
    #    pre-deploy job running `pgcommon-migrate up` — see Migrations).
    runner = Runner(migration_dsn_from_env(), "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_with_identity`** — call once at startup, before `Pool.create`, with the service's `domain`/`service`/`environment`; log the returned warnings ([Observability](#observability--prometheus-and-otel))
- [ ] **`pool_name`** — set `Config.pool_name` (or `PG_POOL_NAME`) when a service runs more than one pool, e.g. `primary` / `replica` ([Prometheus metrics](#prometheus-metrics))
- [ ] **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()`, and log the `DrainTimeoutError` it raises when callbacks had to be interrupted ([Graceful shutdown](#graceful-shutdown))
- [ ] **Transactions** — never `COMMIT` / `ROLLBACK` / `BEGIN` inside a `run_in_tx` callback, and re-raise statement errors instead of swallowing them ([Transactions](#transactions))
- [ ] **Config warnings** — log every `ConfigWarning` `config_from_env()` returns; set `sslmode=verify-full` (or at least `require`) in the DSN ([Configuration reference](#configuration-reference))

### Production checklist

The deployment-time complement of the [Integration checklist](#integration-checklist): what to have in place before a service takes production traffic.

- [ ] **Metrics identity first** — `pgcommon.metrics.init_with_identity(...)` runs once at startup, **before** `Pool.create`, and its warnings are logged ([Observability](#observability--prometheus-and-otel))
- [ ] **Three timeouts, nested** — `PG_STATEMENT_TIMEOUT` (the server's limit on a runaway query) < `PG_COMMAND_TIMEOUT` (the client's limit on every statement), and `PG_ACQUIRE_TIMEOUT` (default `30s`) bounds waiting for a connection. `PG_COMMAND_TIMEOUT` is unset by default: without it a frozen or paused database server — which still acknowledges at the TCP level, so keepalive never notices — holds every in-flight query until it answers again ([Suggested defaults](#suggested-defaults-production), [Failure model](#failure-model))
- [ ] **TLS** — `sslmode=verify-full` (or at least `require`) in both the application's and the migration job's DSN ([Configuration reference](#configuration-reference))
- [ ] **Graceful shutdown** — `drain_and_close()` in the app's lifespan shutdown (`create_lifespan` does it), with a drain deadline below the orchestrator's termination grace period ([Graceful shutdown](#graceful-shutdown), [FastAPI integration](#fastapi-integration))
- [ ] **Migrations as a separate step** — an init container or pre-deploy job running `pgcommon-migrate up` on a direct (non-PgBouncer) connection with a DDL role, never `runner.up()` in every replica ([Migrations](#migrations), [The `pgcommon-migrate` CLI](#the-pgcommon-migrate-cli))
- [ ] **Probes** — readiness on `pool.health()` (it reports unhealthy when the database is unreachable); liveness on a check that does not touch the database, so a database outage takes pods out of rotation instead of restarting them. On a saturated pool `health()` can report healthy for up to `saturated_alive_window` (10 s) after the server stopped answering — lower it, or `off`, for faster detection ([Health check and load shedding](#health-check-and-load-shedding))

### Connection pool

`Pool.create` wraps [`asyncpg.create_pool`](https://magicstack.github.io/asyncpg/current/api/index.html#connection-pools) and applies sensible defaults. The configuration is validated and the DSN parsed before anything connects (`InvalidDSNError`, `NegativeDurationError`, `TimeoutOutOfRangeError`, …), and 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),
        statement_timeout=timedelta(seconds=30),  # per transaction, see Transactions
        logger=my_logger,  # implements the pgcommon.Logger protocol
        tracer=my_otel_tracer,  # an opentelemetry.trace.Tracer
    )
)
```

Every sizing and duration field is optional. `None` (or `0`, as in Go) means "the DSN's `pool_*` parameter, else the library default", so the precedence is **explicit `Config` field > DSN `pool_*` parameter > default** — `?pool_max_conns=20` in the DSN is used unless `Config.max_conns` is set. The DSN may be a URL or a libpq keyword/value string (`host=db dbname=app …`); its `pool_*` parameters and `connect_timeout` configure the pool and are never sent to the server. `connect_timeout` (else `PGCONNECT_TIMEOUT`) is whole seconds as pgx parses it — `0` = no limit, a positive value at least 2 s as in libpq, a negative or non-integer value `InvalidDSNError` — exactly as the migration runner parses it; unset, the pool sets no connect limit of its own (an acquire is bounded by `acquire_timeout`).

**Connection lifetime is per connection**, as in pgxpool: each physical connection gets its own deadline (`max_conn_lifetime` plus a random `0…max_conn_lifetime_jitter`, default `max_conn_lifetime / 10`) and is closed when it is released after that deadline — or, if it sat idle past it, when it is next acquired, so it is never handed out — so connections opened together do not all reconnect at once. Idle connections are closed by the health check (every `health_check_period`) once they sat idle for `max_conn_idle_time` — but never below `min_conns`, as in pgxpool — and an idle connection past its lifetime is closed there too; the pool is then topped back up to `min_conns`.

**Dead connections are replaced on acquire.** Whichever statement pgcommon runs on a connection before handing it out — GUC injection (direct mode), `BEGIN` and the transaction's setup (`run_in_tx`, PgBouncer-mode `with_conn` with an identity), or a ping of a connection idle for more than a second when nothing else runs first (pgxpool's default `ShouldPing`) — finding it dead (a failover, a restart) makes the pool take another one, for as many dead connections as there are, within `acquire_timeout`. Never once the callback started: its statements are never retried.

**Unresponsive server.** Every connection's socket gets TCP keepalive (probes after 30 s of silence, every 10 s, 3 unanswered close it; on Linux also `TCP_USER_TIMEOUT` = 60 s), so a peer that vanished is noticed in about a minute. A server that is up but not answering (a frozen process or host, `docker pause`) still acknowledges at the TCP level: only a client-side timeout bounds that. Set `command_timeout` (`PG_COMMAND_TIMEOUT`, unset by default) — asyncpg's limit on every statement, above `statement_timeout` so the server's own limit normally fires first — or bound calls with `asyncio.timeout`. Either way pgcommon does not wait for an interrupted statement's cancel request: the connection is terminated (the server rolls the transaction back), and every cleanup statement (ROLLBACK, asyncpg's reset on release) is bounded to 5 s, after which the connection is terminated too. A release never fails the caller: committed work is returned even if the connection died afterwards (the failure is logged at `WARN`).

#### 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`, `new_conns_count` (connections opened, replacements included), `max_lifetime_destroy_count` (closed past their lifetime), `max_idle_destroy_count` (closed by the health check after `max_conn_idle_time`, above `min_conns`), `empty_acquire_wait_time` (total wait of acquires that found no idle connection). pgxpool's `ConstructingConns` has no equivalent: asyncpg does not expose connections that are still being dialled.

#### Graceful shutdown

Use `drain_and_close` when shutting down. All three close methods mark the pool closed first, so new `with_conn` / `run_in_tx` / `ping` calls raise `PoolClosedError` at once; they differ in what happens to callbacks still running:

| Method | In-flight callbacks | Returns |
|---|---|---|
| `await pool.drain_and_close(timeout=30)` | Waited for until `timeout`; then every acquired connection is closed gracefully — asyncpg first sends the server a cancel request for a statement still running, so the server stops it and rolls back its transaction, releasing its locks — and terminated after 0.5 s at most. In-flight queries fail; `is_pool_closed` matches their errors | Within about `timeout` + 1 s; raises `DrainTimeoutError` (naming the connections still acquired and the callers still waiting for one) when it had to interrupt. The wait for connections no callback holds (an in-flight `ping` / `health`) is bounded too. Cancelling the call terminates the pool and leaves nothing running |
| `await pool.aclose()` | Waited for **without a limit** (Go `Close`) | When every acquired connection is released — a single stuck query blocks it forever |
| `pool.close()` | Interrupted at once (synchronous hard stop): sockets are closed, no cancel request is sent, so the server keeps running an in-flight statement (and holding its locks) until it finishes or `statement_timeout` ends it | Immediately |

```python
from pgcommon import DrainTimeoutError

try:
    await pool.drain_and_close(timeout=30)  # keep it below terminationGracePeriodSeconds
except DrainTimeoutError as exc:
    logger.warning("db drain timed out", error=str(exc))
    # connections were interrupted and the pool is closed; safe to exit
```

A failure while asyncpg closes the pool is logged and the pool terminated, never raised. Stop accepting requests (the HTTP server) before draining.

### 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, inside pgcommon's acquire before `fn` runs (Go: the `PrepareConn` hook) | Each `with_conn`/`run_in_tx` call, inside a wrapping transaction |
| **Scope** | Session-level (`is_local=false`) — cleared by asyncpg's reset (`RESET ALL`) when the connection is released; re-injected on the next acquire | 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:**

```
with_conn / run_in_tx called
    │
    ▼
guc_provider() → GUCSet.validate()   ← before waiting for a connection: an invalid identity
    │                                  fails at once (IDENTITY_ERRORS), costs no connection
    ▼
connection acquired from asyncpg
    │
    ▼
pgcommon injects ──── 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 — asyncpg's reset (RESET ALL) clears the session GUCs
```

If the injection fails, the connection is not simply discarded: a cancelled caller's connection is released (still healthy); one whose session the server already ended (terminated backend, server restart) is replaced and the injection retried on another connection — logged at Info (`pgcommon: connection found dead during acquire; replaced`) and not counted as a query error, since the caller never sees it; one `command_timeout` interrupted (the server did not answer) is terminated and the injection retried on another connection, logged at Warn — when every attempt times out the caller gets `AcquireTimeoutError`; a PostgreSQL error keeps the connection and fails the request; anything else terminates it. A failure the caller gets is logged at Warn (`pgcommon: GUC injection failed in PrepareConn`) and counted as a failed query. An invalid identity is raised as `pgcommon: invalid GUC identity: …` — the same class as `GUCSet.validate()`'s error, which is its `__cause__`.

**PgBouncer mode timing:**

```
identity resolved and validated (before waiting; invalid → "pgcommon: invalid GUC identity: …")
    │
    ▼
connection acquired (no GUC injection at acquire in PgBouncer mode)
    │
    ▼
with_conn opens an implicit wrapping transaction (run_in_tx: its own transaction)
    │
    ▼
SELECT set_config('app.user_id', $1, true), set_config(..., true), set_config(..., true)
    │
    ▼
statement_timeout / lock_timeout applied with set_config(..., true), when configured
    │
    ▼
fn(conn) — GUCs in place for this transaction only (fn must not end it: TxEndedInCallbackError)
    │
    ▼
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 IDENTITY_ERRORS, Config, GUCSet, Pool, guc_set_from_context, with_guc_set

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


# 2. In ASGI middleware — validate the identity, then store it for the duration of the request.
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:
        guc_set.validate()
    except IDENTITY_ERRORS:  # every GUCSet.validate() error: a caller error, not a DB failure
        return Response("invalid identity", status_code=401)
    with with_guc_set(guc_set):
        return await call_next(request)


# 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). `with_validated_guc_set(guc_set)` combines both steps when you would rather catch the error around the whole block.

`GUCSet.validate()` raises, in order: `InconsistentGUCSetError` (`user_id` without `tenant_id`), `TenantRolesWithoutTenantError` (`tenant_roles` without `tenant_id` — a policy checking only the role list would grant that role in every tenant), `InvalidGUCValueError` (a NUL character or a value not encodable as UTF-8, which PostgreSQL rejects), and per role `EmptyTenantRoleError`, `InvalidTenantRoleError` (a comma, the GUC separator) or `InvalidGUCValueError`. `IDENTITY_ERRORS` is the tuple of all five. `tenant_id` alone is valid (tenant-only RLS); a zero `GUCSet` injects nothing.

#### 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`. Use it only on a direct connection: a pooled connection's session settings are cleared when it is released, but behind PgBouncer transaction pooling a session-level GUC reaches whichever client gets that backend next. 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()` (30 s) at shutdown — never reconstructed per request; a `DrainTimeoutError` from that drain propagates out of the lifespan. `get_pool` is the corresponding `Depends()` provider.

`GUCAuthMiddleware` validates the extracted identity and responds `401` for any of `IDENTITY_ERRORS`; only that validation is guarded, so an identity error raised later by a handler (or by the pool, for a `GUCSet` a handler set itself) is not turned into this request's 401.

> **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_guc_set(...)` — entered around `call_next(request)` after validation 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.

> **Work started from a request runs as that request's identity.** A task copies the context when it is created, so Starlette `BackgroundTasks` and `asyncio.create_task(...)` called from a handler run as the requesting user and tenant even after the response — right for per-request work (its data stays scoped to the tenant that asked). The hazard is a **long-lived worker started lazily inside a request** (cache refresher, queue consumer, "start on first use" singleton): it acts as the *first* requester's tenant forever. Start such workers from the app's lifespan (outside any request, so no `GUCSet`), or start them in a fresh context — `asyncio.create_task(worker(), context=contextvars.Context())` — or give the worker an explicit identity with `with with_guc_set(GUCSet()):` (an empty `GUCSet`: no GUC injection).

### 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 → pgcommon sends **`COMMIT`** and checks the server's answer
- `fn` raises → **rollback**, original exception re-raised
- the surrounding task is cancelled while `fn` is running → **rollback**, `asyncio.CancelledError` propagates
- `fn` ended the transaction itself → **rollback** of whatever is open, `TxEndedInCallbackError` (see below)
- `fn` swallowed a failed statement → **rollback**, `TxCommitRollbackError` (see below)

`isolation` is `"read_committed"`, `"repeatable_read"`, `"serializable"` or `None` (session default); `read_only=True` and `deferrable=True` set the access and deferrable modes. Every rollback and cleanup statement pgcommon sends is bounded to 5 seconds, so a half-open peer cannot pin the connection (and `drain_and_close`) for the TCP timeout.

#### Statement and lock timeouts

`Config.statement_timeout` / `lock_timeout` (`PG_STATEMENT_TIMEOUT` / `PG_LOCK_TIMEOUT`) are applied right after `BEGIN` with `set_config(…, true)` — transaction-local, so they hold behind PgBouncer transaction pooling and never leak to the next user of the connection. They apply to every `run_in_tx` / `with_tx` and to the wrapping transaction of a PgBouncer-mode `with_conn` that carries an identity (Go 2.0.1). Values are rounded up to whole milliseconds; above INT_MAX ms (about 24.8 days) `Pool.create` raises `TimeoutOutOfRangeError`. A timed-out statement fails with SQLSTATE `57014` (`is_query_canceled`), a lock wait with `55P03` (`is_lock_not_available`).

> **No timeout without a wrapping transaction.** A `with_conn` in direct mode, or in PgBouncer mode without an identity, applies neither timeout. Back them with a role setting (`ALTER ROLE <role> SET statement_timeout = …`), which holds behind PgBouncer too.

#### Do not end the transaction in the callback

`fn` runs inside the transaction `run_in_tx` opened for it. If `fn` commits, rolls back or otherwise ends it — `COMMIT`, `ROLLBACK [AND CHAIN]`, `END`, `ABORT`, `PREPARE TRANSACTION`, or a multi-statement string such as `"COMMIT; BEGIN; SELECT 1"` or `"SELECT 1; BEGIN"` — the statements after that point ran outside it, so the work is not atomic. pgcommon notices, rolls back whatever transaction is open (never commits it) and raises `TxEndedInCallbackError`. When `fn` also raised, its exception is the error's `callback_error` and `__cause__`, and the message ends `(callback error: …)`. The retry helpers never retry it. Work that autocommitted after the end stays committed.

- A prepared transaction outlives the session and keeps its locks: pgcommon rolls back, with `ROLLBACK PREPARED`, the one the server shows was made of *this* transaction — however `fn` ran the `PREPARE TRANSACTION` — and never touches any other: a `PREPARE TRANSACTION` the server answered with `ROLLBACK` (the transaction was already aborted) prepared nothing, and its gid may be another session's. The error names what it rolled back, which gids exist but are not this transaction's, and anything left for you to resolve by hand (`pg_prepared_xacts`). To tell them apart, on a server with `max_prepared_transactions > 0` pgcommon takes a transaction-level shared advisory lock on a random key (`pg_try_advisory_xact_lock_shared`) in the same round trip as the transaction timeouts (one extra round trip per transaction when no timeouts are set); on a server that cannot prepare transactions (the default) it asks once per connection and then does nothing.
- **The trade-off (`Config.prepared_xact_guard`, `PG_PREPARED_XACT_GUARD`, on by default).** That marker is what lets pgcommon prove a prepared transaction is this one's. On a server with `max_prepared_transactions > 0` and no transaction timeouts or PgBouncer-mode identity set, it costs one extra round trip per transaction. Turning it off saves that round trip, and pgcommon then never runs `ROLLBACK PREPARED`: every prepared transaction `fn` made is left in place and named in the `TxEndedInCallbackError` (`Config.prepared_xact_guard is off`), for you to resolve by hand. Turn it off only if no code path can run `PREPARE TRANSACTION` inside `run_in_tx`, or if you monitor `pg_prepared_xacts` — a forgotten prepared transaction keeps its locks and holds back VACUUM until resolved.
- Not flagged: savepoints (`run_in_savepoint`, a nested `async with conn.transaction()`, `ROLLBACK TO SAVEPOINT`) and a lone `BEGIN` (PostgreSQL only warns).
- How: the pool's query logger reports every statement `fn` runs through `execute` / `fetch*` / `executemany`, with no extra round trip, and multi-statement strings are lexed by a port of Go's `sqlutil.AnalyzeTx` (comments, literals, quoted names, dollar quotes, `BEGIN ATOMIC` bodies) — only when a transaction-control word appears in them, so a large batch without one costs a substring scan, not a lexing pass. A connection `fn` leaves outside any transaction is caught too.
- **Not detected:** transaction control run through a prepared statement's own methods (`conn.prepare(...)`), a cursor or a `copy_*` method (asyncpg's query logger does not see them) when a transaction is open again by the time `fn` returns — e.g. a prepared `COMMIT` followed by a prepared `BEGIN` — and SQL the lexer reads differently from the server because `standard_conforming_strings` is off.

#### Swallowed errors: COMMIT answered by ROLLBACK

A failed statement aborts the transaction; if `fn` catches the error and returns normally, the server answers the `COMMIT` with `ROLLBACK` and the work is lost. pgcommon sends `COMMIT` itself, reads that answer and raises `TxCommitRollbackError` instead of reporting success (Go: `pgx.ErrTxCommitRollback`) — also for a failure on a path the query logger never sees (a prepared statement, cursor or `copy_*`). It is counted as a failed `COMMIT` query. Re-raise the statement's error from `fn`, or wrap an expected failure in `run_in_savepoint`.

> Detection covers transactions pgcommon owns (`run_in_tx` and the PgBouncer-mode wrapping transaction). A caller that manages its own transaction inside a standard-mode `with_conn`, swallows a failed statement and sends `COMMIT` itself is not detected (asyncpg's query logger carries no command tag); use `run_in_tx`.

#### Nested savepoints

```python
import asyncpg

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")

    try:
        await run_in_savepoint(conn, "sp_audit", audit)
    except asyncpg.PostgresError:
        pass  # the audit failed; the order insert is intact


await run_in_tx(pool, fn)
```

A savepoint recovers from a statement the **server** failed. A statement interrupted on the client side — `command_timeout` (`TimeoutError`) or a cancellation — is different: its cancel request may still be outstanding, and on a server that stopped answering `ROLLBACK TO` would wait for it (up to 5 s per cleanup statement). `run_in_savepoint` does not try: it terminates the connection at once (the server rolls back the whole outer transaction) and raises `SavepointCleanupError` (the `TimeoutError` is its `callback_error`), or lets the cancellation propagate. As in Go, where an interrupted query closes its connection, the outer transaction cannot continue after that.

On success: `RELEASE SAVEPOINT`. On exception: `ROLLBACK TO SAVEPOINT`, then `RELEASE SAVEPOINT` (so repeated failures under one name do not nest), and `fn`'s exception is re-raised. If that cleanup itself fails, `SavepointCleanupError` is raised with `fn`'s exception as `callback_error` and the cleanup failure as `__cause__`; the outer transaction is then typically aborted and must be rolled back. 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 — and `TxEndedInCallbackError`, since a callback that ended its own transaction may have committed part of its work — are raised unchanged on the first occurrence. Retries are immediate with no delay (`max_attempts` below 1 counts as 1). Both helpers take `isolation`, `read_only` and `deferrable` like `run_in_tx`:

```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 and stays a `CancelledError`, with the last attempt's error as its `__cause__` (so `is_serialization_failure` / `is_deadlock` still match it; check for cancellation first in an outer retry loop). When every attempt fails, the last attempt's error is raised with its original type and the note `pgcommon: transaction failed after N attempts`:

```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; below 1 means 3 |
| `initial_wait` | `timedelta(0)` (immediate) | Delay before the 2nd attempt |
| `max_wait` | `timedelta(0)` (no cap) | Ceiling for exponential growth; the delay saturates instead of overflowing, so set it to bound the wait meaningfully |
| `multiplier` | 1.0 (constant wait) | Set `2.0` for standard doubling backoff; below 1 (or NaN) counts as 1 |
| `jitter_fraction` | 0.0 (no jitter) | Varies delay by ±fraction; clamped to `[0, 1]` (NaN → 0) |

#### 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 (or a closed pool) is reported as `healthy=False`.

A ping must acquire a connection, so on a fully checked-out pool it would queue behind the load and time out — failing readiness exactly when the service is busiest. When the pool is **saturated**, `health()` therefore skips the ping if the server answered any statement within `saturated_alive_window` (`PG_SATURATED_ALIVE_WINDOW`, 10 seconds by default, Go's fixed `saturatedAliveWindow`), and reports `healthy=True` with `saturated=True`. That window is also how long a saturated pool can keep reporting healthy after the database stopped answering: lower it for faster detection (keep it above the gap between two statements under normal load), or set it to `off` to always ping (on a saturated pool the ping then waits for a connection, up to `health()`'s timeout). Alert on saturation separately (Go's `PgcommonPoolSaturated` alert works unchanged on the Tier 1 metrics). Use `health()` for readiness, never liveness.

#### `/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 — or, on a saturated pool, when the server answered a statement within `saturated_alive_window` (10 s by default) |
| `saturated` | `bool` | Every connection was checked out, so `healthy` was judged from the most recent server response instead of a ping |
| `total_conns` | `int` | Acquired + idle |
| `idle_conns` | `int` | Waiting in the pool |
| `acquired_conns` | `int` | Currently held by callers |
| `max_conns` | `int` | Pool maximum (the resolved `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 PostgreSQL SQLSTATE codes without matching on strings. Like Go's helpers (which unwrap with `errors.As`), these look through an exception's `__cause__` chain (`raise … from err`), the members of an `ExceptionGroup`, and the `callback_error` pgcommon's own exceptions carry — so they still match after a service, or pgcommon itself, wraps the database error. They accept `None`.

```python
from pgcommon import (
    constraint_name,
    is_check_violation,
    is_connection_lost,
    is_foreign_key_violation,
    is_not_null_violation,
    is_pg_error,
    is_pool_closed,
    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
        if is_pool_closed(exc) or is_connection_lost(exc) or not is_pg_error(exc):
            # the database, not the request, is the problem: answer 503. is_connection_lost
            # includes a server shutting down or starting up (57P01 / 57P02 / 57P03).
            raise RuntimeError("database unavailable") from exc
        raise
```

| Helper | Matches | Typical meaning |
|---|---|---|
| `is_unique_violation` · `constraint_name` | `23505` · the offending constraint name, or `""` | 409 conflict (which field) |
| `is_foreign_key_violation` · `is_not_null_violation` · `is_check_violation` | `23503` · `23502` · `23514` | 422 / 400 validation |
| `is_deadlock` · `is_serialization_failure` | `40P01` · `40001` | Retried by the retry helpers |
| `is_lock_not_available` · `is_query_canceled` | `55P03` · `57014` | `lock_timeout` / `statement_timeout` expired (or `NOWAIT`, a cancel request) |
| `is_connection_exception` · `is_insufficient_resources` | class `08` · class `53` | 503 dependency unavailable |
| `is_connection_lost` | class `08`; `57P01` admin_shutdown, `57P02` crash_shutdown, `57P03` cannot_connect_now (a restart or failover: this connection or attempt is unusable, another one shortly may work); asyncpg's `ConnectionDoesNotExistError`; a socket `ConnectionError`; or any error pgcommon raised from a connection that turned out to be closed (`CONNECTION_LOST_NOTE`, e.g. asyncpg's `InternalClientError` after a failover) | 503; never retried (the statement may have committed) |
| `is_operator_intervention` · `is_system_error` | class `57` other than `57014` (so `57P01`–`57P03` match both this and `is_connection_lost`) · class `58` | Server shutting down or crashed: 503 |
| `is_pg_error` | any PostgreSQL server error | The server answered (vs a transport failure that never reached it) |
| `sql_state` | — | The raw SQLSTATE (e.g. `"23505"`), or `""`; for codes no helper covers, or logging |
| `is_pool_closed` | `PoolClosedError`, asyncpg's closed-pool error, or a query a timed-out `drain_and_close` interrupted | Shutting down: final, do not retry |

`is_pg_error`, `sql_state` and the class helpers ignore asyncpg's client-side errors that inherit a SQLSTATE but were never sent by a server (e.g. `ConnectionDoesNotExistError`, raised when the connection is lost mid-query), as Go's do for errors that are not a `*pgconn.PgError`.

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

```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 `platform_db_slow_queries_total` Prometheus counter is incremented on every slow query regardless of the rate limit.

`duration_ms` is whole milliseconds (truncated, as Go's `elapsed.Milliseconds()`). Unlike Go's entry there is no `caller` field: asyncpg runs query loggers from the event loop after the statement finished, when the caller's stack is gone. Library-internal statements (GUC injection, `BEGIN` / `COMMIT`, savepoints, timeouts) never reach the slow-query log.

**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; the value is read from the request's `GUCSet` context.

`slow_query_threshold=None` (or `timedelta(0)`) means the 200 ms default; a **negative** value (`PG_SLOW_QUERY_THRESHOLD=off`) disables the log, as does omitting `logger`. 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.

Run migrations from a **direct** PostgreSQL connection with a DDL-privileged role, never through PgBouncer: the runner holds a session advisory lock, which transaction pooling cannot keep.

#### The `pgcommon-migrate` CLI

The `pgcommon-migrate` console script (also `python -m pgcommon.migrate`) is the port of Go's `cmd/migrate` — the same commands, environment variables, messages and exit codes, so either CLI can run a service's migrations. Use it as your init container or pre-deploy job entrypoint:

```bash
# In an init container or pre-deploy job:
MIGRATIONS_DIR=./migrations MIGRATION_DATABASE_URL='postgres://migrator:…@db:5432/app?sslmode=verify-full' \
  pgcommon-migrate up
```

| Command | Purpose |
|---|---|
| `up` | Apply all pending migrations |
| `down [N]` | Roll back `N` migrations (default 1). Fewer than `N` applied rolls back all of them and exits 0 |
| `status` | Print `version: <N> dirty: <bool>` (or `version: none (no migrations applied)` / `version: none (dirty)`); exits 3 when dirty and 4 when the version is not in `MIGRATIONS_DIR`. **Not read-only:** it creates the tracking table if missing (needs DDL rights) and waits for the advisory lock |
| `force <V>` | Record version `V` and clear the dirty flag without running anything (`-1` = no version) — after repairing a failed migration by hand. `V` must be a migration in `MIGRATIONS_DIR` |
| `version` / `--version` | Print the package version |
| `-h` / `--help` | Print the usage and exit 0 (a Python extra; Go's CLI has no help). A bare run prints the usage and exits 1 |

Configuration is environment-only:

| Variable | Default | Purpose |
|---|---|---|
| `MIGRATION_DATABASE_URL` | `DATABASE_URL` / `PG_*` | DSN for migrations: a DDL role, **direct** to Postgres. The fallback is refused when `PG_BOUNCER_MODE` is true (it then addresses PgBouncer) — set `MIGRATION_DATABASE_URL`. An insecure or missing `sslmode` is warned about as for `DATABASE_URL` |
| `MIGRATIONS_DIR` | — (**required**) | Directory of `{version}_{title}.up.sql` / `.down.sql` files at its top level |
| `MIGRATIONS_TABLE` | `pgcommon_migrations` | Tracking table (one per migration set) |
| `MIGRATIONS_LOCK_TIMEOUT` | `30s` | Session `lock_timeout` (a Go duration): the wait for another runner's advisory lock and for the table locks your DDL needs. `0` = 30 s, negative = the server/role setting, at most INT_MAX ms (`596h31m23.647s`). An unparseable value is an error |

Exit codes: **0** success (including "no change"); **1** usage, configuration, connection or migration error; **3** the schema is dirty; **4** the database's version is not in `MIGRATIONS_DIR` (an image rollback, the wrong directory or `MIGRATIONS_TABLE`). Errors are printed as `migration error: …`, configuration warnings as `config warning: <key>: <reason>`. Progress lines (`applied migration version=N direction=up`) and runner warnings (`warning: …`, e.g. an ignored unknown `x-` DSN option) go to stderr too, so stdout carries only the command's result.

SIGINT / SIGTERM stop the run after the migration that is executing (and abandon a connection attempt or lock wait); further signals are absorbed, so a second Ctrl-C cannot cut a migration's DDL short. A stopped run exits 1, unless its last pending migration had already been applied.

The same CLI ships as a container image for init containers and deploy jobs — see [Docker → Running a migration with the image](#running-a-migration-with-the-image).

#### The `Runner` API

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

```python
import importlib.resources

from pgcommon import migration_dsn_from_env
from pgcommon.migrate import MigrationDirtyError, Runner, VersionNotInSourceError

runner = Runner(
    migration_dsn_from_env(),  # MIGRATION_DATABASE_URL, else the app DSN
    importlib.resources.files("myservice") / "migrations",  # or a directory path
    logger=logger,  # logs each step
)

try:
    await runner.up()  # apply all pending migrations
except MigrationDirtyError:
    # A migration failed mid-run: repair the schema by hand, then
    # await runner.force(last_good_version)  (CLI: pgcommon-migrate force <V>)
    raise
except VersionNotInSourceError:
    # The database is at a version this build has no migration for:
    # an image rollback, or the wrong source / table.
    raise

version, dirty = await runner.version()  # (None, False) when nothing is applied
await runner.down(1)  # roll back one step
await runner.steps(2)  # apply exactly two steps forward (or -2 to roll back two)
```

| `Runner` argument / method | Behaviour |
|---|---|
| `source` | A directory path, or any `importlib.resources` `Traversable` (Go's `embed.FS`). Only its top level is read; files follow golang-migrate's `{version}_{title}.up.{ext}` / `.down.{ext}` naming. No migration file at the top level is `NoMigrationsFoundError` (naming a subdirectory that holds some, or a misnamed file); two files for one version and direction are `DuplicateMigrationError` |
| `table=` | Tracking table, default `DEFAULT_MIGRATIONS_TABLE` (`pgcommon_migrations`). Every migration set in a database needs its own. An `x-migrations-table` DSN option wins. Quoted, so case-sensitive (0.2.0 folded it to lower case: see [Upgrading from 0.2.0](#migration-runner-and-cli)) |
| `lock_timeout=` | Default `DEFAULT_LOCK_TIMEOUT` (30 s); negative = the server/role setting; rounded up to whole milliseconds; above INT_MAX ms is `TimeoutOutOfRangeError`. A `lock_timeout` in the DSN (parameter, or `-c lock_timeout=…` in `options`) wins |
| connect timeout | `DEFAULT_CONNECT_TIMEOUT` (10 s) when neither the DSN nor `PGCONNECT_TIMEOUT` sets `connect_timeout`; an explicit `0` waits indefinitely. Parsed as the pool parses it (whole seconds, at least 2; negative or non-integer is an error) — only the unset default differs: the pool sets none |
| `stop=` | An `asyncio.Event` accepted by `up` / `down` / `steps` / `version` / `force`. Once set, the run ends before the next migration with `MigrationStoppedError` (`migrations: stopped at version <N|none>[ (dirty)]: stop requested`, or plain `stop requested` before the database was reached, while connecting or waiting for the lock); a migration is never interrupted. Cancelling the task likewise lets the executing migration finish |
| `up()` · `down(n)` · `steps(±n)` | Raise `MigrationDirtyError` for a dirty schema and `VersionNotInSourceError` when the recorded version has no migration in the source. Fewer than `n` to run applies what exists and raises `ShortLimitError`. `steps(0)` only checks the recorded version against the source |
| `version()` | `(version, dirty)`; `version` is `None` when nothing is recorded. `(None, True)` means a down of the first migration failed: repair, then `force(-1)`. Raises `VersionNotInSourceError` (carrying `.version` / `.dirty`). Not read-only, like `status` |
| `force(v)` | Record `v` (`-1` = none) clean without running anything. A `v` with no migration in the source is `VersionNotInSourceError` |

The DSN may be a URL or a keyword/value string. golang-migrate's `x-migrations-table`, `x-migrations-table-quoted`, `x-statement-timeout` (ms), `x-multi-statement` and `x-multi-statement-max-size` options are honoured; every `x-` option and every `pool_*` parameter is stripped before connecting (an unknown `x-` option is logged and ignored). Errors that would quote the DSN password are sanitized.

**State compatible with golang-migrate and the Go runner.** The tracking table is `"<schema>"."<table>" (version bigint not null primary key, dirty boolean not null)` in `current_schema()` — no row means "no version". Each migration records its target version as dirty, runs its body (not wrapped in a transaction, so a single-statement file can use `CREATE INDEX CONCURRENTLY`), then records it clean, so a failure leaves the schema dirty at the version being migrated to. The run holds golang-migrate's session advisory lock (keyed by database, schema and table), so concurrent runners — Python or Go — serialise. The default table is deliberately *not* `schema_migrations`, so this runner never collides with a consuming service's own golang-migrate-tracked table; pass `table="schema_migrations"` to adopt one.

**Rolling back an image.** Once a newer release has migrated, an **older** image's `up()` fails with `VersionNotInSourceError` (CLI exit 4) instead of reporting "no changes", so a service that runs migrations at startup refuses to start the older image. Run `down` with the **newer** image's migrations first, or run migrations as a separate job.

> **Coming from 0.2.0?** The runner's state, advisory-lock key and table quoting changed. Read [Upgrading from 0.2.0 → Migration runner and CLI](#migration-runner-and-cli) before the first 0.3.0 run.

### 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 acquire (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. The wrapping transaction also gets `statement_timeout` / `lock_timeout` (see [Statement and lock timeouts](#statement-and-lock-timeouts)).

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

> **`with_conn` in PgBouncer mode:** when an identity is active and `pgbouncer_mode` is `True`, the connection passed to `fn` is already inside an open transaction (used to scope GUCs as `is_local=true`). **Never commit, roll back or begin a transaction in `fn`** in this mode: statements after that point would run without the RLS GUCs. pgcommon detects it (as for `run_in_tx` — see [Do not end the transaction in the callback](#do-not-end-the-transaction-in-the-callback)), rolls back whatever is open and raises `TxEndedInCallbackError`. A nested `conn.transaction()` is a `SAVEPOINT`, which is allowed but not a fresh top-level transaction. Use `run_in_tx` / `with_tx` for transactional work.

### Failure model

| Failure | Behaviour |
|---|---|
| Invalid configuration / unparseable DSN | `Pool.create` raises the matching sentinel (`DSNRequiredError`, `InvalidDSNError`, `NegativeDurationError`, `TimeoutOutOfRangeError`, …) before connecting |
| Unreachable host / failed initial ping | `Pool.create` raises `ConnectionError` immediately — the process should not start |
| Connection error during `with_conn` / `run_in_tx` | The callback's own `asyncpg` error is raised to the caller (nothing is rolled back on the lost session; a callback that swallowed it gets the failed COMMIT's error); the connection is not returned to the idle pool |
| Server ended the session before the callback started | The connection is replaced on acquire — GUC injection, `BEGIN` and the transaction's setup, or the idle ping found it dead — and another one taken, for every dead connection, within `acquire_timeout`; including asyncpg's `InternalClientError` ("cannot switch to state …") on a connection whose session ended just before (a failover). Logged at Info; the failed statement is not a query error (the caller never sees it) |
| Connection died while being released | Never reported to the caller: the callback's result (or its own error) is returned unchanged, the connection terminated and the failure logged at `WARN` |
| Server stops answering (frozen host or process) | TCP keepalive does not notice it; `command_timeout` (or the caller's own timeout) bounds a statement in flight, the connection is then terminated rather than waited for, cleanup is bounded to 5 s, and new calls fail with `AcquireTimeoutError` after `acquire_timeout` — also when `command_timeout` interrupts pgcommon's own setup before the callback (GUC injection, `BEGIN`): that is part of acquiring a usable connection |
| Statement interrupted by `command_timeout` | asyncpg's `TimeoutError` is raised; pgcommon terminates the connection rather than wait for the cancel, but the error does **not** match `is_connection_lost` — it is a slow statement, not a lost connection |
| Connection lost while held (failover, restart, network) | The error raised matches `is_connection_lost` (asyncpg reports it as `ConnectionDoesNotExistError`, `InterfaceError` or `InternalClientError`); the lost statement is counted as a failed query (operation `other`) when asyncpg never reported it; the slot always goes back to the pool, and the pool is topped back up to `min_conns` about 0.5 s after a connection closes |
| Invalid identity (`GUCSet.validate()` fails) | Raised (`IDENTITY_ERRORS`) before any connection is waited for; never reaches the server; not an acquire error |
| Callback ended its transaction / swallowed a failed statement | Rolled back; `TxEndedInCallbackError` / `TxCommitRollbackError` — never reported as success |
| Statement / lock timeout inside a transaction | The statement fails with `57014` / `55P03` (`is_query_canceled` / `is_lock_not_available`); the transaction is rolled back |
| 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` waits for a connection up to `acquire_timeout` (30 s by default), then raises `AcquireTimeoutError` (a `TimeoutError`, counted as an acquire error) |
| `pool.close()` / `aclose()` / `drain_and_close()` called | All subsequent `with_conn`/`run_in_tx`/`ping` calls raise `PoolClosedError` immediately; `health()` never raises — it reports `healthy=False` |
| `drain_and_close()` timeout expires | In-flight connections are interrupted — their running statements cancelled on the server — (their errors match `is_pool_closed`), the pool closes and `DrainTimeoutError` is raised within about `timeout` + 1 s |
| 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`). The one exception to "original": an error that would expose the DSN password anywhere in its chain (message, notes, `__cause__`, `__context__`, group members) is replaced in the chain by a redacted stand-in; PostgreSQL server errors are always kept, so `sql_state` and the `is_*` helpers still work.

### 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 |
| Transaction setup | Once per transaction, when `statement_timeout` / `lock_timeout` is set, in PgBouncer mode with an identity, or when the server can prepare transactions (`max_prepared_transactions` > 0: the prepared-transaction marker, unless `prepared_xact_guard` is off) | One round trip: the transaction-local GUCs, timeouts and marker share a single statement | Right after `BEGIN` |
| GUC injection (PgBouncer mode) | Once per `with_conn`/`run_in_tx` | Part of the transaction setup statement | 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 |

**The per-statement in-process work above is a few microseconds; the cost worth budgeting for is per transaction.** `run_in_tx` sends `BEGIN`, at most one setup statement (transaction-local GUCs, timeouts, prepared-transaction marker) and `COMMIT` around your work: measured manually on a laptop against PostgreSQL 16 in Docker, a `run_in_tx` around a single `SELECT 1` took about 18% longer than a raw asyncpg transaction, more with GUCs or in PgBouncer mode, where every `with_conn` with an identity is wrapped in such a transaction. For real OLTP transactions (several statements, 1–50 ms) it is a small fraction. See [ARCHITECTURE.md](ARCHITECTURE.md#performance-characteristics).

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 `platform_db_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 interrupts them at the deadline
try:
    await pool.drain_and_close(timeout=30)
except DrainTimeoutError:
    logger.warning("pool drain timed out")
    # connections were interrupted and the pool is closed; safe to exit
```

`await pool.aclose()` waits for callbacks too, but without any limit — one stuck query blocks shutdown forever.

---

#### ❌ Ending or swallowing inside `run_in_tx`

`run_in_tx` owns the transaction. Committing it yourself, or catching a failed statement and carrying on, means the work is not atomic — pgcommon rolls back and raises instead of reporting success.

```python
# ❌ WRONG: commits half-way (TxEndedInCallbackError), or swallows a failure that aborted the
# transaction (TxCommitRollbackError)
async def fn(conn):
    await conn.execute("INSERT INTO orders ...")
    await conn.execute("COMMIT")
    try:
        await conn.execute("INSERT INTO audit_log ...")
    except asyncpg.PostgresError:
        pass
```

```python
# ✅ CORRECT: let pgcommon commit; isolate an expected failure in a savepoint
async def fn(conn):
    await conn.execute("INSERT INTO orders ...")
    try:
        await run_in_savepoint(conn, "sp_audit", lambda c: c.execute("INSERT INTO audit_log ..."))
    except asyncpg.PostgresError:
        pass  # rolled back to the savepoint; the transaction is intact
```

---

#### ❌ Ignoring `PoolClosedError`

After `pool.close()`, `aclose()` or `drain_and_close()`, every subsequent `with_conn`/`run_in_tx`/`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 Exception as exc:
        if is_pool_closed(exc):
            return  # pool is shutting down — stop retrying
        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

Every field but `dsn` is optional. For the pool settings a DSN can carry, the precedence is **explicit `Config` field > DSN `pool_*` parameter > default**; `None` and `0` / `timedelta(0)` both mean "not set", as in Go.

| `Config` field | Env var | Default | Notes |
|---|---|---|---|
| `dsn` | `DATABASE_URL` | — | Required. A URL or a libpq keyword/value string. Its `pool_max_conns`, `pool_min_conns`, `pool_max_conn_lifetime`, `pool_max_conn_lifetime_jitter`, `pool_max_conn_idle_time`, `pool_health_check_period` and `connect_timeout` parameters configure the pool and are never sent to the server |
| `max_conns` | `PG_MAX_CONNS` | `10` | Pool maximum connections |
| `min_conns` | `PG_MIN_CONNS` | `2` capped at `max_conns`; `0` in PgBouncer mode | Minimum connections kept open. An explicit `0` is honoured (no idle connections) |
| `max_conn_lifetime` | `PG_MAX_CONN_LIFETIME` | `1h` | Max age of each connection; it is closed when released, or acquired from idle, past its own deadline |
| `max_conn_lifetime_jitter` | `PG_MAX_CONN_LIFETIME_JITTER` | `max_conn_lifetime / 10` | Random `0…jitter` added to each connection's lifetime, so connections opened together do not reconnect at once. Negative (`off`), or `pool_max_conn_lifetime_jitter=0` in the DSN, disables it |
| `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` (with a `logger`) | Log queries exceeding this duration. Negative (`off`) disables |
| `statement_timeout` | `PG_STATEMENT_TIMEOUT` | unset (server/role setting) | Applied transaction-locally to every `run_in_tx` and PgBouncer-mode `with_conn` wrapping transaction; rounded up to whole ms; above INT_MAX ms is `TimeoutOutOfRangeError` |
| `lock_timeout` | `PG_LOCK_TIMEOUT` | unset (server/role setting) | As `statement_timeout` |
| `logger` | — | `None` | Implements the `pgcommon.Logger` protocol — use `StdlibLogger` (fields appended to the message as ` key=value`, so the default formatter shows them, and passed as `extra={"pgcommon_fields": …}`) or `StructlogLogger` |
| `tracer` | — | `None` | An `opentelemetry.trace.Tracer` |
| `query_logger` | — | `None` | An extra asyncpg query logger; may implement `AcquireTracer` / `ReleaseTracer` (see [Observability](#observability--prometheus-and-otel)) |
| `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 the full SQL (`db.query.text`) and error messages in OTel spans, and the full SQL in the slow-query log; the default records the verb and SQLSTATE only, to prevent PII leakage |
| `pool_name` | `PG_POOL_NAME` | `"default"` | The `pool` label on every metric. Must match `^[a-z][a-z0-9_]{0,31}$` (`InvalidPoolNameError`); empty means `"default"`. Read verbatim from the env, as Go does |
| `acquire_timeout` | `PG_ACQUIRE_TIMEOUT` | `30s` | Bounds every acquire — waiting for a free connection, dialling, GUC injection, `BEGIN` and the transaction's setup, and their retries on another connection — then `AcquireTimeoutError`. Negative (`off`) waits without limit. A Python addition: Go bounds the acquire with the request's context deadline |
| `command_timeout` | `PG_COMMAND_TIMEOUT` | unset (no limit) | asyncpg's client-side limit on every statement (`TimeoutError`, not `is_connection_lost`; the connection is then terminated). On pgcommon's own setup before the callback it is retried on another connection, then `AcquireTimeoutError`. The bound for a server that stops answering — set it above `statement_timeout`, which the server enforces itself. A Python addition: Go bounds statements with the request's context |
| `saturated_alive_window` | `PG_SATURATED_ALIVE_WINDOW` | `10s` | How recent a server response lets `health()` report a saturated pool healthy without a ping (Go: `saturatedAliveWindow`); negative (`off`): always ping |
| `prepared_xact_guard` | `PG_PREPARED_XACT_GUARD` | `True` | Marks each transaction so a prepared transaction the callback made of it can be proven pgcommon's and rolled back; off saves a round trip per transaction on servers with `max_prepared_transactions > 0`, and pgcommon then never runs `ROLLBACK PREPARED` (see [Do not end the transaction in the callback](#do-not-end-the-transaction-in-the-callback)). A Python addition |

`Pool.create` rejects a negative count (`InvalidMaxConnsError` / `InvalidMinConnsError`), a negative lifetime, idle time, health-check period or `command_timeout` in the `Config`, or a negative `pool_*` duration in the DSN (`NegativeDurationError`), and `min_conns > max_conns` after the DSN values and defaults are applied (`MinConnsExceedsMaxError`). `Config.validate()` runs the checks that need no DSN parsing.

#### 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 |
| `statement_timeout` | a bound above your slowest legitimate transaction | Stops a runaway query from holding a connection (and its locks) indefinitely; back it with `ALTER ROLE … SET statement_timeout` for `with_conn` work outside a transaction |
| `command_timeout` (`PG_COMMAND_TIMEOUT`) | set it, above `statement_timeout` (e.g. `statement_timeout` + 5 s) | Unset by default, and then nothing bounds a statement on a server that stops answering: a frozen or paused database (VM pause, storage stall, `SIGSTOP`) still acknowledges at the TCP level, so TCP keepalive never fires and every in-flight query waits until it answers again. Above `statement_timeout`, the server's own limit fires first for a merely slow query. Meanwhile `health()` on a saturated pool can keep reporting healthy for up to `saturated_alive_window` (10 s) |
| `acquire_timeout` (`PG_ACQUIRE_TIMEOUT`) | `30s` (default), or below your request deadline | Bounds waiting for a connection when the pool is exhausted or the server unreachable |
| `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.

Every field can also come from the environment: see [Environment variables](#environment-variables).

---

## Upgrading from 0.2.0

0.3.0 brings the port in line with Go `platform-pgcommon` **v2.0.1** and is breaking in several places (while the package is `0.y.z`, a MINOR bump may break the public API). The authoritative list is [CHANGELOG § Upgrade notes](CHANGELOG.md#upgrade-notes); this is the checklist. `platform-eventcommon` pins `pgcommon>=0.2.0,<0.3.0` and needs a coordinated change.

| Change | What to check |
|---|---|
| Only the Tier 1 `platform_db_*` metrics remain; the `pgcommon_*` legacy metrics and `metrics.init` / `init_with_registry` / `Metrics` / `current()` are gone ([CHANGELOG](CHANGELOG.md#removed--breaking-legacy-metrics-with-no-compatibility-period)) | Call `pgcommon.metrics.init_with_identity(Identity(domain=…, service=…, environment=…))` before `Pool.create`; update dashboards and queries to the `platform_db_*` names |
| `run_in_tx` (and a PgBouncer-mode `with_conn` with an identity) raise `TxEndedInCallbackError` / `TxCommitRollbackError` where they used to report success | Callbacks that run `COMMIT` / `ROLLBACK` themselves, or catch a failed statement and carry on |
| `Config` sizing and duration fields default to `None` (DSN `pool_*`, else the default); `timedelta(0)` means "default"; `max_conn_lifetime_jitter` defaults to lifetime ÷ 10 | `slow_query_threshold=timedelta(0)` no longer disables the log (use a negative value); a DSN carrying `pool_max_conns` now sizes the pool |
| Every acquire is bounded by `acquire_timeout` / `PG_ACQUIRE_TIMEOUT` (default 30 s) → `AcquireTimeoutError` | Requests that used to wait forever on an exhausted pool now fail after 30 s |
| `PG_STATEMENT_TIMEOUT` / `PG_LOCK_TIMEOUT` also apply inside a PgBouncer-mode `with_conn`'s wrapping transaction | Long statements through PgBouncer-mode `with_conn` are now cancelled (`57014`) |
| `drain_and_close` raises `DrainTimeoutError` when it had to interrupt connections; its message counts acquired connections and waiting callers separately | Shutdown code that relied on a silent return; log matchers |
| An invalid `GUCSet` is rejected before any connection is waited for, in PgBouncer mode too (`pgcommon: invalid GUC identity: …`) | Middleware that relied on a late failure |
| `is_connection_lost` also matches `57P01` / `57P02` / `57P03` | Error mapping that treated a server restart as a query error |
| Releasing a connection never fails the call; a dead connection replaced during acquire is logged at Info and not counted as a query error | Query error ratios drop to their true values |
| A statement interrupted on the client side (`command_timeout`, cancellation) terminates its connection; `run_in_savepoint` then raises `SavepointCleanupError` instead of attempting cleanup | Code that continued an outer transaction after a timed-out savepoint |
| Idle connections are closed by the health check, never below `min_conns` | — (no more redial churn every `max_conn_idle_time`) |
| `connect_timeout` is parsed the same way by the pool and the runner: a negative or non-integer value is `InvalidDSNError`; `PGCONNECT_TIMEOUT` is honoured | DSNs with `connect_timeout=-1` |
| `PG_BOUNCER_MODE` accepts Go's `strconv.ParseBool` spellings again and warns on anything else | — |
| `StdlibLogger` appends the fields to the message as ` key=value` | Handlers or tests that match exact message text |
| Slow-query logs read `tenant_id` only from the request's `GUCSet` context | Services that set `log_tenant_id=True` but supply the identity only through `guc_provider` |
| `asyncpg>=0.30,<0.32` (0.29 dropped) | Your lock file |

### Migration runner and CLI

`pgcommon-migrate` now has Go's command line ([CHANGELOG](CHANGELOG.md#changed--breaking-migrations-cli)): configuration is environment-only, `MIGRATIONS_DIR` is **required**, `--dsn` / `--source` / `steps` are gone, the old `version` command is now `status`, and exit codes are 0 / 1 / 3 / 4. Update init containers and CI jobs. The runner reads and writes golang-migrate's state ([CHANGELOG](CHANGELOG.md#changed--breaking-migration-runner)):

- **Stop every 0.2.0 runner before the first 0.3.0 run, and never mix them.** 0.2.0 locked one fixed advisory-lock key; 0.3.0 locks golang-migrate's per-table key, so a 0.2.0 and a 0.3.0 runner do **not** serialise: both can apply the same migration at once, and they disagree on how a failure or a full rollback is recorded. Roll the new version out to every job and init container that runs migrations (or pause them) before any of them runs `up`.
- **A mixed-case `table=` now refuses to run.** 0.2.0 did not quote the table name, so PostgreSQL folded `Runner(table="AppMigrations")` to `appmigrations`; the name is quoted now (as golang-migrate does). When `"AppMigrations"` does not exist but `appmigrations` does, every operation raises `InvalidMigrationsTableError` naming both, rather than creating a second, empty table and re-applying every migration. Either rename the table (`ALTER TABLE "<schema>"."appmigrations" RENAME TO "AppMigrations"`) or pass the lower-case name (`table="appmigrations"` / `MIGRATIONS_TABLE=appmigrations`). Lower-case names (the default `pgcommon_migrations` included) are unaffected.
- 0.2.0 recorded "everything rolled back" as version `0`; the runner now reads that as a version missing from the source (exit 4) — run `pgcommon-migrate force -1` once. (A source with a version-`0` migration: 0.2.0 never applied it, since it read `0` as "none", while 0.3.0 reads a recorded `0` as "migration 0 applied" — check, and `force -1` then `up` if it must run.)
- A dirty 0.2.0 table names the failing migration's own version (also on the way down), not golang-migrate's target version: repair the schema, then `force` the last fully applied version.
- The table itself is read as is: 0.2.0's `(version bigint PRIMARY KEY, dirty boolean NOT NULL DEFAULT false)` works unchanged.

---

## Local development

### Prerequisites

- Python ≥ 3.12 (CI runs 3.12, 3.13 and 3.14; `uv python install 3.12` if you have none)
- [uv](https://docs.astral.sh/uv/) — the environment is `uv sync` from `uv.lock`
- Docker — testcontainers for the integration and e2e suites, and `make docker-up` for a local Postgres
- Optional: `hadolint` and `trivy` for the image gates (`brew install hadolint trivy`); `gh` and `cosign` ≥ 2.4 to verify a release

### Setup

```bash
git clone https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
cd platform-pgcommon-py
make setup       # uv sync (the dev group includes the fastapi / structlog extras) + copy .env-example → .env + git hooks
make docker-up   # local Postgres on :5432 (user, password and database: test); make docker-up-all adds PgBouncer on :6432
make ci          # everything CI's validate jobs run, in the same order (Docker required for the test suites)
```

### Common commands

| Command | Description |
|---|---|
| `make setup` / `make install` | `uv sync`, copy `.env-example` to `.env` if missing (never overwrites) and `make install-hooks` / `uv sync` only |
| `make install-hooks` | Point git at `.githooks/` (`core.hooksPath`): the pre-commit hook runs `ruff format --check` + `ruff check` on staged Python files, `docs-check` when the architecture docs are staged, `lock-check` when `pyproject.toml` / `uv.lock` are (`git commit --no-verify` skips it) |
| `make pin-base-images` / `make docs-serve` | Re-resolve the Dockerfile's `FROM <image>:<tag>` lines to their current `@sha256:` digests and rewrite them, printing what changed / browse the API docs with `pydoc` on `PYDOC_PORT` (8080) |
| `make format` / `make format-check` | `ruff format` + `ruff check --fix` (rewrites files) / `ruff format --check` (CI; never modifies files). Aliases `fmt` / `fmt-check` |
| `make lint` / `make typecheck` | `ruff check` / `mypy --strict` |
| `make lock-check` / `make pip-check` / `make vulncheck` | `uv lock --check` (the lock matches `pyproject.toml`) / `uv pip check` (the `.venv` is consistent) / `pip-audit` (alias `vuln-check`) |
| `make arch-lint` | The import-linter contracts (the same script CI runs) |
| `make metrics-lint` / `make metrics-doc` | Observability-standard conformance + inventory in sync / regenerate `docs/observability/metrics-registry.md` |
| `make docs-check` | Every `docs/architecture/mermaid/*.mmd` embedded byte-identically in `ARCHITECTURE.md` |
| `make ci-scripts-test` | Regression tests for the CI shell scripts (`detect-changes.sh`) |
| `make test-unit` · `make test-int` · `make test-e2e` · `make test` | Unit (no Docker) · integration (Postgres; alias `test-integration`) · e2e (Postgres + PgBouncer) · all three in one run (alias `test-all`) |
| `make test-ci` / `make cover-check` / `make cover` / `make cover-func` | Every suite with coverage, no threshold (CI's test gate) / the same + the ≥ 97% gate / `test` + an HTML report in `htmlcov/` / per-file report with missing lines from the last run's `.coverage` |
| `make build` / `make build-constraints` | Wheel + sdist into `dist/`, `twine check --strict`, and an install check of the `pgcommon-migrate` console script / re-lock the build backend with hashes |
| `make api-compat` / `make release-guard TAG=vX.Y.Z` | griffe public-API diff against the last release tag / pre-tag check (package version = tag, CHANGELOG section exists) |
| `make docker-build` / `make lint-docker` / `make scan` / `make smoke-test` | Build the CLI image `pgcommon-migrate:local` / hadolint / Trivy, CI's severities / build + CI's `smoke-tests.sh` (alias `test-smoke-image`) |
| `make docker-up` / `make docker-up-all` / `make docker-down` / `make docker-logs` | Postgres / Postgres + PgBouncer / stop and remove both, volumes included / tail their logs. Host ports: `POSTGRES_HOST_PORT` (5432) / `PGBOUNCER_HOST_PORT` (6432) when another stack holds them |
| `make pgbouncer-up` / `make pgbouncer-down` / `make docker-reset` / `make docker-ps` | Add PgBouncer to a running stack / stop PgBouncer only / `docker-down` + `docker-up` (an empty database) / list the stack's containers |
| `make migrate-up` / `make migrate-down` / `make migrate-status` | `pgcommon-migrate up` / `down $MIGRATE_DOWN_STEPS` (1) / `status` with `MIGRATIONS_DIR` (`migrations`) against the local Postgres (`MIGRATION_DATABASE_URL` overrides) |
| `make rls-check` | The RLS tests in both modes (`tests/integration/test_rls_guc_injection.py`, `tests/e2e/test_rls_pgbouncer_e2e.py`): a non-superuser never sees or writes another tenant's rows (Docker; own containers) |
| `make psql` / `make show-config` | psql in the local Postgres container / `config_from_env()` with `.env` loaded, DSN password redacted, plus config warnings |
| `make ci` | lock-check + format-check + lint + typecheck + arch-lint + metrics-lint + docs-check + ci-scripts-test → test-ci → coverage gate → build |
| `make clean` / `make help` | Remove `.venv`, caches, build artefacts and coverage output / list every target |

### Running a single test

```bash
uv run pytest tests/unit/test_pool.py -k test_with_conn_raises_when_closed -v --no-cov
uv run pytest tests/integration -k test_gucset_scopes_queries_to_own_tenant -v -m integration --no-cov
uv run pytest tests/e2e/test_rls_pgbouncer_e2e.py -v -m e2e --no-cov
```

`--no-cov` matters: `pyproject.toml`'s pytest `addopts` collect coverage and enforce the 97% `fail_under` on every run, so a subset of the suite fails the run on coverage even when its tests pass.

### Developer tools

The migrate CLI runs against the local database with the example migration in `migrations/`:

```bash
make docker-up
export MIGRATIONS_DIR=./migrations MIGRATION_DATABASE_URL='postgres://test:test@localhost:5432/test?sslmode=disable'
uv run pgcommon-migrate up       # then: status · down · force <V>; or python -m pgcommon.migrate
```

It prints a `config warning: MIGRATION_DATABASE_URL: sslmode=disable …` line first — expected against the local container, which has no TLS. `uv run python` gives a REPL with the package installed in editable mode, for trying the API against the same database.

---

## Testing RLS locally

The fastest way to see the identity contract working, or to debug a consumer's policy, is to reproduce it in `psql`, then through pgcommon. pgcommon does exactly what these statements do: `set_config(…, false)` on every acquire in direct mode, asyncpg's `RESET ALL` on release, and `set_config(…, true)` inside a transaction in PgBouncer mode. The repository's own isolation tests build the same fixture: a non-superuser role and `FORCE ROW LEVEL SECURITY` ([tests/integration/test_rls_guc_injection.py](tests/integration/test_rls_guc_injection.py), [tests/e2e/conftest.py](tests/e2e/conftest.py)).

### How it works

```
request (tenant A) ── acquire ── set_config('app.tenant_id', A, false) ── queries see A's rows ── release: RESET ALL
job (no identity)  ── acquire (same connection) ── app.tenant_id is '' ── queries see 0 rows
PgBouncer mode     ── BEGIN ── set_config('app.tenant_id', A, true) ── queries see A's rows ── COMMIT: gone
```

### Step 1 — Start Postgres

```bash
make docker-up
docker compose exec postgres psql -U test -d test
```

### Step 2 — Create a tenant-scoped table and a non-superuser role

The compose user `test` is a superuser, and superusers bypass RLS even with `FORCE` (so does a `BYPASSRLS` role, and the table owner without `FORCE`), so test as a plain role:

```sql
CREATE ROLE app_user LOGIN PASSWORD 'app' NOSUPERUSER;
CREATE TABLE orders (id serial PRIMARY KEY, tenant_id uuid NOT NULL, amount int NOT NULL);
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE orders FORCE  ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON orders
  USING      (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')::uuid)
  WITH CHECK (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')::uuid);
GRANT SELECT, INSERT ON orders TO app_user;
GRANT USAGE ON SEQUENCE orders_id_seq TO app_user;
SET ROLE app_user;
```

### Step 3 — Act as two tenants, then as nobody

```sql
SELECT set_config('app.tenant_id', '11111111-1111-1111-1111-111111111111', false);
INSERT INTO orders (tenant_id, amount) VALUES ('11111111-1111-1111-1111-111111111111', 10);

SELECT set_config('app.tenant_id', '22222222-2222-2222-2222-222222222222', false);
INSERT INTO orders (tenant_id, amount) VALUES ('22222222-2222-2222-2222-222222222222', 20);
SELECT count(*) FROM orders;                    -- 1: only tenant B's row

RESET ALL;                                      -- what asyncpg's reset does on release
SELECT current_setting('app.tenant_id', true);  -- '' (not NULL): hence NULLIF in the policy
SELECT count(*) FROM orders;                    -- 0: no identity, no rows (not an error)

SELECT set_config('app.tenant_id', '11111111-1111-1111-1111-111111111111', false);
INSERT INTO orders (tenant_id, amount) VALUES ('22222222-2222-2222-2222-222222222222', 1);
-- ERROR: new row violates row-level security policy for table "orders"
```

### Step 4 — Repeat through pgcommon

With the rows from step 3 in place, save this as `rls_demo.py` and run `uv run python rls_demo.py` from the repository root. `max_conns=1` makes both calls share one backend connection, so the second call proves the first caller's identity did not survive the release:

```python
import asyncio

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

TENANT_A = "11111111-1111-1111-1111-111111111111"
COUNT = "SELECT count(*) FROM orders"


async def main() -> None:
    pool = await Pool.create(
        Config(
            dsn="postgres://app_user:app@localhost:5432/test",
            max_conns=1,
            guc_provider=guc_set_from_context,
        )
    )
    try:
        with with_guc_set(GUCSet(user_id="u1", tenant_id=TENANT_A)):
            print("tenant A:", await pool.with_conn(lambda conn: conn.fetchval(COUNT)))  # 1
        print("no identity:", await pool.with_conn(lambda conn: conn.fetchval(COUNT)))  # 0
    finally:
        await pool.drain_and_close()


asyncio.run(main())
```

### Step 5 — Repeat through PgBouncer

`make docker-up-all` starts PgBouncer in **transaction** pooling mode (host port `6432`). Its user list is generated from its own `DATABASE_URL` when it starts, so it knows only `test` — the reason the e2e suite creates its role *before* starting PgBouncer. Connect as `test` and drop to `app_user` per transaction with `SET LOCAL ROLE`:

```bash
make docker-up-all
docker compose exec postgres psql "postgres://test:test@pgbouncer:5432/test"
```

```sql
BEGIN;
SET LOCAL ROLE app_user;
SELECT set_config('app.tenant_id', '11111111-1111-1111-1111-111111111111', true);
SELECT count(*) FROM orders;   -- 1
COMMIT;

BEGIN;
SET LOCAL ROLE app_user;
SELECT count(*) FROM orders;   -- 0: the transaction-local identity ended at COMMIT
COMMIT;
```

After `COMMIT`, the next transaction — possibly on a different backend — sees no identity. That is why PgBouncer mode sets the identity per transaction (`is_local=true`) and never at session level.

### Troubleshooting RLS

| Symptom | Likely cause | Fix |
|---|---|---|
| Every tenant sees every row | Connecting as a superuser (the compose `test` user is one), the table owner without `FORCE`, or a `BYPASSRLS` role | Use a plain application role; `FORCE ROW LEVEL SECURITY` |
| `invalid input syntax for type uuid: ""` | The policy casts `current_setting(…)::uuid` without `NULLIF`; after asyncpg's `RESET ALL` the GUC is `''`, not NULL | `NULLIF(current_setting('app.tenant_id', true), '')::uuid` |
| `unrecognized configuration parameter "app.tenant_id"` | `current_setting` without the `true` (missing-ok) argument | `current_setting('app.tenant_id', true)` |
| Queries return 0 rows for a real tenant | No identity on the context (middleware order), `guc_provider` unset, the `with_guc_set` block exited before the query ran, or the work runs in a task created outside the request | Log `guc_set_from_context()` in the repository; check the middleware wraps the handler |
| A long-lived worker sees one tenant's rows forever | It was started inside a request, so it copied that request's context | Start it from the app's lifespan, or with `asyncio.create_task(worker(), context=contextvars.Context())` |
| A tenant's identity shows up on a later request behind PgBouncer | Application code ran a session-level `SET app.…` / `set_config(…, false)`, or `inject_gucs` on a PgBouncer connection | `SET LOCAL` inside `run_in_tx`; `inject_gucs` only on direct connections |
| `TxEndedInCallbackError` | A transaction begun, committed or rolled back inside a PgBouncer-mode `with_conn`, or inside a `run_in_tx` callback | Use `run_in_tx`; for partial rollback use `run_in_savepoint` |
| `pgcommon: invalid GUC identity: …` at acquire | The identity was stored with `with_guc_set` (unvalidated) and fails `GUCSet.validate()` | Validate at the edge with `with_validated_guc_set` or `GUCAuthMiddleware`, and answer 401 |

---

## 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"
```

Integration and e2e tests use `testcontainers-python` — Docker must be running locally. Unit tests have no external dependencies and run without Docker. `make rls-check` runs only the two RLS suites (direct and PgBouncer mode).

**Status (2026-10-15):** 2375 tests pass (1 skipped) across the three suites, with **100% line and branch coverage** of `src/pgcommon`. Every invariant of the [LLD](docs/lld/platform-lld-pgcommon-py.md) and every 0.3.0 CHANGELOG fix is pinned by an asserting test, in dedicated invariant suites: `tests/unit/test_core_invariants.py`, `test_core_redaction_and_config.py` (password redaction across every DSN form and the whole exception chain; config and environment matrices), `test_txshape_go_oracle.py` (the transaction-shape lexer against answers produced by Go's own `AnalyzeTx`), `test_migrate_invariants.py`, `test_migrate_cli_invariants.py`, `test_metrics_invariants.py`, `test_fastapi_invariants.py`, `test_public_api.py`, and on real servers `tests/integration/test_core_invariants_integration.py`, `test_migrate_invariants_integration.py` and `tests/e2e/test_core_invariants_e2e.py`. Mutation testing (run by hand) drove them: surviving mutants fell from 537 to 204 in the core and from 299 to 106 in the migration runner, metrics and CLI.

Single tests and the `--no-cov` flag a subset needs: [Local development → Running a single test](#running-a-single-test). The suites never use the `docker-compose.yml` stack; they start their own containers.

### Canonical tests (do not break)

- **`tests/integration/test_rls_guc_injection.py`**: standard-mode isolation end to end, through a dedicated non-superuser role on a `FORCE ROW LEVEL SECURITY` table — a superuser would make RLS silently do nothing.
- **`tests/e2e/test_rls_pgbouncer_e2e.py`**: the same guarantee behind a real PgBouncer in transaction mode, including 90 concurrent requests for three tenants over a 10-connection backend pool with no identity leak.
- **`tests/e2e/test_core_invariants_e2e.py`**: every PgBouncer backend checked — after identity-bearing work an identity-less pool holds all ten server connections at once, and none may show an identity or a row. Sampling backends, as the test above does, would miss a session-scoped GUC.
- **`tests/integration/test_core_invariants_integration.py`**: through a non-superuser `FORCE ROW LEVEL SECURITY` role, an identity never survives the release of its connection (also when the callback set a session GUC itself), concurrent callers see only their own tenant, and RLS refuses a cross-tenant write.
- **`tests/integration/test_tx_escape.py`, `tests/e2e/test_pgbouncer_callback_escape_e2e.py`**: `TxEndedInCallbackError` / `TxCommitRollbackError` detection, and `test_released_connection_never_carries_the_previous_identity`.
- **`tests/integration/test_pool_lifecycle.py::test_pool_helpers_are_the_only_supported_entry_point`**: `Pool` exposes no raw asyncpg pool or connection.
- **`tests/integration/test_migrate_runner_integration.py`** and **`tests/unit/test_migrate_runner.py::test_advisory_lock_id_matches_go_runner`**: concurrent runners serialise on golang-migrate's advisory lock, with the Go runner's lock IDs (the database name follows Go's `databasePath`, `test_database_name_follows_gos_url_parse`).
- **`tests/integration/test_failover.py`**: repeated failovers keep full pool capacity; the pool tops back up to `min_conns`.
- **`tests/unit/test_metrics_standard.py`, `tests/unit/test_metrics_inventory.py`** (`make metrics-lint`): every registered metric conforms to the registry, Tier 1 matches the vectors generated from Go's `pgmetrics` (`tests/fixtures/go_pgmetrics_vectors.json`), and `docs/observability/metrics-registry.md` is current.
- **`tests/unit/test_config_parity.py`, `test_pool_parity.py`, `test_internal_parity.py`**: Go-parity edge cases of config resolution and `config_from_env`, the pool (lifetimes, `Stat`, GUC-injection failures, shutdown, DSN redaction) and the internal helpers (DSN and duration parsing, backoff, SQLSTATE).

### Coverage

`pyproject.toml` sets `fail_under = 97` (branch coverage of `src/pgcommon`); the full run currently reaches 100%. `make test-ci` runs every suite once with coverage and `.github/scripts/coverage-gate.sh` checks the threshold separately, so a failing test and a coverage drop are told apart; `make cover-check` runs both, and `make cover-func` lists missing lines per file from the last run. The unit suite alone clears the gate, so `uv run pytest -m "not integration and not e2e"` passes on a machine without Docker. No reachable code is excluded: the few `# pragma: no branch` / `# pragma: no cover` lines are unreachable guards, each commented in the code and listed in the [LLD § 14.7](docs/lld/platform-lld-pgcommon-py.md#147-coverage-exclusions).

---

## Environment variables

Read only when the service calls `config_from_env()` (the pool), `migration_dsn_from_env()`, a `pgcommon.metrics` identity helper, or runs the migrate CLI. Nothing is read at import time. pgcommon reads **no** `OTEL_*` or log-level variable: the logger, tracer and Prometheus registry are passed in by the service. `.env-example` documents the local defaults (`make setup` copies it to `.env`).

### Pool (`config_from_env`)

`config_from_env()` (Go: `ConfigFromEnv`) builds a `Config` from the environment. Nothing is read at import time. Invalid values never raise: each is replaced by its default with a `ConfigWarning`.

| Env var | Default | Notes |
|---|---|---|
| `DATABASE_URL` | — | Full DSN (URL or keyword/value); takes precedence over every `PG_*` connection variable below. Its `pool_max_conns` / `pool_min_conns` apply unless `PG_MAX_CONNS` / `PG_MIN_CONNS` is set |
| `PG_HOST` / `PG_PORT` | `localhost` / `5432` | Used when `DATABASE_URL` is unset. `PG_HOST=db:5433` is split into host and port; a set `PG_PORT` wins, with a warning when they differ |
| `PG_USER` / `PG_PASSWORD` / `PG_DBNAME` | — | Used when `DATABASE_URL` is unset; user and dbname are required (otherwise a warning and an empty DSN). Percent-encoded into the DSN |
| `PG_SSLMODE` | `require` | Used when `DATABASE_URL` is unset; `disable` / `allow` / `prefer` warn |
| `PG_MAX_CONNS` / `PG_MIN_CONNS` | `10` / `2` capped at `PG_MAX_CONNS` (`0` in PgBouncer mode) | `PG_MAX_CONNS` must be positive, `PG_MIN_CONNS` zero or more |
| `PG_BOUNCER_MODE` | `false` | Parsed like Go's `strconv.ParseBool`, after trimming: `1`, `t`, `T`, `TRUE`, `true`, `True` / `0`, `f`, `F`, `FALSE`, `false`, `False`; anything else warns and means `false` |
| `PG_MAX_CONN_LIFETIME` / `PG_MAX_CONN_IDLE_TIME` / `PG_HEALTH_CHECK_PERIOD` | `1h` / `30m` / `1m` | Go durations (`500ms`, `1h30m`); negative or unparseable warns |
| `PG_MAX_CONN_LIFETIME_JITTER` | lifetime ÷ 10 | `off` (any case) or a negative value disables it. "Negative value" means a negative Go duration or a negative number as Go's `strconv.ParseFloat` reads it (`-1`, `-1_0`, `-0x1p-2`, `-Inf`), after Go's whitespace trimming; anything else Go rejects (a non-ASCII digit, `-1e400`) warns and keeps the default — this applies to every `off`-able variable |
| `PG_SLOW_QUERY_THRESHOLD` | `200ms` | `off` or a negative value disables it; `0` means the default |
| `PG_STATEMENT_TIMEOUT` / `PG_LOCK_TIMEOUT` | unset | Transaction-local timeouts (see [Statement and lock timeouts](#statement-and-lock-timeouts)) |
| `PG_POOL_NAME` | `default` | The `pool` metric label, taken verbatim and validated by `Pool.create` |
| `PG_ACQUIRE_TIMEOUT` | `30s` | `off` or a negative value: no limit; `0` means the default |
| `PG_COMMAND_TIMEOUT` | unset | Client-side limit per statement (e.g. `60s`); negative or unparseable warns |
| `PG_SATURATED_ALIVE_WINDOW` | `10s` | `off` or a negative value: `health()` always pings; `0` means the default |
| `PG_PREPARED_XACT_GUARD` | `true` | Parsed like `PG_BOUNCER_MODE`; see `prepared_xact_guard` |
| `MIGRATION_DATABASE_URL` | the app DSN | Read by `migration_dsn_from_env()` and the migrate CLI — see [Migrations](#migrations) |

Like Go's, the returned `Config` carries `max_conns` and `min_conns` explicitly (`10` and `2` by default) — except that an env *default* never overrides a DSN `pool_max_conns` / `pool_min_conns`, which `Pool.create` then applies.

> **TLS warnings.** Log every warning `config_from_env()` returns at startup, so accidental TLS disablement or a typo'd value is visible:
>
> ```python
> config, warnings = config_from_env()
> for w in warnings:
>     logger.warning("config warning", key=w.key, reason=w.reason)
> ```
>
> - An explicit insecure `sslmode` (`disable`, `allow`, `prefer`) warns — keyed `PG_SSLMODE` for the `PG_*` DSN, `DATABASE_URL` for `DATABASE_URL`.
> - A `DATABASE_URL` that sets **no** `sslmode` warns when the connection asyncpg would make can fall back to plaintext: asyncpg then uses `PGSSLMODE`, else `prefer`, for any TCP host (Unix-socket hosts carry no TLS). The DSN's `service=` file is honoured (its `sslmode` counts), and `sslnegotiation=direct` makes `prefer` TLS-only, so it does not warn. The warning is keyed `PGSSLMODE` when an insecure `PGSSLMODE` applies, else `DATABASE_URL`.
> - `sslrootcert=system` **still warns**, unlike Go: asyncpg does not implement libpq's `sslrootcert=system` (which forces `verify-full` in pgx). Set `sslmode=verify-full` explicitly.
> - A `DATABASE_URL` Go's URL parser would reject (an invalid `%` escape, a bad port or IPv6 host, a control character — typically a password with unescaped reserved characters) warns without quoting it.
>
> Prefer `sslmode=verify-full` in production.

### Metrics identity and migrations

| Env var | Default | Notes |
|---|---|---|
| `APP_ENV` (then `ENVIRONMENT`) | `dev` | The `environment` label when `Identity.environment` is empty: trimmed and lower-cased, never re-spelled (`environment_from_env()`) |
| `APP_NAME` | — | The `service` label, only through `identity_from_env(domain)` with no `service`; otherwise the service passes `Identity.service` |
| `MIGRATION_DATABASE_URL` | the app DSN | Migrations: a DDL role, **direct** to Postgres. Read by `migration_dsn_from_env()` and the CLI |
| `MIGRATIONS_DIR` · `MIGRATIONS_TABLE` · `MIGRATIONS_LOCK_TIMEOUT` | — (**required** by the CLI) · `pgcommon_migrations` · `30s` | CLI only — see [The `pgcommon-migrate` CLI](#the-pgcommon-migrate-cli) |
| `PGCONNECT_TIMEOUT` | — | The connect timeout when the DSN sets no `connect_timeout` (pool and runner, as pgx) |
| `PGSSLMODE` | — | The `sslmode` asyncpg uses when `DATABASE_URL` sets none; an insecure value is warned about |

---

## Security

| Topic | Guidance |
|---|---|
| **Tenant isolation** | RLS is the floor: `FORCE ROW LEVEL SECURITY` plus a `NULLIF(current_setting(…, true), '')` policy, and an application role that is neither superuser, `BYPASSRLS` nor the owner. pgcommon guarantees the identity on the connection is the current request's, or none — it cannot verify that your policies exist or are correct |
| **No identity leak across pooled connections** | Direct mode: asyncpg's `RESET ALL` on release clears the session GUCs, and a connection whose reset fails or times out is terminated, never reused. PgBouncer mode: identity is only ever transaction-local. Both are integration-tested on real Postgres and PgBouncer |
| **No identity leak across tasks** | The identity lives in a `ContextVar`, copied per `asyncio` task. `with_guc_set` / `with_validated_guc_set` are context managers that always reset their token; a worker started inside a request inherits that request's identity — start long-lived workers from the lifespan ([FastAPI integration](#fastapi-integration)) |
| **Cross-tenant access** | A separate `Pool` on a `BYPASSRLS` role with **no** `guc_provider`, used only by reconcilers and exporters. Never widen the app pool's role |
| **Credentials** | An error that would expose the DSN password anywhere in its exception chain (message, notes, `__cause__`, `__context__`, exception-group members) is replaced in the chain by a redacted stand-in — in every spelling: raw, quoted, decoded, re-escaped, and a percent-encoded `?password=` value exactly as written; PostgreSQL server errors are kept, so `sql_state` and the `is_*` helpers still work. `PG_SSLMODE` defaults to `require`; weaker modes, and a `DATABASE_URL` / `MIGRATION_DATABASE_URL` that could fall back to plaintext, produce a warning. Prefer `sslmode=verify-full` |
| **Telemetry privacy** | SQL text is never a metric label. Spans and slow-query logs carry the SQL verb only, and span errors only the SQLSTATE (PostgreSQL messages quote input values), unless `allow_full_statements=True`. `tenant_id` is logged only with `log_tenant_id=True`; bind values are never logged or traced |
| **Migration image** | `python:3.14-slim`, both bases digest-pinned, numeric non-root UID 65532, no `pip` in the final image, dependencies installed from `uv.lock` with `--require-hashes`. Trivy-gated (CRITICAL / HIGH / UNKNOWN with a fix), with a CycloneDX SBOM, SLSA provenance and a Cosign keyless signature |
| **Supply chain** | PyPI uploads only through Trusted Publishing (OIDC, no stored token) from `release.yml`, with PEP 740 attestations; the release's `SHA256SUMS` is Cosign-signed ([Deployment](#deployment)). The build backend is hash-locked (`build-constraints.txt`), `pip-audit` runs in CI, every third-party action is pinned by commit SHA, and Dependabot proposes pip, GitHub Actions and Docker updates weekly |

[SECURITY.md](.github/SECURITY.md) has the trust model, the scope of vulnerability reports and how to report one privately.

---

## 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_with_identity` or supplying `Config.tracer`.

### Prometheus metrics

pgcommon implements the **Enterprise Platform Observability Standard** exactly as the Go `platform-pgcommon` library does. Its metrics are **Tier 1 `platform_db_*`**: every service in every domain that reaches PostgreSQL through pgcommon emits them with the same names, labels, label values, buckets and help text, whichever language it is written in. **These are the only metrics pgcommon emits**: the pre-standard legacy metrics were removed with no compatibility period, because no library emitting them had been deployed. The full story (classification, wiring, divergences from Go) is in [docs/observability/](docs/observability/README.md), and every metric's ratification packet is in [docs/observability/metrics-registry.md](docs/observability/metrics-registry.md).

Initialise once at startup, **before** `Pool.create`:

```python
import pgcommon.metrics as pgmetrics

warnings = pgmetrics.init_with_identity(
    pgmetrics.Identity(
        domain="iam",  # the service's business domain
        service="event-consumer",  # the service name
        environment="",  # "" → APP_ENV, then ENVIRONMENT, else "dev"
    ),
    # registry=...  (default: prometheus_client.REGISTRY)
)
for w in warnings:
    logger.warning(str(w))  # a platform_db_* metric was disabled (name clash)
```

The identity is **mandatory**: `init_with_identity` is the only way to register pgcommon's metrics. `pgmetrics.identity_from_env("iam")` builds the identity from the environment (`service` from `APP_NAME`). `pgmetrics.identity_from_labels(labels)` builds it from a `{domain, service, environment}` const-label map (other keys are ignored), so pgcommon's metrics carry exactly the values the service's own metrics do.

- **Required labels are bound centrally.** `domain`, `service` and `environment` are applied to every `platform_db_*` series at registration, so no recording site can omit or misspell them. Each must match `^[a-z][a-z0-9_-]{0,62}$`; an invalid identity raises `InvalidIdentityError` and changes nothing. If the bad value came from `APP_ENV`/`ENVIRONMENT`, the message names the variable.
- **`pool` is `Config.pool_name`** (env `PG_POOL_NAME`, default `"default"`). `Pool.create` rejects anything but `^[a-z][a-z0-9_]{0,31}$` with `InvalidPoolNameError`, so the label stays bounded.
- **Pool gauges are automatic.** After `init_with_identity`, every `Pool.create` registers the pool's `platform_db_pool_*` gauges on the same registry, and `close` / `drain_and_close` remove them. Live pools with the same `pool_name` (a credential rotation, say) share one gauge set reporting their sum.
- **Counters start at 0.** `Pool.create` pre-creates each pool's acquire, slow-query and retry counters at 0, and every `init_with_identity` pre-creates them again for every live pool — including pools created before the first init. Otherwise a counter first appears at 1, and `increase()`/`rate()` miss that first event, so the first retry exhaustion after a deploy would never alert.
- **Re-init.** Re-initialising with the same identity keeps every series and its value; with another identity, the previous identity's series are removed from the collectors pgcommon registered (never from a collector the service registered itself). Concurrent inits are serialised. A pool created **before** `init_with_identity` records its query, acquire and retry metrics but exports no gauges: initialise first.
- **Fail-soft registration.** A `platform_db_*` metric whose name is already registered with another shape is disabled and returned as a `RegistrationWarning`; the service keeps running and every other `platform_db_*` metric keeps recording.
- **What is counted.** Query metrics count the caller's statements only, whether or not `Config.tracer` is set. Library-internal statements (the GUC `set_config`, `BEGIN` / `COMMIT` / `ROLLBACK`, savepoints, the PgBouncer wrapping transaction) are counted only **if they fail**: a failed `COMMIT` is the caller's work failing. They never reach the slow-query log or counter. The pool `ping` and asyncpg's reset query on release are never counted. An acquire that fails because the caller was cancelled, or because the request's `GUCSet` is invalid, is not an acquire error, though its wait is still observed.

Tier 1 metrics (Proposed: not yet a valid alerting target until governance ratifies them):

| Metric | Type | Labels (plus `domain`, `service`, `environment`) |
|---|---|---|
| `platform_db_queries_total` | Counter | `pool`, `operation`, `outcome` |
| `platform_db_query_duration_seconds` | Histogram | `pool`, `operation` |
| `platform_db_pool_acquires_total` | Counter | `pool`, `outcome` |
| `platform_db_pool_acquire_duration_seconds` | Histogram | `pool` |
| `platform_db_slow_queries_total` | Counter | `pool` |
| `platform_db_transaction_retries_total` | Counter | `pool`, `reason` |
| `platform_db_transaction_retries_exhausted_total` | Counter | `pool`, `reason` |
| `platform_db_pool_connections` | Gauge | `pool`, `state` (`idle`, `acquired`) |
| `platform_db_pool_max_connections` | Gauge | `pool` |
| `platform_db_pool_empty_acquires_total` | Counter | `pool` |

`operation` is the SQL verb from a closed allowlist (`SELECT`, `INSERT`, `UPDATE`, `DELETE`, `WITH`, …, else `other`), never the raw SQL. Leading `/* … */` and `-- …` comments are skipped, exactly as Go does. `outcome` is `success`/`error`; `reason` is `deadlock`/`serialization`. `state="constructing"` is not emitted: asyncpg can't report connections that are still being established.

For isolated test registries, pass a fresh `prometheus_client.CollectorRegistry()` as `registry=`. Use `pgmetrics.replace_current(None)` to reset the process-wide set between tests.

### OpenTelemetry query spans

```python
from opentelemetry import trace

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

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

Each statement produces a `db.query` span (kind `CLIENT`) with the stable OpenTelemetry database semantic-convention attributes, the same ones the Go library records:

| Attribute | Value |
|---|---|
| `db.system.name` | `postgresql` |
| `db.operation.name` | The SQL verb from `sql_operation` (`SELECT`, `INSERT`, …, `other`) — never literals |
| `db.namespace` | The database name |
| `server.address` / `server.port` | The server the connection is attached to |
| `db.query.text` | The full SQL — **only** with `Config.allow_full_statements = True` |

A failed statement records only its SQLSTATE — `db.response.status_code`, `error.type` (the code), `db.response.severity` and the status `SQLSTATE <code>` — because PostgreSQL error messages quote input values. A non-PostgreSQL error gets `error.type` `context_canceled` (cancellation), `context_deadline_exceeded` (timeout) or `_OTHER`, and the status `database error`. The error message and an exception event are added only with `allow_full_statements`.

> **SQL privacy:** by default spans carry only the SQL verb and the SQLSTATE, to prevent PII leakage to tracing backends. Set `Config.allow_full_statements = True` to add the full SQL and error messages — 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.

### Extra query logger and acquire / release hooks

`Config.query_logger` (Go: `Config.QueryTracer`) is an extra asyncpg query logger, registered on every connection after the built-in metrics, slow-query and span recorders. It receives every caller statement (and library-internal ones), but not the pool `ping` or asyncpg's reset query. If it also implements the `AcquireTracer` / `ReleaseTracer` protocols (`pgcommon.AcquireTracer`, `pgcommon.ReleaseTracer`), it receives the pool's acquire and release events:

```python
class AuditLogger:
    def __call__(self, record) -> None: ...  # every statement: query, args, elapsed, exception
    def trace_acquire_start(self, pool) -> None: ...  # before with_conn / run_in_tx wait
    def trace_acquire_end(self, pool, conn, error) -> None: ...  # conn or error
    def trace_release(self, pool, conn) -> None: ...  # before the connection goes back


pool = await Pool.create(Config(dsn=dsn, query_logger=AuditLogger()))
```

An exception any of these hooks — or a built-in sink — raises is logged at Error level through `Config.logger` and otherwise ignored: it never fails the query or the acquire, and never leaks the connection.

### Alerts, rules and dashboard

This port ships no recording rules, alerts, dashboard or runbook of its own. The Tier 1 metric names, labels, help text and buckets are identical in Go and Python, so the Go repository's reference assets apply to Python services **unchanged**, and aggregate both languages in one query: `monitoring/prometheus/pgcommon.rules.yml` (recording rules, SLIs and the six `Pgcommon*` alerts), `monitoring/grafana/pgcommon-overview.json` and `docs/observability/runbook.md`. Group every `platform_db_*` query `by (domain, service, environment, pool)`, never by `service` alone. [docs/observability/README.md § Alerts, rules and dashboard](docs/observability/README.md#alerts-rules-and-dashboard) lists the few panels that differ for Python pools (no `state="constructing"` series).

---

## Deployment

pgcommon itself is released to PyPI, and services adopt it by bumping their pin; it is never deployed on its own. The migrate CLI image is the only deployable.

### Installing a release

Pin an exact version in production ([VERSIONING.md § Consume a release](VERSIONING.md#consume-a-release) has the pin styles):

```bash
pip install "pgcommon==0.3.0"            # or "pgcommon[fastapi]==0.3.0" / "pgcommon[structlog]==0.3.0"
uv add "pgcommon==0.3.0"
```

The files on PyPI are byte-identical to the GitHub Release's wheel and sdist: the release builds them once and uploads the same files to both.

### Container image — one entrypoint

| Executable | In the image | Purpose |
|---|---|---|
| `pgcommon-migrate` | `ENTRYPOINT ["pgcommon-migrate"]`, no default `CMD` | Runs `up` / `down` / `status` / `force` / `version` against `MIGRATION_DATABASE_URL` with the SQL in `MIGRATIONS_DIR` (mounted: the image ships no migrations) |

Two-stage `Dockerfile`. The build stage (`python:3.14-slim` with uv `0.11.32`, both digest-pinned) exports the runtime dependencies from `uv.lock` with hashes and builds the wheel with the hash-locked backend from `build-constraints.txt`, under a reproducible `SOURCE_DATE_EPOCH`. The runtime stage (`python:3.14-slim`, digest-pinned) installs them with `--require-hashes`, then the wheel with `--no-deps`, removes `pip`, and runs as numeric `USER 65532:65532` so `runAsNonRoot` is enforceable. The image is about 175 MB uncompressed (the smoke gate fails above 250 MB). Image tags have no `v`: `X.Y.Z` always, `X.Y` and `X` only when the release is the newest stable version of that line, never `latest` (a CLI utility image, not a service). A pre-release gets only its own tag.

### Verifying a release

Needs [cosign](https://docs.sigstore.dev/cosign/system_config/installation/) ≥ 2.4. A tag-triggered release is signed by `release.yml@refs/tags/<tag>`; a re-run dispatched from `main` by `release.yml@refs/heads/main`, which the regular expression below also accepts.

```bash
TAG=v0.3.0   # the release to verify
REPO=BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
IDENTITY="^https://github\.com/${REPO}/\.github/workflows/release\.yml@refs/(tags/${TAG}|heads/main)$"

# Wheel + sdist: verify the signed checksums, then the files against them
gh release download "$TAG" -R "$REPO" -p SHA256SUMS -p SHA256SUMS.sigstore.json -p '*.whl' -p '*.tar.gz'
cosign verify-blob --bundle SHA256SUMS.sigstore.json \
  --certificate-identity-regexp "$IDENTITY" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  SHA256SUMS
sha256sum --check SHA256SUMS   # macOS: shasum -a 256 -c SHA256SUMS

# Migrate CLI image (tag without the v)
cosign verify --certificate-identity-regexp "$IDENTITY" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  "ghcr.io/bcbp-solutions-fzc-llc/platform-pgcommon-py:${TAG#v}"
```

Each GitHub Release carries the wheel, the sdist, `SHA256SUMS`, `SHA256SUMS.sigstore.json`, the image's SLSA provenance (`provenance.slsa.json`) and its CycloneDX SBOM (`sbom.cyclonedx.json`); PyPI additionally shows the Trusted Publishing attestations. Releases up to `v0.2.0` predate signing and `SHA256SUMS`. The workflow itself verifies with the looser `refs/(tags/v.+|heads/main)`; pinning the exact tag, as above, is stricter. [VERSIONING.md § Verifying a release](VERSIONING.md#verifying-a-release) is the canonical copy of these commands.

### Migration safety

Run migrations as a separate init container or pre-deploy job, to completion, over a direct connection (`MIGRATION_DATABASE_URL`), before the new version rolls out. Keep every migration compatible with the previous release (expand/contract). See [Migrations](#migrations).

### Versioning and releases

**Release status:** 0.3.0 was released on 2026-10-15 (`CHANGELOG.md` `[0.3.0]`): Go v2.0.1 parity and production hardening, breaking against 0.2.0 — read the CHANGELOG's Upgrade notes before upgrading. `platform-eventcommon` supports it from its `pgcommon>=0.2.0,<0.4.0` range, and the cross-language interop harness checks the pair at `platform-interop-tests` `9bd44ca`.

**Go parity:** this port tracks Go `platform-pgcommon` **v2.0.1** — the same Tier 1 metrics, configuration, transaction, error and migration behaviour, and the same migrate CLI, except for the documented [divergences](ARCHITECTURE.md#divergences-from-go). The Python package keeps its own version line (`0.x`); the CHANGELOG names the Go version each change corresponds to.

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 |

While the package is `0.y.z`, a MINOR bump may break the public API (SemVer §4) and a PATCH may not. The release's `api-compat` gate (griffe) enforces exactly that.

```bash
# Tag a commit on main; pushing the tag triggers the release workflow.
make release-guard TAG=vX.Y.Z          # pyproject version == tag, CHANGELOG has [X.Y.Z]
git tag -a vX.Y.Z -m "vX.Y.Z"
git push origin vX.Y.Z
```

`release.yml` mirrors the Go repo's release. `verify` checks the dispatch ref, that the tag is at HEAD and on `origin/main`, that `pyproject.toml`'s version equals the tag (`release_guard.py`), the CHANGELOG section, and which floating image tags may move. Then the CI gates run again at the tag: tests, quality, `api-compat` (blocking), Hadolint, Trivy, smoke and interop. `dist` builds the wheel and sdist **once**, reproducibly. Only after every gate, `push` publishes the fingerprint-checked image (`X.Y.Z`, plus `X.Y` / `X` only when it is the newest of that line; never `latest`) with SBOM, SLSA provenance and a Cosign signature. Next, `pypi-publish` uploads those same files via Trusted Publishing. Last, `publish` creates the GitHub Release: the CHANGELOG section as notes, with the wheel, the sdist, a Cosign-signed `SHA256SUMS`, the provenance and the SBOM attached. Interop gates `push`, so a red cross-language check stops the whole release. [VERSIONING.md § Verifying a release](VERSIONING.md#verifying-a-release) has the `cosign` commands.

---

## CI

The pipeline has the same structure, jobs, gates and scripts as the Go `platform-pgcommon`'s, translated to Python tooling: six workflow files, job logic in shellcheck-clean `.github/scripts/*`, and every third-party action pinned by commit SHA (`# vX.Y.Z` comment).

- **`ci.yml`** (the orchestrator; PRs and pushes to `main`). `changes` (`detect-changes.sh`) decides whether the change is docs-only (any `*.md`, `docs/architecture/*.mmd`; never `docs/observability/`). The workflow always runs, so the required checks always report; on a docs-only change the jobs behind them are skipped (a skipped job counts as success). `validate-test` and `validate-quality` run in parallel with `build-image` (Hadolint → cached build). The image then goes to `trivy` and `smoke`, each recording the image's content fingerprint. `interop` runs platform-interop-tests' reusable workflow, pinned to a commit SHA, and `pr-summary` comments on the PR. On a push to `main`, `push` publishes to GHCR (`main` / `sha-<short>`), but only an image whose fingerprint matches the one Trivy and smoke judged; it adds SBOM and provenance, then Cosign keyless-signs and verifies.
- **`validate-test.yml`** (reusable): `make test-ci` (unit + integration + e2e, coverage), the coverage gate (`coverage-gate.sh`, **≥ 97%** from `pyproject.toml`'s `fail_under`) and `make arch-lint` (import-linter) on Python 3.12, plus the same suite on 3.13 / 3.14 and on asyncpg 0.30.0.
- **`validate-quality.yml`** (reusable): `uv lock --check`, `uv pip check`, the HTML-entity check, the RLS session-`SET app.*` guard, `ruff format --check`, `ruff check`, `mypy --strict`, `make metrics-lint`, `make docs-check`, `make ci-scripts-test`, `pip-audit`, the Dockerfile digest-pin check, and `make build` (wheel + sdist, `twine check`, console script).
- **`release.yml`**: runs on `v*` tags. It is the same pipeline at the tag, with a release front and a release back; see [Versioning and releases](#versioning-and-releases).
- **`docs.yml`**: diagram sync on docs-only changes (not a required check).
- **`changelog-check.yml`**: fails a PR that touches `src/`, `pyproject.toml`, `uv.lock`, `build-constraints.txt` or the `Dockerfile` without updating `CHANGELOG.md`. Dependabot PRs are exempt.

| Job (`ci.yml`) | Check name | What it gates | Required |
|---|---|---|---|
| `changes` | `Detect changes` | Docs-only decision + the shared `SOURCE_DATE_EPOCH` | — |
| `validate-test` → `test` | **`Validate / Test / test`** | Full suite on Python 3.12 + coverage ≥ 97% + import-linter; gates `trivy`, `smoke`, `interop`, `push` | ✅ |
| `validate-test` → `python` | `Validate / Test / python 3.13`, `… / python 3.14` | Full suite (coverage gate included) on the other supported Pythons; gates like `test` | — |
| `validate-test` → `asyncpg-compat` | `Validate / Test / asyncpg 0.30.0` | Full suite on the oldest admitted asyncpg minor (`uv sync` + `uv pip install asyncpg==0.30.0`, `--no-cov`): pgcommon relies on non-public asyncpg behaviour; gates like `test` | — |
| `validate-quality` → `quality` | **`Validate / Quality / quality`** | Lock, format, lint, types, RLS guard, metrics standard, docs, CI-script tests, pip-audit, digest pins, packaging; gates `push` | ✅ |
| `api-compat` | `API compatibility` | `griffe check pgcommon` vs the last release tag; **warning only** here, blocking in `release.yml` | — |
| `build-image` | `Build image (cache)` | Hadolint + `.dockerignore`, then a cached build (GHA + registry cache) | — |
| `trivy` | `Trivy CVE scan` | CRITICAL / HIGH / UNKNOWN with a fix → fail; SARIF to the Security tab; CycloneDX SBOM artifact | — |
| `smoke` | `Smoke tests` | `smoke-tests.sh`: size ≤ 250 MB; a bare run prints `usage: pgcommon-migrate` and exits 1; `version` / `--version`; `up` exits 1 without a DSN, without `MIGRATIONS_DIR` and against an empty one (`no migration files found`); non-root numeric UID | — |
| `interop` | `Cross-language compatibility / Cross-language compatibility` | platform-interop-tests (SHA-pinned). Not required in CI; **gates the release's `push`** | — |
| `pr-summary` | `PR summary` | PRs only: one status comment (fails only if `changes` failed) | — |
| `push` | `Push image → GHCR` | Pushes only: fingerprint-checked push, SBOM + provenance, Cosign sign + verify | — |

**Private repository:** checkouts use `CI_REPO_READ_TOKEN || github.token` without persisting credentials. Reusable workflows get explicit `secrets:` maps, never `secrets: inherit`, and no credential is ever written for the test code. `GO_PRIVATE_TOKEN` (org secret, read access to the six platform libraries) goes only to the interop job. `CODEOWNERS` routes reviews to `@BCBP-SOLUTIONS-FZC-LLC/python-platfrom` (sic). It takes effect only once that team has **write** access.

Every image build (`build-image`, `trivy`, `smoke`, `push`, and the release's) uses the same build args. `SOURCE_DATE_EPOCH` is the commit time, passed as the Dockerfile build arg and as BuildKit's environment, with `rewrite-timestamp=true` on the loaded and pushed outputs. The fingerprint check then *proves* that the image pushed is the one Trivy scanned and smoke tested. Locally: `SOURCE_DATE_EPOCH=$(git log -1 --format=%ct) docker buildx build --build-arg SOURCE_DATE_EPOCH --output type=docker,rewrite-timestamp=true .` (the `rewrite-timestamp` option needs a BuildKit ≥ 0.13 `docker-container` builder or the containerd image store).

Run the same gates locally before pushing:

```bash
make ci            # lock-check + format-check + lint + typecheck + arch-lint + metrics-lint + docs-check
                   # + ci-scripts-test + test-ci + coverage gate + build
make api-compat    # griffe vs the last release tag (warning; API_NEW_VERSION=vX.Y.Z makes it the release gate)
make lint-docker   # hadolint (CI: first step of build-image)
make scan          # Trivy, same severities as CI
make smoke-test    # build the image + .github/scripts/smoke-tests.sh
```

`make docs-check` checks that every `docs/architecture/mermaid/*.mmd` is embedded byte-identically in `ARCHITECTURE.md`. Edit a `.mmd` file and its embedded block together. [CONTRIBUTING.md § CI pipeline](CONTRIBUTING.md#ci-pipeline-go--python-mapping) maps every Go job and script to its Python counterpart.

---

## Docker

### What the bundled `docker-compose.yml` starts

| Container | Image | Host port | Purpose |
|---|---|---|---|
| `postgres` | `postgres:16-alpine` | `5432` | Local database `test` (user and password `test`, a superuser) |
| `pgbouncer` | `edoburu/pgbouncer:latest` (compose profile `pgbouncer`) | `6432` | Transaction-pooling proxy in front of `postgres` (`AUTH_TYPE=plain`, pool size 20), for trying `pgbouncer_mode` locally |

`make docker-up` starts only `postgres` and waits for `pg_isready`; `make docker-up-all` starts both; `make docker-down` removes them **with their volumes**. The integration and e2e suites don't use these: they start their own containers with testcontainers (the same `postgres:16-alpine` and `edoburu/pgbouncer:latest` images).

### Building the migrate image

```bash
make docker-build   # → pgcommon-migrate:local (BUILD_VERSION from git describe, SOURCE_DATE_EPOCH = commit time)
make smoke-test     # build + CI's image gates (.github/scripts/smoke-tests.sh)
make lint-docker    # hadolint, as the first step of CI's build-image job
make scan           # Trivy, CI's severities
```

### Running a migration with the image

```bash
docker run --rm -v "$PWD/migrations:/migrations:ro" -e MIGRATIONS_DIR=/migrations \
  -e MIGRATION_DATABASE_URL='postgres://migrator:…@db:5432/app?sslmode=verify-full' \
  ghcr.io/bcbp-solutions-fzc-llc/platform-pgcommon-py:<version> up
```

The image contains only the CLI (`ENTRYPOINT ["pgcommon-migrate"]`, no default `CMD`, so a bare `docker run` prints the usage and exits 1). It ships no migrations — mount the service's directory, or build a service image `FROM` it with the migrations copied in. The exit code is 0 on success (including "no change"), 1 on any usage, configuration, connection or migration error, 3 when the schema is dirty and 4 when the database's version is not in `MIGRATIONS_DIR`; `status` uses the same codes. Locally, against the compose database, use `--network host` (Linux) or `host.docker.internal` (Docker Desktop) in the DSN.

### Minimum required environment variables

```bash
# A consuming service's pool (config_from_env)
DATABASE_URL=postgres://app_user:…@pgbouncer:6432/app?sslmode=verify-full
PG_BOUNCER_MODE=true            # when DATABASE_URL points at PgBouncer / RDS Proxy
APP_ENV=production              # the environment label — use the platform's agreed spelling

# Its migrations (init container or pre-deploy job)
MIGRATION_DATABASE_URL=postgres://migrator:…@postgres:5432/app?sslmode=verify-full   # direct, never PgBouncer
MIGRATIONS_DIR=/migrations
```

---

## Dependencies

### 1. Consumers (projects that import this library)

| Consumer | pgcommon pin | Uses |
|---|---|---|
| `platform-eventcommon` (Python port of `platform-events`) | `>=0.2.0,<0.3.0` — moving to 0.3.0 needs a coordinated change ([Upgrading from 0.2.0](#upgrading-from-020)) | `Pool` and `run_in_tx` for its outbox, inbox and DLQ stores; `with_guc_set` so message handlers run under RLS; `MigrationDirtyError` for its own schema bookkeeping |
| `platform-interop-tests` (Python probe) | — | Cross-language compatibility with the Go library (CI's `interop` job) |
| FastAPI platform services | their own | Pool, RLS injection, transactions, error helpers, metrics, `pgcommon.integrations.fastapi` |

### 2. Upstream libraries

[Runtime dependencies](#runtime-dependencies-what-consumers-inherit) lists the ranges. asyncpg is the load-bearing one, and it is capped at `<0.32` because pgcommon relies on asyncpg behaviour that is not public API: the query logger's `call_soon` dispatch (which carries the caller's `contextvars` context), the pool proxy's detachment on connection loss, `Connection._top_xact` (cleared after pgcommon's own `COMMIT`), and the slot asyncpg 0.31 leaks when a protocol error closes a connection. CI's `asyncpg-compat` job runs the whole suite against the oldest admitted minor, `asyncpg==0.30.0`; the required `test` job runs the locked 0.31.0. The floor is 0.30: with 0.29, `command_timeout` does not free the slots of callers stuck on a frozen server. [docs/lld/platform-lld-pgcommon-py.md](docs/lld/platform-lld-pgcommon-py.md) § 3.1 lists every such behaviour and what an upgrade that changed it would break.

### 3. Platform repos

| Repo | Relationship |
|---|---|
| `platform-pgcommon` (Go) | The source of this port: parity target (currently v2.0.1), and the home of the reference rules, dashboard and runbook Python services use |
| `platform-fastapicommon` | No dependency either way. A service using both passes the same `CollectorRegistry` and identity values to `init_with_identity` |
| `platform-eventcommon` | Consumer (above) |
| `platform-interop-tests` | Cross-language compatibility suite run by CI's `interop` job (pinned to a commit SHA) |

### 4. Infrastructure

| System | Role |
|---|---|
| **PostgreSQL** (each consumer's own database) | The only runtime dependency. RLS policies and roles are the consumer's |
| **PgBouncer / RDS Proxy** (optional) | Transaction pooling; enable `pgbouncer_mode` |
| **Prometheus / OTel collector** | Scrape and collect through the consumer's own endpoints; pgcommon opens no listener |
| **PyPI** · **GHCR** | Host the package (Trusted Publishing) · the signed migrate image |

### 5. Development tooling

The `dev` dependency group (`uv sync` installs it by default): `pytest`, `pytest-asyncio`, `pytest-cov`, `testcontainers[postgres]`, `opentelemetry-sdk`, `httpx`, `ruff`, `mypy`, `pip-audit`, `griffe` (`make api-compat`), `import-linter` (`make arch-lint`), plus `fastapi` and `structlog` so the extras' code is tested without `--extra` flags. The build backend, `hatchling`, is pinned and hash-locked in `build-constraints.txt`.

---

## Out of scope

| Concern | Where it lives |
|---|---|
| Authentication and resolving the caller's tenant | Each service's middleware (pgcommon only validates the `GUCSet` it is handed) |
| Tables, RLS policies, database roles and grants | Each consuming service's migrations |
| Transactional outbox / inbox, event envelopes, SNS / SQS | `platform-eventcommon` |
| HTTP app factory, request middleware, logging setup, the Prometheus exposition | `platform-fastapicommon` / the service |
| Alert rules, dashboards, runbook | Go `platform-pgcommon` (they apply unchanged) |
| PgBouncer, RDS Proxy, database provisioning, backups | Platform infrastructure |
| Metric naming ratification | Observability governance |
| Synchronous (non-`asyncio`) code, multiple event loops or threads sharing a pool | Not supported: one `Pool` per event loop and process |
| Non-PostgreSQL databases | Not supported |

---

## Contributing

See [CONTRIBUTING.md](CONTRIBUTING.md) for the development setup, project layout, testing requirements, branching (trunk-based on `main`), the PR checklist and commit style. Run `make ci` before pushing. Update `CHANGELOG.md` for any change to `src/`, `pyproject.toml`, `uv.lock`, `build-constraints.txt` or the `Dockerfile` (`changelog-check.yml` enforces it), keep this README's API, configuration and error tables in step with the code, and edit a `docs/architecture/mermaid/*.mmd` file together with its embedded copy in `ARCHITECTURE.md` (`make docs-check`). [CONTRIBUTING.md § CI pipeline](CONTRIBUTING.md#ci-pipeline-go--python-mapping) maps every Go job and script to its Python counterpart.

---

## 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/lld/platform-lld-pgcommon-py.md](./docs/lld/platform-lld-pgcommon-py.md) | Low-level design: module layout, API contract, GUC contract, acquire / transaction / migration algorithms, failure handling, telemetry, invariants and open questions |
| [docs/architecture/README.md](./docs/architecture/README.md) | The standalone Mermaid sources of every ARCHITECTURE.md diagram, and how to keep them in sync |
| [docs/observability/README.md](./docs/observability/README.md) | The Enterprise Platform Observability Standard as implemented here: classification, wiring, divergences from Go, alerts and dashboard |
| [docs/observability/metrics-registry.md](./docs/observability/metrics-registry.md) | Generated metric inventory and ratification packets |
| [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](.github/SECURITY.md) | Vulnerability reporting, trust model, supported versions (the root `SECURITY.md` points here) |
| [.claude/CLAUDE.md](./.claude/CLAUDE.md) | Agent reference set for Claude Code (with `architecture.md`, `api-reference.md`, `database-schema.md`, `request-flows.md`, `operations.md`, `settings.json`): commands, critical rules and condensed contracts |

---

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.
