Auditing with SQLAlchemy event listeners

Capsule overview

Capsules 03, 04, and 05 implemented auditing with PostgreSQL triggers. It's the default approach and, when it's viable, the right one. But "viable" isn't always. There are teams on managed databases that restrict custom functions (Aurora Serverless v1, some RDS configurations, certain Heroku Postgres plans). There are teams that need the tests to run against SQLite (with no PL/pgSQL triggers) and the same logic to run against PostgreSQL in production. There are teams that prefer to keep all the audit logic in Python for codebase uniformity, debugging, and onboarding. For those cases, SQLAlchemy offers a native mechanism: event.listens_for(Session, "after_flush").

This capsule teaches you to implement the same audit log from capsule 03, but at the ORM level instead of the DB level. You're going to understand the mechanics of the before_flush and after_flush events, how to capture modified entities with session.dirty, session.new, and session.deleted, how to generate the diff using SQLAlchemy's inspector, and how to write the audit rows in the same transaction. You're also going to see the critical detail that differentiates the approach: the listener lives in Python, so bypassing it is trivial — a script with psql, an op.execute() in Alembic, a session.execute(text(...)) that doesn't use the ORM, and the auditing doesn't run. You're going to learn to mitigate that and to know when the bypass is acceptable.

By the end you'll have criteria for choosing between triggers and event listeners based on the context, complete code for the Python approach (which you can copy into a project), and tests that verify the listener fires correctly. It's the capsule that closes the module's architectural decision: three approaches in SQL (capsules 03-05), one in Python (this one), pick according to the case.

Mental model: a trigger in the DB vs a trigger in the ORM

Think about the difference with an analogy. A PostgreSQL trigger is like a guard at a building's front door — it checks EVERYONE who comes in, whether a regular visitor, an employee, a delivery person, or a construction worker. It's exhaustive: nothing gets past without the guard seeing it.

A SQLAlchemy event listener is like a guard at the 5th-floor reception — it checks whoever enters the 5th floor through the main door, but employees with a service-elevator access card can go up without passing reception. It's complete only for whoever follows the expected path.

That's the operational difference. Triggers are unavoidable: any INSERT/UPDATE/DELETE on the table fires them, whether it came from the ORM, from raw SQL, from a script, or from psql. Event listeners are avoidable: any code that bypasses the ORM (e.g. session.execute(text("UPDATE tasks SET ..."))) also bypasses the listener.

This is a trade-off, not an absolute problem. If your team has discipline ("every change goes through the ORM"), event listeners work. If your team is heterogeneous (data scientists running raw SQL, DBAs doing maintenance with psql, batch jobs using non-ORM tools), triggers are the only guarantee. The decision is context-dependent.

When to choose event listeners instead of triggers

There are three situations where event listeners are the right answer:

Situation 1: the DB doesn't allow custom triggers

