pg-infra

Safe, typed PostgreSQL application infrastructure built on psycopg 3.

This is not a thin wrapper. It turns raw psycopg into the pieces every app rebuilds by hand — cursor pagination, set-based bulk writes, full-text and fuzzy search, and transactions that retry correctly under serialization failures — with a sync and an async API in parallel.

• Injection-safe by construction. Every table, column, and value is composed through psycopg.sql (Identifier/Placeholder). The SQL builders are pure and never use f-strings or %-formatting to assemble queries.
• Composes with your transactions. Capability functions take a live Connection/AsyncConnection you pass in; they never open hidden transactions except where documented (fuzzy search, which needs a per-query GUC).
• Honest about guarantees. Keyset pagination requires a unique tiebreaker (documented and validated). Retries fire only on SQLSTATE 40001/40P01, never blindly. Errors are wrapped, never swallowed.

Requires Python ≥ 3.11 and PostgreSQL. Fuzzy search needs the pg_trgm extension.

Install

pip install pg-infra          # includes psycopg[binary,pool]

Quick start

from pg_infra import PgConfig, PgClient, paginate, bulk_insert, transaction

cfg = PgConfig(host="localhost", dbname="app", user="app", password="secret")
client = PgClient.from_config(cfg)

with client.connection() as conn:
    # Bulk insert (executemany under the hood)
    bulk_insert(conn, "users", ["name", "email"], [("Ada", "ada@x.io"), ("Alan", "alan@x.io")])

    # Keyset pagination — O(page) no matter how deep
    page = paginate(conn, "users", order_by=[("created_at", "DESC"), ("id", "DESC")], limit=50)
    for row in page.items:
        ...
    if page.has_more:
        next_page = paginate(
            conn,
            "users",
            order_by=[("created_at", "DESC"), ("id", "DESC")],
            limit=50,
            cursor=page.next_cursor,
        )

client.close()

Config from the environment

cfg = PgConfig.from_env(
    "PG_"
)  # PG_HOST, PG_PORT, PG_DBNAME, PG_USER, PG_PASSWORD, PG_CONNINFO, ...

No credentials live in source: pass them in or read them from the environment.

Capabilities

Timeouts

Three distinct layers, deliberately separate:

• connect_timeout → libpq connect timeout (in the conninfo).
• statement_timeout_ms → Postgres statement_timeout GUC, applied per connection by the pool's configure hook.
• pool_timeout → how long to wait for a free pooled connection.

Async

Every capability has an a-prefixed async twin with an identical signature; only await differs.

from pg_infra import AsyncPgClient, apaginate

client = await AsyncPgClient.from_config(cfg)
async with client.connection() as conn:
    page = await apaginate(conn, "users", order_by=[("id", "ASC")], limit=100)
await client.aclose()

On Windows, psycopg's async connections cannot run on the default ProactorEventLoop; start your entry point on a selector loop instead. async_pg_pool raises ConnectionError saying so rather than letting the pool time out.

asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)  # 3.12+

Guarantees and non-guarantees

• Keyset pagination is correct only with a unique tiebreaker. End order_by with a unique column (usually the primary key), or pages can skip or repeat rows. The order_by columns must be in the selected columns (rows come back as dicts so the cursor can read them).
• Cursors are opaque, not secret and not tamper-proof. They are base64-wrapped JSON; a corrupt or tampered cursor raises PaginationError rather than being trusted. Sort keys may be JSON scalars or datetime/date/time/Decimal/UUID/bytes, which round-trip as their original Python type; anything else raises PaginationError.
• Isolation must be set before the connection's first query. PostgreSQL accepts SET TRANSACTION only as a transaction's opening statement, so transaction(conn, isolation=...) requires an idle connection and raises TransactionError otherwise. Commit, roll back, or take a fresh connection first.
• Retries are not blind. retry_on_serialization_failure re-runs only on serialization failure (40001) and deadlock (40P01); every other error propagates immediately. The decorated function must open its own transaction so each attempt is a clean re-run.
• No hidden COPY↔executemany switch. You pick copy_load (fast, no RETURNING) or the executemany writers explicitly; behaviour never changes silently as row counts grow.
• Full-text search picks exactly one source. Provide text_columns (built on the fly, zero setup) or tsvector_column (precomputed + GIN-indexed, recommended for production) — never both.

Development

pip install -e ".[dev]"
ruff check . && ruff format --check .
mypy src/pg_infra
pytest tests/unit            # pure unit tests, no database required
# integration tests (need a live PostgreSQL) are marked `integration` and deferred

License

MIT