Metadata-Version: 2.4
Name: audit-framework-postgres
Version: 0.1.2
Summary: PostgreSQL adapters for audit-framework — append-only AuditStore, NotificationStore, and a LISTEN/NOTIFY EventBus.
Project-URL: Homepage, https://github.com/vanmarkic/audit-logger
Project-URL: Repository, https://github.com/vanmarkic/audit-logger
License-Expression: MIT
Keywords: append-only,asyncpg,audit,audit-log,listen-notify,plugin,postgresql
Requires-Python: >=3.11
Requires-Dist: audit-framework<0.2,>=0.1
Provides-Extra: asyncpg
Requires-Dist: asyncpg>=0.29; extra == 'asyncpg'
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == 'dev'
Description-Content-Type: text/markdown

# audit-framework-postgres

The **PostgreSQL adapters** for [`audit-framework`](../audit-framework), three
implementations sharing one database:

| Adapter | Port | Role |
|---|---|---|
| `PostgresAuditStore` | `AuditStore` | append-only, queryable system of record for the audit log (vs. the best-effort `ExternalSink` fan-out) |
| `PostgresNotificationStore` | `NotificationStore` | the **mutable** notification delivery lifecycle (`pending → delivered → read` / `failed`) |
| `PgListenNotifyBus` | `EventBus` | cross-context bus on `LISTEN/NOTIFY` (e.g. audit → case-management `escalation.requested`) without Redis/NATS |

Ships the `audit_log` schema (compact `LISTEN/NOTIFY` trigger + append-only
guards) and the `notifications` schema.

## Install

```bash
pip install audit-framework-postgres            # bring your own executor
pip install audit-framework-postgres[asyncpg]   # + asyncpg for a real pool
```

## Use

```python
import asyncpg
from audit_framework_postgres import PostgresAuditStore, apply_schema
from audit_framework.core.middlewares.store import StoreMiddleware

pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)
await apply_schema(pool, app_role="app_user")     # create table + triggers (run once, as a privileged role)

store = PostgresAuditStore(pool)                  # the asyncpg pool *is* the executor
pipeline.use(StoreMiddleware(store))              # now events are persisted authoritatively
```

Query it back (this is what backs an admin audit-log view — sinks can't do this):

```python
await store.query({"actor_id": "alice", "action": "DELETE", "from": "2026-06-01", "to": "2026-07-01"},
                  offset=0, limit=50)
await store.get_by_resource("contract", "c-42")
```

### Notifications + event bus

```python
from audit_framework_postgres import (
    PostgresNotificationStore, PgListenNotifyBus, apply_notifications_schema,
)

await apply_notifications_schema(pool)            # create the notifications table (run once)
notifications = PostgresNotificationStore(pool)   # mutable delivery lifecycle
await notifications.get_unread("alice")           # initial UI load (not read, not failed)

# The bus needs a *dedicated* asyncpg connection (LISTEN holds it open), not the pool:
conn = await asyncpg.connect(dsn)
bus = PgListenNotifyBus(conn, max_queue=1000)     # per-subscription buffer (bounded)
sub = await bus.subscribe("escalation.requested")
msg = await sub.receive(timeout=30)               # None on timeout; raises once closed
# sub.dropped counts messages evicted when a slow consumer overflows the buffer
```

`publish()` raises `NotifyPayloadTooLargeError` (before the DB call) if the
serialised payload would reach Postgres's NOTIFY size limit — the backend rejects
payloads of **8000 bytes or more**, so keep bus payloads to identifiers/ids and
let the subscriber fetch the full row by id (the same trade the `audit_log`
NOTIFY trigger makes). Each subscription's buffer is **bounded** (`max_queue` ≥ 1;
default 1000); when a slow consumer overflows it the oldest message is evicted,
logged once, and counted in `sub.dropped` rather than growing without limit.

All three register under the `postgres` provider name via the
`audit_framework.plugins` entry point (`audit_store`, `notification_store`,
`event_bus`), so they're discoverable through the registry.

## No hard driver dependency

