sqlalchemy-audit-trail

Audit trail for SQLAlchemy on PostgreSQL: automatic entity change diffs and explicit domain events in one chronological, partitioned log, with who, where and how on every entry. Not a version-table library like sqlalchemy-continuum.

Alpha. The API may change before 1.0.

Install

pip install "sqlalchemy-audit-trail[psycopg]"   # Session on psycopg 3
pip install "sqlalchemy-audit-trail[asyncpg]"   # AsyncSession on asyncpg
                                                # more extras: asyncio, pydantic (payload schemas), fastapi

Requires Python 3.10+, SQLAlchemy 2.0 or 2.1, and PostgreSQL 14+.

Quickstart

Entity changes: mark a model Audited, install the listener, and every flush records entity.created / entity.updated / entity.deleted with {"field": [old, new]} diffs:

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

from audit_trail import Audited, AuditOptions, AuditTrail
from audit_trail.migrations import create_audit_tables


class Base(DeclarativeBase):
    pass


class Invoice(Base, Audited):
    __tablename__ = "invoice"

    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str] = mapped_column(default="draft")
    payment_token: Mapped[str | None] = mapped_column(info={"audit": "redact"})

    __audit__ = AuditOptions(label=lambda invoice: f"Invoice {invoice.id}")


engine = create_engine("postgresql+psycopg://localhost/app")
audit = AuditTrail(engine, events=[])
SessionLocal = sessionmaker(engine)
audit.install(SessionLocal)

with engine.begin() as connection:  # in a migration, in a real application
    Base.metadata.create_all(connection)
    create_audit_tables(connection, audit.tables, audit.severities)
audit.maintenance.ensure_partitions()  # and on a schedule

with audit.context(actor_type="user", actor_id="42", actor_label="ada@example.com"):
    with SessionLocal() as session:
        invoice = Invoice()
        session.add(invoice)
        session.commit()
        invoice.status = "sent"
        session.commit()

with SessionLocal() as session:
    for group in audit.query.object_history(session, "Invoice", "1").groups:
        for activity in group.activities:
            print(activity["verb"], activity["data"]["changes"])
# entity.updated {'status': ['draft', 'sent']}
# entity.created {'id': [None, 1], 'status': [None, 'draft'], 'payment_token': [None, None]}

Domain events: declare what else is worth auditing once, with a severity, then log it:

from audit_trail import AuditEvent, Severity, event


class AuthEvent(AuditEvent):
    LOGIN = event("auth.login", Severity.INFO)
    LOGIN_FAILED = event("auth.login_failed", Severity.WARNING, durable=True)


audit = AuditTrail(engine, events=[AuthEvent])
audit.log(session, AuthEvent.LOGIN_FAILED, payload={"reason": "bad password"})

Docs: wiktorb2004.github.io/sqlalchemy-audit-trail. Runnable, tested quickstarts for sync, async and FastAPI: examples/. A complete multi-tenant FastAPI app with login events, history, an activity feed and GDPR erasure, with a curl walkthrough: examples/fastapi_app.

Audit rows are written in your transaction: roll back and they are gone with the change. A failed audit write does not fail your transaction by default (on_error="log"). durable events are committed on their own connection and survive a rollback; fail_closed events raise AuditWriteError when they cannot be written.

Results

Measured on a laptop: PostgreSQL 18 in Docker with stock settings, one connection, synthetic data. Protocol, environment and limits: benchmark/README.md.

Reads on 5M entries in 80 partitions: the first page of list_groups takes 18.5 ms at p50 (15.1 ms with a 30-day window), of which PostgreSQL spends under 1 ms; a next page reads 5 of 80 partitions. Queries with no time bound (object_history, access_summary, by actor_id) read every monthly partition and take 10 to 22 ms here.

Why this instead of sqlalchemy-continuum, triggers or pgaudit

sqlalchemy-continuum copies every changed row into a version table per model. That answers "what did this row look like then", but a login, a download or someone viewing sensitive data has no row to version. Database triggers and pgaudit see statements, not the application's user, request or intent.

This library stores what changed as diffs, next to the domain events you declare and the request context (actor, IP, user agent, path, channel), in one log you can list chronologically, per object or per actor. The trade-off: there is no revert and no state at a point in time. See migrating from continuum and limitations.

Design

One partitioned log

Two tables: audit_transaction (who, where and how, once per database transaction) and audit_activity (one row per object and flush, or per event). Both are partitioned by severity, then by month, so retention drops whole partitions per severity and reads touch only the partitions they need. There is no default partition: schedule ensure_partitions() (see operations).

In your transaction, unless you say otherwise

Privacy by default, erasure on request

Per-column exclude, redact and hash policies, Pseudonymized payload fields with versioned keys, and scrub() / scrub_actor() to erase an object's values or an actor's personal context, each recorded as an audit.scrubbed entry. See privacy.

Reads fail closed

list_groups() lists entries grouped by transaction, with keyset pagination. Visibility restrictions fail closed: None means no restriction, an empty set allows nothing, and a NULL column never passes a restriction that is set.

Sync, async and FastAPI

Session and AsyncSession on psycopg 3 or asyncpg. audit_trail.integrations.fastapi adds ASGI middleware for the request context (the client IP is read from X-Forwarded-For only behind trusted proxies), a session dependency, and set_actor() for after authentication.

Tests

Tests under tests/db run against a real PostgreSQL. With Docker running, a throwaway postgres:18 container is started with testcontainers.

uv sync --frozen

uv run pytest tests/unit                          # no database needed
uv run pytest                                     # everything, needs Docker
AUDIT_TEST_PG_IMAGE=postgres:14 uv run pytest     # oldest supported PostgreSQL

uv run ruff check . && uv run mypy

To use a server you already have, set AUDIT_TEST_DATABASE_URL. CI runs PostgreSQL 14, 16, 17 and 18, SQLAlchemy 2.0 and 2.1, and Python 3.10 to 3.13.

Docs

uv sync --group docs
uv run mkdocs serve

Published at wiktorb2004.github.io/sqlalchemy-audit-trail.

Benchmark

The numbers above come from benchmark/. It needs Docker (or --database-url) and takes about 12 minutes:

uv run python -m benchmark.run_benchmark

Contributing

Issues and pull requests are welcome. See CONTRIBUTING.md and the Code of Conduct. To report a vulnerability, use SECURITY.md.

License

MIT. See LICENSE. Changelog: CHANGELOG.md. Cite this repo with CITATION.cff.