• Aurora Serverless v1: restrictions on PL/pgSQL functions.
• Heroku Postgres basic plans: some restrictions depending on the plan.
• Some Managed Postgres offerings on smaller clouds: check what your vendor allows.
• Non-PostgreSQL DBs: SQLite (doesn't support complex triggers), MySQL in some configurations (more limited triggers), CockroachDB (different syntax).

If you work in a stack where you can't create custom PL/pgSQL functions, an event listener is your only option.

Situation 2: a heterogeneous stack with different DBs in different environments

Some teams use SQLite in tests (faster, no external deps) and PostgreSQL in production. SQL triggers are vendor-specific — a PostgreSQL trigger doesn't work in SQLite. Maintaining two versions of the trigger is technical debt.

Event listeners live in Python, they're DB-agnostic. The same logic runs on SQLite (tests) and PostgreSQL (production). That's valuable for fast, deterministic test suites.

Situation 3: the team prefers codebase uniformity

Some teams make a deliberate decision: "all the logic lives in Python." Legitimate reasons: unified debugging (logs and stack traces in the same place), easier onboarding (no need to learn PL/pgSQL), safer refactoring (linters and type checkers cover everything).

The cost is losing the "no change escapes" guarantee. But if the team accepts it consciously and mitigates it (with tests, code review, lint rules), the approach is valid.

The mechanics: before_flush and after_flush

SQLAlchemy 2.0 exposes events in the session's lifecycle. The two relevant to auditing are:

• before_flush: fires before SQLAlchemy emits the SQL for the pending changes. It gives access to session.new, session.dirty, session.deleted. Useful when you need to modify the entities before persisting or compute diffs by comparing with the original values.
• after_flush: fires after SQLAlchemy emitted the SQL but before the commit. New entities have their (autogenerated) IDs. Useful for INSERTing into the log with the ID already known.

For an audit log we want after_flush: we need the IDs of new entities, and we want the INSERT into the log to happen in the same transaction as the changes.

The basic listener

# app/audit_listener.py
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import event, inspect
from sqlalchemy.orm import Session

from app.models import Task  # assumes we have the models
from app.audit_models import TaskAuditLog  # the audit table as a SQLAlchemy model


# Columns we NEVER want to audit (same as in capsule 03)
EXCLUDED_COLUMNS_BY_TABLE = {
    "tasks": {"updated_at", "created_at", "internal_notes", "view_count"},
    # Other tables if it applies
}


def _get_inspector_changes(instance: Any) -> dict[str, tuple[Any, Any]]:
    """
    Uses SQLAlchemy's inspector to detect which attributes changed.
    Returns a dict {column_name: (old_value, new_value)} of only the modified ones.
    """
    insp = inspect(instance)
    changes = {}
    for attr in insp.mapper.column_attrs:
        attr_state = insp.attrs[attr.key]
        if attr_state.history.has_changes():
            old_values = attr_state.history.deleted
            new_values = attr_state.history.added
            old = old_values[0] if old_values else None
            new = new_values[0] if new_values else None
            changes[attr.key] = (old, new)
    return changes


@event.listens_for(Session, "after_flush")
def _audit_after_flush(session: Session, flush_context):
    """
    After the flush:
    - session.new: instances just INSERTed (with IDs already assigned).
    - session.dirty: UPDATEd instances.
    - session.deleted: DELETEd instances.

    Note: the modifications we make here (adding TaskAuditLog) get
    flushed in the next cycle. SQLAlchemy handles this well.
    """
    user_id = getattr(session.info, "audit_user_id", None)
    request_id = getattr(session.info, "audit_request_id", None)
    source = getattr(session.info, "audit_source", "api")

    audit_entries = []

    # INSERTs
    for instance in session.new:
        if not isinstance(instance, Task):
            continue
        diff = _build_diff_for_insert(instance)
        if diff:  # only audit if there's something to record
            audit_entries.append(TaskAuditLog(
                entity_id=instance.id,
                action="INSERT",
                changed_by=user_id,
                request_id=request_id,
                source=source,
                diff=diff,
            ))

    # UPDATEs
    for instance in session.dirty:
        if not isinstance(instance, Task):
            continue
        if not session.is_modified(instance, include_collections=False):
            continue
        diff = _build_diff_for_update(instance)
        if diff:  # skip if only excluded columns changed
            audit_entries.append(TaskAuditLog(
                entity_id=instance.id,
                action="UPDATE",
                changed_by=user_id,
                request_id=request_id,
                source=source,
                diff=diff,
            ))

    # DELETEs
    for instance in session.deleted:
        if not isinstance(instance, Task):
            continue
        diff = _build_diff_for_delete(instance)
        if diff:
            audit_entries.append(TaskAuditLog(
                entity_id=instance.id,
                action="DELETE",
                changed_by=user_id,
                request_id=request_id,
                source=source,
                diff=diff,
            ))

    if audit_entries:
        session.add_all(audit_entries)


def _build_diff_for_insert(instance: Task) -> dict:
    """On INSERT: each column as [null, value]."""
    excluded = EXCLUDED_COLUMNS_BY_TABLE.get(instance.__tablename__, set())
    diff = {}
    for col in inspect(instance).mapper.column_attrs:
        if col.key in excluded:
            continue
        new_value = getattr(instance, col.key)
        diff[col.key] = [None, _serialize(new_value)]
    return diff


def _build_diff_for_update(instance: Task) -> dict:
    """On UPDATE: only the modified columns, as [old, new]."""
    excluded = EXCLUDED_COLUMNS_BY_TABLE.get(instance.__tablename__, set())
    changes = _get_inspector_changes(instance)
    diff = {}
    for col_name, (old, new) in changes.items():
        if col_name in excluded:
            continue
        diff[col_name] = [_serialize(old), _serialize(new)]
    return diff


def _build_diff_for_delete(instance: Task) -> dict:
    """On DELETE: each column as [value, null]."""
    excluded = EXCLUDED_COLUMNS_BY_TABLE.get(instance.__tablename__, set())
    diff = {}
    for col in inspect(instance).mapper.column_attrs:
        if col.key in excluded:
            continue
        old_value = getattr(instance, col.key)
        diff[col.key] = [_serialize(old_value), None]
    return diff


def _serialize(value: Any) -> Any:
    """Serializes non-JSON-friendly values (datetimes, etc) into a JSON-able form."""
    if isinstance(value, datetime):
        return value.isoformat()
    return value

Important details:

• session.info: a mutable dict available on every session. We use it to pass the context (user_id, request_id, source) from the dependency to the subsequent modifications in the session.
• inspect(): SQLAlchemy's API for introspecting entities. attr_state.history gives you the values before and after the modification.
• session.is_modified(): checks whether an instance has real changes. Without this, an entity that got "touched" but didn't change would show up in session.dirty without having changed anything.
• audit_entries.append(...) + session.add_all(...): the new TaskAuditLogs get added to the session. The next flush (the transaction's autoflush at commit) will persist them. SQLAlchemy handles this correctly.

The TaskAuditLog model

# app/audit_models.py
from datetime import datetime
from typing import Any
from uuid import UUID

from sqlalchemy import BigInteger, DateTime, Text, func
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID
from sqlalchemy.orm import Mapped, mapped_column

from app.models import Base  # your declarative Base


class TaskAuditLog(Base):
    __tablename__ = "task_log"
    __table_args__ = {"schema": "audit"}

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    entity_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    action: Mapped[str] = mapped_column(Text, nullable=False)
    changed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    changed_by: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    diff: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
    request_id: Mapped[UUID | None] = mapped_column(PG_UUID, nullable=True)
    source: Mapped[str | None] = mapped_column(Text, nullable=True)

The dependency that passes the context

# app/audit_context.py
from typing import Optional
from uuid import UUID, uuid4

from fastapi import Depends, Header
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_session


async def audit_context(
    db: AsyncSession = Depends(get_session),
    x_user_id: Optional[int] = Header(default=None),
    x_request_id: Optional[str] = Header(default=None),
) -> AsyncSession:
    """
    Sets the audit context in session.info, where the listener reads it.
    The difference from capsule 03: we don't use SET LOCAL (there's no SQL trigger),
    we pass the context via session.info.
    """
    db.info["audit_user_id"] = x_user_id
    db.info["audit_request_id"] = UUID(x_request_id) if x_request_id else uuid4()
    db.info["audit_source"] = "api"
    return db

The difference from capsule 03: there we used SET LOCAL audit.user_id so the PL/pgSQL trigger would read it with current_setting(). Here we pass the context via session.info so the Python listener reads it. Equivalent in purpose, different in mechanism.

End-to-end verification

# scripts/verify_listener_audit.py
import asyncio

from app.db import AsyncSessionLocal
from app.models import Task


async def main():
    async with AsyncSessionLocal() as session:
        # Set the context
        session.info["audit_user_id"] = 47
        session.info["audit_source"] = "manual"

        # INSERT
        task = Task(title="Buy milk", priority=1)
        session.add(task)
        await session.commit()
        print(f"INSERT: task #{task.id}")

        # UPDATE
        task.title = "Buy milk and bread"
        await session.commit()
        print(f"UPDATE: title changed")

        # Switch the context to another user
        session.info["audit_user_id"] = 23
        task.status = "closed"
        await session.commit()
        print(f"UPDATE: status changed by user 23")

        # Look at the audit log
        from sqlalchemy import text
        result = await session.execute(text(
            "SELECT id, entity_id, action, changed_by, source, diff "
            "FROM audit.task_log ORDER BY id"
        ))
        for row in result:
            print(row)

python scripts/verify_listener_audit.py
# INSERT: task #1
# UPDATE: title changed
# UPDATE: status changed by user 23
# (1, 1, 'INSERT', 47, 'manual', {'id': [None, 1], 'title': [None, 'Buy milk'], ...})
# (2, 1, 'UPDATE', 47, 'manual', {'title': ['Buy milk', 'Buy milk and bread']})
# (3, 1, 'UPDATE', 23, 'manual', {'status': ['open', 'closed']})

It works. The listener captured the three events, the correct changed_by on each one, and the diff is clean (no excluded columns).

Side-by-side comparison: triggers vs event listeners

Verdict:

• Triggers win when: the DB allows it, there's raw SQL / scripts / migrations that also have to be audited, compliance is strict about completeness.
• Event listeners win when: the DB doesn't allow custom triggers, tests have to run on a different DB from production, the team prefers codebase uniformity.
• Neither wins absolutely. It's a decision by context.

Bypass mitigations

If you choose event listeners, you have to mitigate the bypass risk. Three combined mitigations are enough for most teams:

Mitigation 1: a lint rule against text()/session.execute(text(...)) in application code

Your codebase can have a rule: no file in app/api/ may use text() for mutations. Only the ORM. Reports and complex queries that need raw SQL go in app/admin/ or app/internal/ with a documented justification.

# tests/architecture/test_no_raw_sql_in_api.py
import re
from pathlib import Path


def test_no_text_in_api_layer():
    """Mutations in app/api/ have to go through the ORM, not through text()."""
    forbidden_pattern = re.compile(r"text\(\s*['\"](INSERT|UPDATE|DELETE)", re.IGNORECASE)
    violations = []

    for py_file in Path("app/api/").rglob("*.py"):
        content = py_file.read_text()
        for line_num, line in enumerate(content.splitlines(), 1):
            if forbidden_pattern.search(line):
                violations.append(f"{py_file}:{line_num}: {line.strip()}")

    assert not violations, (
        "Mutations via text() in app/api/. Move them to the ORM or document them:\n"
        + "\n".join(violations)
    )

Mitigation 2: document and track any legitimate bypass

When there's a legitimate bypass case (a maintenance script, a migration with a bulk update), document it:

• In the code: a # AUDIT-BYPASS: specific justification comment.
• In AUDIT-DECISIONS.md: a list of all the known bypasses with their reason.
• In audit.task_log manually: emit an explicit row "operation X bypassed the listener, run by Y, date Z".

Mitigation 3: post-hoc verification

A nightly job that compares the count in audit.task_log with the expected count of operations. If there's a significant discrepancy, alert:

# A job that detects possible bypasses
async def detect_audit_gaps(session):
    """
    Heuristic: compare the count of changes in `tasks` (deduced from
    `updated_at`) with the count of rows in `audit.task_log`.
    If they differ a lot, there were bypasses.
    """
    tasks_modified_today = await session.scalar(text("""
        SELECT COUNT(*) FROM tasks
        WHERE updated_at >= CURRENT_DATE
    """))

    audit_entries_today = await session.scalar(text("""
        SELECT COUNT(*) FROM audit.task_log
        WHERE changed_at >= CURRENT_DATE
    """))

    # Every modification should generate at least one audit entry
    if tasks_modified_today > audit_entries_today * 1.2:  # 20% tolerance
        await alert(f"Audit gap detected: {tasks_modified_today} tasks modified, only {audit_entries_today} audit entries")

Why does this matter in real work?

1. It's the only option when triggers aren't viable. Aurora Serverless v1, certain RDS configurations, restrictive Heroku plans — teams on these stacks have no choice. Knowing how to implement event listeners is a necessary skill, not an optional one.

2. It's the right approach when the team prefers Python uniformity. Faster onboarding, more unified debugging, safer refactoring with type checkers. The loss of the "no bypass" guarantee is a conscious trade-off.

3. It's the basis for deterministic tests. Test suites that run on SQLite (faster than Postgres) without sacrificing the audit logic. The listener works the same; the tests are fast.

4. It teaches you architectural flexibility. There are teams that combine: triggers for critical tables (where the no-bypass guarantee is essential) and event listeners for secondary tables (where Python's flexibility weighs more). Knowing both approaches enables that per-table decision.

Traps and common mistakes

Mistake 1 (conceptual): assuming the listener captures raw SQL

Symptom: a maintenance script runs await session.execute(text("UPDATE tasks SET status = 'archived' WHERE created_at < '2020-01-01'")). It modifies thousands of rows. The audit log records nothing. The team discovers the bug weeks later when a compliance officer asks about those changes.

Why it happens: text() runs SQL directly without going through the ORM system. before_flush and after_flush only fire for operations via the ORM (session.add, session.delete, session.dirty).

How to tell: review every place in the code where text() is used with mutation verbs. Each one is a bypass candidate.

How to fix it:

1. Apply the lint rule (mitigation 1 above).
2. For legitimate cases, write to audit.task_log manually after the raw SQL:

async def archive_old_tasks(session, user_id):
    # 1. Run the bulk UPDATE
    result = await session.execute(text("""
        UPDATE tasks SET status = 'archived'
        WHERE created_at < '2020-01-01' AND status != 'archived'
        RETURNING id
    """))
    archived_ids = [row.id for row in result]

    # 2. Manually write to the audit log
    if archived_ids:
        for task_id in archived_ids:
            audit = TaskAuditLog(
                entity_id=task_id,
                action="UPDATE",
                changed_by=user_id,
                source="cron",
                diff={"status": ["open", "archived"]},  # the diff is known to the script
            )
            session.add(audit)

    await session.commit()

3. For the Alembic migrations case, similar: write to the log explicitly.

Mistake 2 (operational): the listener gets registered twice

Symptom: every change generates two rows in audit.task_log, identical except for the id.

Why it happens: the module where the listener lives gets imported twice (typically in tests with re-imports, or from uvicorn's hot-reload). The @event.listens_for decorator runs twice and the listener ends up registered twice.

How to tell: run a simple change in a test and count the rows in audit.task_log. If there are 2 for each change, this is the bug.

How to fix it: a single-registration guard:

_LISTENER_REGISTERED = False


def install_audit_listener():
    global _LISTENER_REGISTERED
    if _LISTENER_REGISTERED:
        return
    _LISTENER_REGISTERED = True

    @event.listens_for(Session, "after_flush")
    def _audit_after_flush(session, flush_context):
        # ... the logic
        pass

# app/main.py
from app.audit_listener import install_audit_listener

install_audit_listener()
app = FastAPI(...)

Alternatively, check whether it's already registered:

from sqlalchemy.event import contains

if not contains(Session, "after_flush", _audit_after_flush):
    event.listen(Session, "after_flush", _audit_after_flush)

Mistake 3 (conceptual): the listener runs inside the change's transaction

Symptom: the audit_context dependency sets session.info["audit_user_id"] = 47. The listener reads the value and everything works. But a test fails intermittently: sometimes changed_by is 47, sometimes it's NULL.

Why it happens: session.info is per session, not per transaction. If two requests reuse the same session (unlikely in async but possible in custom test setups), the second one's context overwrites the first's. The tests aren't isolating correctly.

How to tell: in the test setup, is a new session created per test, or is it reused? If it's reused, this bug is likely.

How to fix it: make sure each request/test has its own session. In FastAPI with async_sessionmaker, the standard pattern (one session per request via a dependency) already guarantees it. In tests:

@pytest_asyncio.fixture
async def session():
    """Each test gets its own session, completely isolated."""
    async with AsyncSessionLocal() as s:
        yield s
        await s.rollback()  # rollback at the end, doesn't contaminate other tests

Mistake 4 (operational): the INSERT into the audit log inside the listener fails and breaks the main transaction

Symptom: a user UPDATEs a task. The listener tries to insert into audit.task_log but the DB is temporarily loaded and the INSERT fails with a timeout. The whole transaction rolls back. The user sees a 500 error. The task wasn't updated.

Why it happens: the listener writes in the same transaction. If its INSERT fails, the whole transaction goes down. That's typically what we want (atomicity: if it isn't audited, the change doesn't get applied), but it can be undesirable if the audit log is non-critical.

How to tell: review the app's error logs. If there are errors like "task not updated because the audit failed," this is the case.

How to fix it (three options):

1. Accept the atomicity (the recommended default for a compliance audit log): if it isn't audited, the change doesn't get applied. It's what the business wants.

2. Asynchronous audit via the outbox pattern: the listener writes the event to an audit_outbox table (a cheap insert), a separate job processes it and moves it into the real audit.task_log. If the outbox fails, everything fails (correct for compliance). If the async processing fails, it only delays the audit, it doesn't break the transaction.

3. Best-effort audit via try/except: wrap the session.add(audit) in a try/except. Log errors without propagating. It sacrifices the guarantee of complete auditing for availability.

# Option 3 (best-effort, NOT recommended for compliance)
@event.listens_for(Session, "after_flush")
def _audit_after_flush(session, flush_context):
    try:
        # ... generate audit_entries
        session.add_all(audit_entries)
    except Exception as e:
        logger.error("Audit failed: %s", e)
        # Don't propagate — the main transaction continues

Mistake 5 (conceptual): mixing approaches (trigger + listener) with no coordination

Symptom: the team migrates from triggers to event listeners. It forgets to disable the old trigger. Every change generates two rows in audit.task_log: one from the trigger, one from the listener.

Why it happens: the two approaches are independent. If both are active, both audit.

How to tell: run a simple change, count the rows. If there are 2, there's duplication.

How to fix it: pick one and disable the other explicitly:

-- If you migrate to the listener, remove the trigger
DROP TRIGGER IF EXISTS task_log_trigger ON tasks;
DROP FUNCTION IF EXISTS audit.task_log_trigger();

Or the other way around (remove the listener):

# Comment out install_audit_listener() or add a global guard
INSTALL_LISTENER = False  # a toggle to disable it

The lesson: during migrations from one approach to the other, there's a period where both are active. Document the plan: when the new one gets enabled, when the old one gets disabled, how to verify it in production.

Exercises

Exercise 1: implement the listener for a new entity

Your app has a comments table. Implement the event listener that audits its changes into audit.comment_log.

# app/audit_listener.py — extend it
from app.models import Comment
from app.audit_models import CommentAuditLog


EXCLUDED_COLUMNS_BY_TABLE = {
    "tasks": {"updated_at", "created_at", "internal_notes", "view_count"},
    "comments": {"created_at", "edited_count"},
}


# Mapping from table to audit class
AUDIT_LOG_BY_TABLE = {
    "tasks": TaskAuditLog,
    "comments": CommentAuditLog,
}


@event.listens_for(Session, "after_flush")
def _audit_after_flush(session: Session, flush_context):
    user_id = session.info.get("audit_user_id")
    request_id = session.info.get("audit_request_id")
    source = session.info.get("audit_source", "api")

    audit_entries = []

    for instance in session.new:
        log_class = AUDIT_LOG_BY_TABLE.get(instance.__tablename__)
        if log_class is None:
            continue
        diff = _build_diff_for_insert(instance)
        if diff:
            audit_entries.append(log_class(
                entity_id=instance.id, action="INSERT",
                changed_by=user_id, request_id=request_id, source=source, diff=diff,
            ))

    for instance in session.dirty:
        log_class = AUDIT_LOG_BY_TABLE.get(instance.__tablename__)
        if log_class is None:
            continue
        if not session.is_modified(instance, include_collections=False):
            continue
        diff = _build_diff_for_update(instance)
        if diff:
            audit_entries.append(log_class(
                entity_id=instance.id, action="UPDATE",
                changed_by=user_id, request_id=request_id, source=source, diff=diff,
            ))

    for instance in session.deleted:
        log_class = AUDIT_LOG_BY_TABLE.get(instance.__tablename__)
        if log_class is None:
            continue
        diff = _build_diff_for_delete(instance)
        if diff:
            audit_entries.append(log_class(
                entity_id=instance.id, action="DELETE",
                changed_by=user_id, request_id=request_id, source=source, diff=diff,
            ))

    if audit_entries:
        session.add_all(audit_entries)

# app/audit_models.py — add the model
class CommentAuditLog(Base):
    __tablename__ = "comment_log"
    __table_args__ = {"schema": "audit"}

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    entity_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    action: Mapped[str] = mapped_column(Text, nullable=False)
    changed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    changed_by: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    diff: Mapped[dict] = mapped_column(JSONB, nullable=False)
    request_id: Mapped[UUID | None] = mapped_column(PG_UUID, nullable=True)
    source: Mapped[str | None] = mapped_column(Text, nullable=True)

Exercise 2: a test that verifies isolation between sessions

Write an async pytest test that verifies:

a) Changes in a session with audit_user_id = 47 record changed_by = 47. b) Changes in another session with audit_user_id = 23 record changed_by = 23. c) The two sessions don't contaminate each other.

# tests/test_audit_listener_isolation.py
import pytest
from sqlalchemy import text

from app.db import AsyncSessionLocal
from app.models import Task

pytestmark = pytest.mark.asyncio


async def test_session_aislamiento_de_user_id():
    """Two sessions with different users don't contaminate each other."""
    async with AsyncSessionLocal() as s1:
        s1.info["audit_user_id"] = 47
        s1.info["audit_source"] = "test"

        task1 = Task(title="Task of user 47")
        s1.add(task1)
        await s1.commit()
        task1_id = task1.id

    async with AsyncSessionLocal() as s2:
        s2.info["audit_user_id"] = 23
        s2.info["audit_source"] = "test"

        task2 = Task(title="Task of user 23")
        s2.add(task2)
        await s2.commit()
        task2_id = task2.id

    # Verify each audit recorded the right user
    async with AsyncSessionLocal() as verify:
        result = await verify.execute(text("""
            SELECT entity_id, changed_by FROM audit.task_log
            WHERE entity_id IN (:t1, :t2)
            ORDER BY entity_id
        """), {"t1": task1_id, "t2": task2_id})
        rows = list(result)

    assert len(rows) == 2
    assert rows[0].entity_id == task1_id
    assert rows[0].changed_by == 47
    assert rows[1].entity_id == task2_id
    assert rows[1].changed_by == 23


async def test_concurrent_sessions_no_se_contaminan():
    """Two sessions running concurrently don't share context."""
    import asyncio

    async def session_for_user(user_id, title_suffix):
        async with AsyncSessionLocal() as s:
            s.info["audit_user_id"] = user_id
            s.info["audit_source"] = "test"
            task = Task(title=f"Concurrent task {title_suffix}")
            s.add(task)
            await s.commit()
            return task.id

    # Run 5 sessions in parallel, each with a different user
    user_ids = [10, 20, 30, 40, 50]
    task_ids = await asyncio.gather(*[
        session_for_user(uid, str(uid)) for uid in user_ids
    ])

    # Verify each audit has the right user
    async with AsyncSessionLocal() as verify:
        for task_id, expected_user in zip(task_ids, user_ids):
            result = await verify.execute(text("""
                SELECT changed_by FROM audit.task_log
                WHERE entity_id = :tid AND action = 'INSERT'
            """), {"tid": task_id})
            assert result.scalar() == expected_user

Exercise 3: detect a bypass post-hoc

Write a script that scans the app's logs and detects operations that probably bypassed the listener (based on the discrepancy between updated_at and the audit log).

# scripts/detect_audit_bypass.py
import asyncio
from datetime import datetime, timedelta, timezone

from sqlalchemy import text

from app.db import AsyncSessionLocal


async def main():
    async with AsyncSessionLocal() as session:
        # For each task modified in the last 24h, verify it has
        # at least one entry in audit.task_log matching its updated_at
        result = await session.execute(text("""
            WITH modified_tasks AS (
                SELECT id, updated_at, status
                FROM tasks
                WHERE updated_at >= NOW() - INTERVAL '24 hours'
            ),
            task_audits AS (
                SELECT entity_id, MAX(changed_at) AS last_audited
                FROM audit.task_log
                WHERE changed_at >= NOW() - INTERVAL '24 hours'
                GROUP BY entity_id
            )
            SELECT
                mt.id,
                mt.updated_at,
                ta.last_audited,
                EXTRACT(EPOCH FROM (mt.updated_at - COALESCE(ta.last_audited, mt.updated_at))) AS gap_seconds
            FROM modified_tasks mt
            LEFT JOIN task_audits ta ON mt.id = ta.entity_id
            WHERE
                ta.last_audited IS NULL
                OR mt.updated_at - ta.last_audited > INTERVAL '5 seconds'
            ORDER BY mt.updated_at DESC
            LIMIT 100
        """))

        suspicious = list(result)

        if not suspicious:
            print("No bypasses detected. All modifications have corresponding audit entries.")
            return

        print(f"WARNING: {len(suspicious)} potentially bypassed modifications detected:")
        for row in suspicious:
            print(f"  task #{row.id}: updated_at={row.updated_at}, last_audited={row.last_audited}, gap={row.gap_seconds:.1f}s")


if __name__ == "__main__":
    asyncio.run(main())

Exercise 4: compare the performance of triggers vs listeners

Write a benchmark that compares the time of:

a) Inserting 1000 tasks with PostgreSQL triggers. b) Inserting 1000 tasks with SQLAlchemy event listeners.