All DB access goes through an injected **`Executor`** — anything exposing
`fetchval` / `fetch` / `execute` (an `asyncpg` pool or connection satisfies this
structurally). So the SQL logic is fully unit-testable without a database
(50 stdlib-only tests use a fake executor/connection), and you control pooling.
The `PgListenNotifyBus` likewise depends only on a `ListenConnection`
(`execute` / `add_listener` / `remove_listener`), satisfied structurally by a
dedicated `asyncpg` connection.

## Schema design (validated against PostgreSQL guidance)

`schema_sql()` / `apply_schema()` emit:

- an **append-only `audit_log`** table — `REVOKE UPDATE, DELETE` from the app
  role **and** a `BEFORE UPDATE OR DELETE` guard trigger that raises, so the log
  is immutable even for the table owner (only a superuser can bypass it). Run
  the app under an **unprivileged, non-owning** role that can only `INSERT`/`SELECT`;
- a **compact `AFTER INSERT` `pg_notify`** trigger on channel `audit_events` that
  sends **identifiers only** — never the unbounded `changes` diff — so every
  payload stays well under PostgreSQL's hard **8000-byte `NOTIFY` limit**.
  A `LISTEN` consumer fetches the full row by id when it needs detail (this is
  also the upgrade path to the multi-worker event bus, AD-6);
- `request_id` / `ip_address` as **`TEXT`** (not `UUID`/`INET`), because the
  event model treats them as free-form strings and redaction can replace
  `ip_address` with a mask.

> **Tamper-evidence:** `REVOKE` + guards make the log *append-only*, but a
> superuser can still rewrite history. For tamper-*detection*, add per-row hash
> chaining (`prev_hash`/`hash`) on top — a natural follow-up.

`notifications_schema_sql()` / `apply_notifications_schema()` emit the
**`notifications`** table. It is intentionally **not** append-only — a
notification has a mutable lifecycle (`pending → delivered → read` / `failed`),
so there is no REVOKE/guard trigger. `created_at` / `delivered_at` are `TEXT`
(not `TIMESTAMPTZ`) so the store round-trips the event model's free-form ISO-8601
strings without a custom asyncpg timestamptz codec; ISO-8601 UTC strings sort
chronologically for the inbox (newest-first) and pending-worker (oldest-first)
read paths.

### SQL-injection posture
The table name is validated against a strict identifier whitelist at
construction; every value is a bound `$n` parameter; query filters are mapped
through a fixed column allow-list (unknown keys raise `ValueError`). The only
interpolation is the validated table name (Postgres can't bind an identifier as
a parameter).

The whitelist accepts a **bare or schema-qualified** name (`audit_log` or
`reporting.audit_log`) — exactly the set the `schema_sql()` /
`notifications_schema_sql()` DDL generators also accept, so any table a store
can target is one the package can also create. For a schema-qualified table the
generated DDL is self-contained: it emits a leading `CREATE SCHEMA IF NOT
EXISTS`, references the table qualified, derives the index/trigger *names* from
the bare table part (Postgres forbids schema-qualifying those), and places the
trigger functions in the table's schema.

## Development

```bash
pip install -e ".[dev]"
pytest        # 50 stdlib-only tests (fake executor/connection; no database needed)
```

A live-DB integration test (real `asyncpg` + a throwaway Postgres) is the
recommended next layer; the unit suite already pins the generated SQL, the
parameter binding, the append-only DDL, and the compact-notify payload.

## License

MIT


## For AI agents & coding assistants

This package ships its agent guide — [`AGENTS.md`](./AGENTS.md) — **inside the
wheel** (installed at `<site-packages>/audit_framework_postgres/AGENTS.md`). Read it offline, with no
docs site and no network, even from an airgapped Nexus PyPI mirror:

```bash
python -m audit_framework_postgres
python -c "import audit_framework_postgres; print(audit_framework_postgres.overview())"
```

`AGENTS.md` is the single source of truth: mental model, one runnable quickstart,
and the exact public API. `audit_framework_postgres.overview()` returns it, and `tests/guide_test.py`
compiles its examples so the guide can't drift from the code.
