audit-framework-postgres

The PostgreSQL adapters for audit-framework, three implementations sharing one database:

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

Install

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

Use

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

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

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

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

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.