What result do you expect and why?

# benchmarks/audit_overhead.py
import asyncio
import time
from sqlalchemy import text

from app.db import AsyncSessionLocal
from app.models import Task


N = 1000


async def benchmark_with_trigger():
    """Inserts N tasks. The PostgreSQL trigger audits each one automatically."""
    async with AsyncSessionLocal() as session:
        await session.execute(text("DELETE FROM audit.task_log"))
        await session.execute(text("DELETE FROM tasks"))
        await session.commit()

        start = time.time()
        for i in range(N):
            task = Task(title=f"Task {i}", priority=0)
            session.add(task)
        await session.commit()
        elapsed = time.time() - start
        return elapsed


async def benchmark_with_listener():
    """Inserts N tasks. The SQLAlchemy listener audits each one."""
    async with AsyncSessionLocal() as session:
        await session.execute(text("DELETE FROM audit.task_log"))
        await session.execute(text("DELETE FROM tasks"))
        await session.commit()

        session.info["audit_user_id"] = 99
        session.info["audit_source"] = "benchmark"

        start = time.time()
        for i in range(N):
            task = Task(title=f"Task {i}", priority=0)
            session.add(task)
        await session.commit()
        elapsed = time.time() - start
        return elapsed


async def benchmark_no_audit():
    """Baseline: insert N tasks with no audit (trigger disabled, listener not registered)."""
    async with AsyncSessionLocal() as session:
        await session.execute(text("ALTER TABLE tasks DISABLE TRIGGER ALL"))
        await session.execute(text("DELETE FROM audit.task_log"))
        await session.execute(text("DELETE FROM tasks"))
        await session.commit()

        start = time.time()
        for i in range(N):
            task = Task(title=f"Task {i}", priority=0)
            session.add(task)
        await session.commit()
        elapsed = time.time() - start

        await session.execute(text("ALTER TABLE tasks ENABLE TRIGGER ALL"))
        await session.commit()
        return elapsed


