Quickstart¶
Three small programs: one with a sync Session, one with an AsyncSession, and a FastAPI application. Each is a complete script in the repository's examples/ directory, and the test suite runs all three against PostgreSQL, so the code below works as shown.
You need a PostgreSQL 14+ database and the extras for your driver:
pip install "sqlalchemy-audit-trail[psycopg]" # sync
pip install "sqlalchemy-audit-trail[asyncpg]" # async
pip install "sqlalchemy-audit-trail[asyncpg,fastapi]" # FastAPI
Sync¶
Models and events¶
A model is audited when it mixes in Audited. Per-column policies go in mapped_column(info={"audit": ...}), model options in __audit__. Domain events are members of an AuditEvent enum.
class Base(DeclarativeBase):
pass
class Invoice(Base, Audited):
__tablename__ = "invoice"
id: Mapped[int] = mapped_column(primary_key=True)
customer: Mapped[str]
total: Mapped[Decimal]
status: Mapped[str] = mapped_column(default="draft")
# Never stored in the audit log; only "***" shows that it changed.
payment_token: Mapped[str | None] = mapped_column(info={"audit": "redact"})
__audit__ = AuditOptions(label=lambda invoice: f"Invoice for {invoice.customer}")
class BillingEvent(AuditEvent):
INVOICE_SENT = event("billing.invoice_sent", Severity.NOTICE)
Setup¶
AuditTrail holds the configuration; install() registers the listeners on your session factory only, not on every Session in the process. The audit tables are partitioned, and PostgreSQL has no partition to put a row in until ensure_partitions() has created the current month's partitions: run it before the first write and then on a schedule (see operations).
engine = create_engine(DATABASE_URL)
audit = AuditTrail(engine, events=[BillingEvent])
SessionLocal = sessionmaker(engine)
audit.install(SessionLocal)
def create_tables() -> None:
# In an application, this belongs in a migration (see the partitions guide).
with engine.begin() as connection:
Base.metadata.create_all(connection)
create_audit_tables(connection, audit.tables, audit.severities)
# Monthly partitions: run this on a schedule too, not only at start-up.
audit.maintenance.ensure_partitions()
In an application, create the tables in a migration rather than at start-up; see partitions and retention.
Writing¶
Changes are recorded when the session flushes, with the context that is active at that moment. audit.log() records a domain event in the same transaction.
def work() -> None:
with (
audit.context(channel="cli"),
SessionLocal() as session,
):
audit.set_actor(Actor(type="user", id="42", label="ada@example.com"))
invoice = Invoice(customer="Acme", total=Decimal("120.00"))
session.add(invoice)
session.commit() # entity.created
invoice.status = "sent"
invoice.payment_token = "tok_123"
audit.log(
session,
BillingEvent.INVOICE_SENT,
obj=invoice,
payload={"channel": "email"},
)
session.commit() # entity.updated and billing.invoice_sent
Reading¶
object_history() pages through one record's history, grouped by database transaction, newest first:
def show_history() -> None:
with SessionLocal() as session:
page = audit.query.object_history(session, "Invoice", "1")
for group in page.groups:
header = group.transaction
print(f"{header.issued_at:%Y-%m-%d %H:%M} by {header.actor_label}")
for activity in group.activities:
data = activity["data"]
print(" ", activity["verb"], data.get("changes") or data["payload"])
Running the script prints something like:
2026-09-25 22:22 by ada@example.com
billing.invoice_sent {'channel': 'email'}
entity.updated {'status': ['draft', 'sent'], 'payment_token': [None, '***']}
2026-09-25 22:22 by ada@example.com
entity.created {'id': [None, 1], 'total': [None, '120.00'], 'status': [None, 'draft'], 'customer': [None, 'Acme'], 'payment_token': [None, None]}
payment_token is redacted: the log shows that it was set, never its value.
Async¶
Under asyncio the listeners run on the Session that AsyncSession drives, so the async_sessionmaker needs a sync_session_class of your own; install() refuses the base Session, which would audit every session in the process.
class AppSession(Session):
"""The listeners are registered on this class, not on every Session."""
engine = create_async_engine(DATABASE_URL)
audit = AuditTrail(engine, events=[])
SessionLocal = async_sessionmaker(engine, sync_session_class=AppSession)
audit.install(SessionLocal)
async def create_tables() -> None:
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.run_sync(create_audit_tables, audit.tables, audit.severities)
await audit.maintenance.aensure_partitions()
This example also tracks a relationship and points each task at its project as the target, so a task's changes appear in the project's history:
class Project(Base, Audited):
__tablename__ = "project"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
tasks: Mapped[list[Task]] = relationship(back_populates="project")
__audit__ = AuditOptions(
label=lambda project: project.name,
track_relationships={"tasks"},
)
class Task(Base, Audited):
__tablename__ = "task"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
project_id: Mapped[int] = mapped_column(ForeignKey("project.id"))
project: Mapped[Project] = relationship(back_populates="tasks")
__audit__ = AuditOptions(
label=lambda task: task.title,
# Changes of a task also show up in its project's history.
target=lambda task: ("Project", task.project_id),
)
async def work() -> None:
actor = Actor(type="user", id="7", label="grace@example.com")
async with audit.context(channel="worker"):
audit.set_actor(actor)
async with SessionLocal() as session:
project = Project(id=1, name="Launch", tasks=[Task(id=1, title="Draft")])
session.add(project)
await session.commit()
async with SessionLocal() as session:
# A tracked collection must be loaded before it is changed:
# AsyncSession cannot lazy-load it.
project = await session.get_one(
Project, 1, options=[selectinload(Project.tasks)]
)
project.tasks.append(Task(id=2, title="Review"))
await session.commit()
AsyncSession cannot lazy-load, so load a tracked collection (here with selectinload) before you change it, and refresh an object expired by a commit before assigning to it; otherwise you get an AsyncLoadError naming the attribute. See limitations.
Every query has an a-prefixed async variant:
async def show_history() -> None:
async with SessionLocal() as session:
page = await audit.query.aobject_history(session, "Project", "1")
for group in page.groups:
for activity in group.activities:
print(
activity["verb"],
activity["object_type"],
activity["object_label"],
activity["data"]["changes"],
)
entity.created Task Review {'id': [None, 2], 'title': [None, 'Review'], 'project_id': [None, 1]}
entity.updated Project Launch {'tasks': {'added': ['2'], 'removed': []}}
entity.created Project Launch {'id': [None, 1], 'name': [None, 'Launch'], 'tasks': {'added': ['1'], 'removed': []}}
entity.created Task Draft {'id': [None, 1], 'title': [None, 'Draft'], 'project_id': [None, 1]}
FastAPI¶
AuditMiddleware fills in the request part of the context (client address, user agent, method, path, request id); session_dependency() hands out sessions of the installed factory; set_actor() records who is calling once they are authenticated.
engine = create_async_engine(DATABASE_URL)
audit = AuditTrail(engine, events=[])
SessionLocal = async_sessionmaker(
engine, sync_session_class=AppSession, expire_on_commit=False
)
audit.install(SessionLocal)
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# Tables belong in migrations; partitions are also created on a schedule.
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.run_sync(create_audit_tables, audit.tables, audit.severities)
await audit.maintenance.aensure_partitions()
yield
await audit.adispose()
await engine.dispose()
app = FastAPI(lifespan=lifespan)
app.add_middleware(AuditMiddleware)
get_session = session_dependency(SessionLocal)
def current_user(x_user: Annotated[str, Header()]) -> str:
# Stand-in for real authentication. A sync dependency runs in the thread
# pool; set_actor still reaches the request's audit context.
set_actor(Actor(type="user", id=x_user), auth_method="header")
return x_user
class NoteIn(BaseModel):
title: str
@app.post("/notes", status_code=201)
async def create_note(
note_in: NoteIn,
user: Annotated[str, Depends(current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
) -> dict[str, int]:
note = Note(title=note_in.title)
session.add(note)
await session.commit()
return {"id": note.id}
@app.patch("/notes/{note_id}")
async def rename_note(
note_id: int,
note_in: NoteIn,
user: Annotated[str, Depends(current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
) -> dict[str, int]:
note = await session.get(Note, note_id)
if note is None:
raise HTTPException(status_code=404)
note.title = note_in.title
await session.commit()
return {"id": note_id}
@app.get("/notes/{note_id}/history")
async def note_history(
note_id: int, session: Annotated[AsyncSession, Depends(get_session)]
) -> list[dict[str, Any]]:
page = await audit.query.aobject_history(session, "Note", str(note_id))
return [
{
"actor_id": activity["actor_id"],
"verb": activity["verb"],
"changes": activity["data"]["changes"],
"path": activity["data"]["context"].get("path"),
}
for group in page.groups
for activity in group.activities
]
After POST /notes as ada and PATCH /notes/1 as bob, GET /notes/1/history returns:
[
{"actor_id": "bob", "verb": "entity.updated", "changes": {"title": ["Hello", "Hi"]}, "path": "/notes/1"},
{"actor_id": "ada", "verb": "entity.created", "changes": {"id": [null, 1], "title": [null, "Hello"]}, "path": "/notes"}
]
The FastAPI guide covers proxies, request ids and sync applications.