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

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

Package: pgcommon 0.3.1 (release status: released 2026-10-15, on PyPI) · 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 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

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.

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). 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 § 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)

Typical meanings and an example are in 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)

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. 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); 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; the standalone Mermaid sources are in docs/architecture/.

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)
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 — the source repository stays private, but the released package is public (see VERSIONING.md § Publishing to PyPI for why).

pip install pgcommon
# or with uv:
uv add pgcommon

Optional extras:

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

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

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

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

Quick start

import asyncio
import os

from pgcommon import Config, Pool, guc_set_from_context, 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)
  • guc_provider — set guc_provider=pgcommon.guc_set_from_context on every multi-tenant service; omit only for internal tooling with no RLS (RLS GUC injection)
  • with_conn / run_in_tx — wrap every DB call in one of these helpers; never reach into the pool's internals (Connection pool)
  • pgcommon.metrics.init_with_identity — call once at startup, before Pool.create, with the service's domain/service/environment; log the returned warnings (Observability)
  • pool_name — set Config.pool_name (or PG_POOL_NAME) when a service runs more than one pool, e.g. primary / replica (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)
  • Health endpoint — wire pool.health() into your /healthz handler and shed load when utilization exceeds your threshold (Health check and load shedding)
  • drain_and_close — call this in your shutdown path, not pool.close(), and log the DrainTimeoutError it raises when callbacks had to be interrupted (Graceful shutdown)
  • Transactions — never COMMIT / ROLLBACK / BEGIN inside a run_in_tx callback, and re-raise statement errors instead of swallowing them (Transactions)
  • Config warnings — log every ConfigWarning config_from_env() returns; set sslmode=verify-full (or at least require) in the DSN (Configuration reference)

Production checklist

The deployment-time complement of the 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)
  • 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, Failure model)
  • TLS — sslmode=verify-full (or at least require) in both the application's and the migration job's DSN (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, 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, 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)

Connection pool

Pool.create wraps asyncpg.create_pool 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.

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

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

Pool statistics

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

Stat fields: size, idle_size, min_size, max_size, acquired_size, acquire_count, acquire_duration (cumulative timedelta), empty_acquire_count, canceled_acquire_count, 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 
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 and tests/e2e/test_rls_pgbouncer_e2e.py for exactly this kind of test.

Wiring (recommended — contextvars-based)

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

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 for the full PgBouncer configuration guide.

FastAPI integration

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

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

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


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


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


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

create_lifespan opens the Pool once at startup (stored on app.state.pool) and calls drain_and_close() (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

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

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:

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:

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

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

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

Utilization-based load shedding

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

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

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

    return await call_next(request)

HealthStatus fields:

Field Type Description
healthy bool True when the database ping succeeded — 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.

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

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:

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

The Runner API

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

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

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

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), 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.

What this means in practice:

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

Tuning

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

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

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

❌ Forgetting guc_provider on a multi-tenant service

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

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

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

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

❌ Transactions without retry on serializable workloads

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

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

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

❌ Opening your own transaction inside with_conn in PgBouncer mode

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

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


await pool.with_conn(fn)

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

❌ pool.close() during graceful shutdown

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

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

# ✅ CORRECT: waits for in-flight callbacks to finish, then 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.

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

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

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

# ✅ CORRECT: break on pool closed; retry only on transient connectivity errors
while True:
    try:
        await pool.ping()
        break
    except 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.

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


await run_in_tx(pool, fn)

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


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


await run_in_tx(pool, fn)

❌ Double-wrapping Pool

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

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

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

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