async def main():
    baseline = await benchmark_no_audit()
    print(f"Baseline (no audit):       {baseline*1000:.0f}ms")

    trigger_time = await benchmark_with_trigger()
    overhead_trigger = (trigger_time - baseline) / baseline * 100
    print(f"With trigger:              {trigger_time*1000:.0f}ms ({overhead_trigger:+.1f}% overhead)")

    listener_time = await benchmark_with_listener()
    overhead_listener = (listener_time - baseline) / baseline * 100
    print(f"With listener:             {listener_time*1000:.0f}ms ({overhead_listener:+.1f}% overhead)")


if __name__ == "__main__":
    asyncio.run(main())

Baseline (no audit):       180ms
With trigger:              210ms (+16.7% overhead)
With listener:             340ms (+88.9% overhead)

Exercise 5: argue trigger vs listener for a specific case

You're given these contexts. Decide trigger or listener and justify it with 3 reasons.

a) A task app (TaskFlow) on self-hosted PostgreSQL, a team of 5 Python devs.

b) A billing app on Aurora Serverless v1, a team of 8 Python devs.

c) An IoT app that receives 50k inserts/second, managed PostgreSQL.

d) An e-commerce app with tests on SQLite, production on PostgreSQL, a team of 12 devs.

Summary and next step

