Metadata-Version: 2.5
Name: pg-infra
Version: 0.1.0
Summary: Developer-friendly production layer over PostgreSQL: pagination, bulk ops, transactions, full-text + fuzzy search (sync + async).
License: MIT
Keywords: async,full-text-search,pagination,pg_trgm,postgresql,psycopg
Requires-Python: >=3.11
Requires-Dist: psycopg[binary,pool]<4,>=3.2
Provides-Extra: dev
Requires-Dist: mypy>=1.11; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Description-Content-Type: text/markdown

# pg-infra

Safe, typed PostgreSQL application infrastructure built on [psycopg 3](https://www.psycopg.org/psycopg3/).

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

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

## Quick start

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

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

| Area | Sync | Async | Notes |
|------|------|-------|-------|
| Config | `PgConfig` | — | frozen, validated, `from_env` |
| Pool / connections | `PgClient`, `pg_pool` | `AsyncPgClient`, `async_pg_pool` | psycopg's own pools; per-connection `statement_timeout` |
| Transactions | `transaction` | `atransaction` | isolation / read-only / deferrable |
| Safe retries | `retry_on_serialization_failure` | `aretry_on_serialization_failure` | `40001`/`40P01` only, backoff + jitter |
| Keyset pagination | `paginate` | `apaginate` | opaque cursor, `LIMIT n+1` for `has_more` |
| Offset pagination | `paginate_offset` | `apaginate_offset` | fallback; degrades with depth |
| Bulk insert | `bulk_insert` | `abulk_insert` | `executemany`, optional `RETURNING` |
| Upsert | `upsert` | `aupsert` | `ON CONFLICT DO UPDATE`/`DO NOTHING` |
| Bulk update | `bulk_update` | `abulk_update` | by key column |
| COPY load | `copy_load` | `acopy_load` | fastest path, no `RETURNING` |
| Full-text search | `search.search` / `fts_search` | `search.asearch` / `afts_search` | `tsvector`/`tsquery`, `ts_rank` |
| Fuzzy search | `fuzzy.search` / `trgm_search` | `fuzzy.asearch` / `atrgm_search` | `pg_trgm` similarity |

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

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

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

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