Configuration reference

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)
guc_provider — None `Callable[[], GUCSet
pgbouncer_mode PG_BOUNCER_MODE False Enable for PgBouncer / RDS Proxy transaction pooling
log_tenant_id — False Include tenant_id in slow-query log entries — opt-in only; tenant IDs are sensitive
allow_full_statements — False Include 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). 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.

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; 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) 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): 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):

  • 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 — 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

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

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/:

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/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

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:

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

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:

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:

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

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

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

Skip Docker-dependent tests:

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

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

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

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:

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
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)
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). 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 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/, and every metric's ratification packet is in docs/observability/metrics-registry.md.

Initialise once at startup, before Pool.create:

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

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:

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 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 has the pin styles):

pip install "pgcommon==0.3.1"            # or "pgcommon[fastapi]==0.3.1" / "pgcommon[structlog]==0.3.1"
uv add "pgcommon==0.3.1"

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

TAG=v0.3.1   # 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 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.

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. The Python package keeps its own version line (0.x); the CHANGELOG names the Go version each change corresponds to.

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

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

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.

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

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

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

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

# 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.4.0 on its main (its next release, 0.3.0, requires >=0.3.0,<0.4.0) 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 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 § 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 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 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 Design decisions, layer diagrams, and the asyncio/contextvars translation notes
docs/MigrationGuide.md Concept-by-concept mapping from the Go platform-pgcommon to this port
docs/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 The standalone Mermaid sources of every ARCHITECTURE.md diagram, and how to keep them in sync
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 Generated metric inventory and ratification packets
docs/adr/ One-decision-per-file records for the riskiest translation choices (GUC propagation, cancellation/rollback safety, GUC injection timing, migration runner design)
VERSIONING.md SemVer rules, supported versions, release process
CHANGELOG.md Per-version changes
CONTRIBUTING.md Development setup, adding middleware, PR checklist
SECURITY.md Vulnerability reporting, trust model, supported versions (the root SECURITY.md points here)
.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.

Project links

Data verified by PyPI on Oct 15, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Newer release available (0.4.1)

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for bcbp_admin from gravatar.com bcbp_admin

Credits

Author: BCBP Solutions FZC LLC

GitHub Statistics

Data verified by PyPI on Oct 15, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

Provenance

Data verified by PyPI on Oct 15, 2026
Data provided by the project maintainers and verified by PyPI at the time of upload.

Full provenance

All 2 files in this release have provenance metadata.

View provenance details

License

Apache-2.0

Requires

Python >=3.12

Provides Extra

fastapi structlog

Classifiers

Metadata

Project links

Data verified by PyPI on Oct 15, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Newer release available (0.4.1)

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for bcbp_admin from gravatar.com bcbp_admin

Credits

Author: BCBP Solutions FZC LLC

GitHub Statistics

Data verified by PyPI on Oct 15, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

Provenance

Data verified by PyPI on Oct 15, 2026
Data provided by the project maintainers and verified by PyPI at the time of upload.

Full provenance

All 2 files in this release have provenance metadata.

View provenance details

License

Apache-2.0

Requires

Python >=3.12

Provides Extra

fastapi structlog

Classifiers

Release files for pgcommon 0.3.1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for pgcommon 0.3.1
File Size Uploaded Provenance
pgcommon-0.3.1.tar.gz 793.1 kB With provenance Details

Built distribution (wheel)

Table of built distributions (wheels) for pgcommon 0.3.1
File Interpreter ABI Platform
pgcommon-0.3.1-py3-none-any.whl 197.4 kB With provenance Python 3 none any Details

Total release size: 990.5 kB

Release files / pgcommon-0.3.1.tar.gz

Download URL pgcommon-0.3.1.tar.gz
Size 793.1 kB
Tags Source
SHA-256 checksum
How to use checksums 		2ac7bf1cac0f4e9ee5176ae6f63d1ac4e5ddf28f21b4f785e7f2c3358e1ac31b
BLAKE2b-256 checksum
How to use checksums 		2899e62678cd14a0169be6851ed44fb927da47355e386fbce36205111d9041f9
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Oct 15, 2026.

Transparency log
Identity
Publishing platform
GitHub Actions
Publishing repository
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
Publishing commit
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/tree/c101015ef3e87fa069d4c50140eb30629d58b1ee
Workflow
Publishing configuration
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/blob/c101015ef3e87fa069d4c50140eb30629d58b1ee/.github/workflows/release.yml
Publishing logs
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/actions/runs/38777472743/attempts/1
Artifact
Subject
pgcommon-0.3.1.tar.gz
SHA-256 checksum
2ac7bf1cac0f4e9ee5176ae6f63d1ac4e5ddf28f21b4f785e7f2c3358e1ac31b Verifying attestations

Release files / pgcommon-0.3.1-py3-none-any.whl

Download URL pgcommon-0.3.1-py3-none-any.whl
Size 197.4 kB
Tags Python 3
SHA-256 checksum
How to use checksums 		3122fdb4c06cc0c6805922853cbd88f8dc6ed5fc6eab8f918e394087714f090d
BLAKE2b-256 checksum
How to use checksums 		b5330352e889af9424826bb8e290d176b12c80df0057019d203a30bcf97a4212
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Oct 15, 2026.

Transparency log
Identity
Publishing platform
GitHub Actions
Publishing repository
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
Publishing commit
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/tree/c101015ef3e87fa069d4c50140eb30629d58b1ee
Workflow
Publishing configuration
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/blob/c101015ef3e87fa069d4c50140eb30629d58b1ee/.github/workflows/release.yml
Publishing logs
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/actions/runs/38777472743/attempts/1
Artifact
Subject
pgcommon-0.3.1-py3-none-any.whl
SHA-256 checksum
3122fdb4c06cc0c6805922853cbd88f8dc6ed5fc6eab8f918e394087714f090d Verifying attestations

Security for pgcommon 0.3.1

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

Full provenance

All 2 files in this release have been uploaded with provenance metadata.

Select from 2 files

Attestation for pgcommon-0.3.1.tar.gz

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Oct 15, 2026.

Transparency log
Identity
Publishing platform
GitHub Actions
Publishing repository
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py
Publishing commit
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/tree/c101015ef3e87fa069d4c50140eb30629d58b1ee
Workflow
Publishing configuration
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/blob/c101015ef3e87fa069d4c50140eb30629d58b1ee/.github/workflows/release.yml
Publishing logs
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-pgcommon-py/actions/runs/38777472743/attempts/1
Artifact
Subject
pgcommon-0.3.1.tar.gz
SHA-256 checksum
2ac7bf1cac0f4e9ee5176ae6f63d1ac4e5ddf28f21b4f785e7f2c3358e1ac31b Verifying attestations

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page