In this capsule you learned:

• SQLAlchemy event listeners are the Python-side alternative to PostgreSQL triggers. The same purpose, a different trade-off.
• after_flush is the key event: the new entities have IDs, the changes have been processed, and there's time to insert into the audit log before the commit.
• session.info is the mechanism for passing context (user_id, request_id, source) from the dependency to the listener. Equivalent to SET LOCAL in the trigger approach.
• SQLAlchemy's inspector (inspect() + attr.history) lets you generate the diff for UPDATEs by comparing old vs new values.
• Bypassing is trivial: text(), op.execute(), non-ORM scripts bypass the listener. Mitigate with lint rules, documentation, and post-hoc verification.
• The trigger vs listener decision: the trigger wins if the DB allows it + strict compliance + no tests on another DB. The listener wins if the DB doesn't allow triggers + tests on SQLite + a Python-only team + a mitigable bypass.
• Typical overhead: trigger ~5-15%, listener ~15-25%. Acceptable for typical apps, problematic for bulk operations.
• For bulk operations: temporarily disabling the trigger or emitting a batch audit at the end are alternatives to accepting the overhead.
• Mixing approaches (a trigger + a listener at the same time) generates silent duplication. Pick one and disable the other explicitly.

Before moving on you should be able to:

• Implement the complete listener (model + listener + dependency) on a new table in under 30 minutes.
• Decide between a trigger and a listener for a specific case with 3 concrete arguments.
• Detect bypasses post-hoc with SQL queries.
• Mitigate the bypass risk with lint rules + documentation + alerts.
• Write tests that verify isolation between concurrent sessions.

Next capsule — Retention and partitioning of audit logs. You already have the approach (a trigger or a listener) and the format (an audit log, a history table, or lightweight event sourcing). Now comes the inevitable operational problem: the log grows without stopping. A table with 10k changes a day generates 3.6M rows/year; in 5 years, 18M. With no retention policy, the audit log eventually weighs more than the audited table and the INSERTs start to feel it. You're going to learn to partition audit.task_log by month with PostgreSQL 16+'s declarative partitioning, to design retention policies that satisfy compliance, and to archive old partitions to S3 (parquet). It's the capsule that prepares you to take an audit log to real production.

Resources

1. SQLAlchemy 2.0 — ORM Events — the official reference. The section on SessionEvents.before_flush and after_flush.
2. SQLAlchemy 2.0 — inspect() API — for understanding how to introspect entities and get the history of changes.
3. SQLAlchemy 2.0 — Session.info and lifespan — the attribute we use to pass context from the dependency to the listener.
4. SQLAlchemy 2.0 — Versioning Examples — official examples of patterns similar to this capsule's. Useful as an additional reference.
5. Aurora Serverless v1 — Limitations — the reference for the restrictions that motivate choosing a listener over a trigger.
6. PostgreSQL Documentation — Server Programming Limitations — for understanding what PL/pgSQL functions can and can't do.
7. Real Python — SQLAlchemy 2.0 ORM Events — a complementary practical tutorial.
8. Vlad Mihalcea — "How to log changes with Hibernate Envers" — the conceptual equivalent in Java/Hibernate. Useful for seeing how other ORMs solve the same problem.

Module 3 — SQL Patterns for Production APIs Guide

Next capsule: Retention and partitioning of audit logs — how to keep the log from growing forever.