PostgreSQL triggers for auditing

Capsule overview

You already made the decisions about what to audit and designed the base schema of audit.task_log (capsule 02). Now comes the implementation: the PL/pgSQL trigger that fires on every INSERT/UPDATE/DELETE of tasks and records the change in audit.task_log automatically, without the app having to remember anything. It's the default approach this module recommends, and for good reasons: it lives in the database, it can't be bypassed from the app, it captures every change — including the ones made by raw SQL, maintenance jobs, or migrations.

But there's one detail that separates a useful trigger from a useless one: PostgreSQL doesn't know who your app's user is. The trigger knows the DB connection (postgres or whatever role your pool uses), it doesn't know the HTTP request came from user_id = 47. Without passing it that context, your audit log ends up with changed_by = NULL on every row — it records "somebody changed something," information with no value for real auditing. The solution is standard: the app runs SET LOCAL audit.user_id = '47' at the start of the transaction, and the trigger reads it with current_setting(). It's the detail this capsule teaches you to implement properly.

By the end you'll have: the complete, runnable PL/pgSQL trigger, a FastAPI dependency that sets the context on every request, code that captures OLD and NEW and generates the diff as JSONB, filters that exclude sensitive and noisy columns, and end-to-end verification with real queries. It's the module's central technical capsule. The patterns you learn here are exactly the ones the module project (capsule 08) and the capstone project of module 8 are going to use.

Mental model: the trigger as a universal interceptor

Think of the trigger as HTTP middleware, but at the database level. Middleware intercepts every HTTP request before it reaches the handler and can add logging, authentication, or headers. The trigger intercepts every INSERT/UPDATE/DELETE before (or after) it runs, and can execute additional logic — in our case, writing to audit.task_log.

The key difference from HTTP middleware: the trigger is unavoidable. There's no way to bypass it from the app. If someone on your team writes a maintenance script that runs UPDATE tasks SET status = 'archived' WHERE created_at < '2020-01-01', the trigger fires for every row and records the change. If someone runs a migration that modifies tasks directly, the trigger fires. If someone uses psql and edits a row by hand, the trigger fires. It's the only approach that guarantees completeness.

That guarantee has a cost: the trigger lives in SQL, not in your main codebase. It's code your team has to learn to read and maintain. Capsule 06 covers the alternative (SQLAlchemy event listeners) for cases where triggers aren't viable. But when they are viable — and for a monolithic API with self-hosted or standard managed PostgreSQL they are —, they're the right answer.

The problem of the unknown user_id

Let's start with the naive version of the trigger. It covers the basic mechanics but has the bug we're going to fix.

-- Naive version: it records the change but does NOT know who made it
CREATE OR REPLACE FUNCTION audit.task_log_trigger()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO audit.task_log (entity_id, action, diff)
    VALUES (
        COALESCE(NEW.id, OLD.id),
        TG_OP,
        CASE TG_OP
            WHEN 'INSERT' THEN to_jsonb(NEW)
            WHEN 'DELETE' THEN to_jsonb(OLD)
            WHEN 'UPDATE' THEN to_jsonb(NEW) - to_jsonb(OLD)
        END
    );
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER task_log_trigger
AFTER INSERT OR UPDATE OR DELETE ON tasks
FOR EACH ROW
EXECUTE FUNCTION audit.task_log_trigger();

It works: every INSERT/UPDATE/DELETE on tasks generates a row in audit.task_log. But there are three critical problems.

Problem 1: changed_by will always be NULL. The INSERT doesn't set changed_by. PostgreSQL doesn't know which app user is behind the connection.

Problem 2: the UPDATE's diff is wrong. to_jsonb(NEW) - to_jsonb(OLD) subtracts keys from the left-hand side. If a column didn't change, it stays in the result (because it doesn't get removed). The "diff" ends up being almost the whole row.

Problem 3: no filter for sensitive columns. If tasks had a secret_token field, it would record it in the log. A security risk.

Let's solve all three, one by one.

Solving problem #1: passing the request's context to the trigger

We need to pass the request's user_id to the trigger. The standard solution in PostgreSQL is custom GUCs (Grand Unified Configuration), specifically with SET LOCAL inside a transaction.

The full flow

1. The FastAPI app receives a request, extracts the user_id from the JWT (or equivalent).
2. A FastAPI dependency opens a transaction in SQLAlchemy and runs SET LOCAL audit.user_id = '47'.
3. The FastAPI handler runs its logic (INSERT/UPDATE/DELETE of tasks).
4. The trigger fires, reads current_setting('audit.user_id'), and stores it in audit.task_log.changed_by.
5. The transaction commits. The SET LOCAL disappears (it's local to the transaction, hence the name).

SET LOCAL is the key piece. Unlike SET (which persists in the session), SET LOCAL only lasts until the end of the transaction. That's exactly what we want: each HTTP request is a transaction, and the context has to be cleared when it ends.

The trigger reading the context

CREATE OR REPLACE FUNCTION audit.task_log_trigger()
RETURNS TRIGGER AS $$
DECLARE
    v_user_id BIGINT;
    v_request_id UUID;
    v_source TEXT;
BEGIN
    -- current_setting with missing_ok=true avoids an error if it isn't set
    v_user_id := NULLIF(current_setting('audit.user_id', true), '')::BIGINT;
    v_request_id := NULLIF(current_setting('audit.request_id', true), '')::UUID;
    v_source := NULLIF(current_setting('audit.source', true), '');

    INSERT INTO audit.task_log (
        entity_id, action, changed_by, request_id, source, diff
    )
    VALUES (
        COALESCE(NEW.id, OLD.id),
        TG_OP,
        v_user_id,
        v_request_id,
        v_source,
        -- (we solve the diff in the next section)
        '{}'::jsonb
    );
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

Important details:

• current_setting('audit.user_id', true): the second argument true means "missing_ok". Without it, if the setting isn't defined, PostgreSQL raises an error. With true, it returns an empty string.
• NULLIF(..., ''): converts an empty string to NULL. This matters because SET LOCAL audit.user_id = '' isn't the same as "not set," and the cast to BIGINT would fail on an empty string.
• The audit.* namespace: PostgreSQL requires custom GUCs to have a dot (namespace.key). By convention we use audit.user_id, audit.request_id, audit.source.

The FastAPI dependency

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

from fastapi import Depends, Request
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_session
from app.auth import get_current_user_id  # assumes your auth system


async def audit_context(
    request: Request,
    db: AsyncSession = Depends(get_session),
    user_id: Optional[int] = Depends(get_current_user_id),
) -> AsyncSession:
    """
    Sets the audit context on the current transaction.

    Each HTTP request is a transaction. SET LOCAL makes the setting
    only last until the COMMIT/ROLLBACK, which is exactly what we want.

    Usage in an endpoint:
        @app.put("/tasks/{id}")
        async def update_task(
            id: int,
            payload: TaskUpdate,
            db: AsyncSession = Depends(audit_context),  # <-- here
        ):
            ...
    """
    request_id = request.headers.get("x-request-id") or str(request.state.__dict__.get("request_id", ""))

    if user_id is not None:
        await db.execute(text("SET LOCAL audit.user_id = :uid"), {"uid": str(user_id)})

    if request_id:
        await db.execute(text("SET LOCAL audit.request_id = :rid"), {"rid": request_id})

    await db.execute(text("SET LOCAL audit.source = 'api'"))

    return db

Why text("SET LOCAL audit.user_id = :uid") with a bind parameter:

PostgreSQL doesn't allow parameterizing the setting's name in SET LOCAL, but it does allow parameterizing its value. The bind parameter prevents SQL injection and also converts the type correctly.

A subtlety of SQLAlchemy 2.0 async: FastAPI dependencies run inside the same session as the handler. The SET LOCAL sets the context on the session's active connection, and it persists until the session commits. That means any query in the handler will see the context set, and the trigger will read it.

Verification: the context reaches the trigger

# scripts/verify_audit_context.py
import asyncio
from sqlalchemy import select, text

from app.db import AsyncSessionLocal


async def main():
    async with AsyncSessionLocal() as session:
        # Simulate the SET LOCAL the dependency does
        await session.execute(text("SET LOCAL audit.user_id = '47'"))

        # Verify the trigger would see it
        result = await session.execute(
            text("SELECT current_setting('audit.user_id', true) AS ctx_user_id")
        )
        row = result.first()
        print(f"Context set: audit.user_id = {row.ctx_user_id}")
        # Expected: audit.user_id = 47

        await session.commit()


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

python scripts/verify_audit_context.py
# Context set: audit.user_id = 47

Done. The first problem is solved: the trigger can now read the request's real user_id.

Solving problem #2: a correct diff

The - operator between two JSONBs in PostgreSQL doesn't do what we need. to_jsonb(NEW) - to_jsonb(OLD) doesn't return "the keys that changed" — it returns "NEW's keys minus OLD's keys," which is always an empty set if they have the same columns.

What we want is to generate a JSONB shaped like {"column": [old_value, new_value]} only for columns that changed. Let's write a helper function.

CREATE OR REPLACE FUNCTION audit.diff_jsonb(
    p_old JSONB,
    p_new JSONB,
    p_excluded_keys TEXT[] DEFAULT ARRAY[]::TEXT[]
)
RETURNS JSONB AS $$
DECLARE
    v_diff JSONB := '{}'::JSONB;
    v_key TEXT;
    v_old_val JSONB;
    v_new_val JSONB;
BEGIN
    -- Walk all of NEW's keys (OLD's keys that aren't in NEW are deletions,
    -- but in a table with a stable schema the keys are the same)
    FOR v_key IN SELECT jsonb_object_keys(p_new)
    LOOP
        -- Skip explicitly excluded columns (passwords, noise)
        IF v_key = ANY(p_excluded_keys) THEN
            CONTINUE;
        END IF;

        v_old_val := p_old -> v_key;
        v_new_val := p_new -> v_key;

        -- IS DISTINCT FROM handles NULL correctly: NULL != 'x' is TRUE
        IF v_old_val IS DISTINCT FROM v_new_val THEN
            v_diff := v_diff || jsonb_build_object(
                v_key,
                jsonb_build_array(v_old_val, v_new_val)
            );
        END IF;
    END LOOP;

    RETURN v_diff;
END;
$$ LANGUAGE plpgsql IMMUTABLE;

Why this function is the way it is:

• It walks NEW's keys, not OLD's. On an INSERT, OLD is NULL and only NEW has data. On an UPDATE, both have the same keys. On a DELETE we'll handle OLD separately.
• IS DISTINCT FROM instead of !=: it handles NULL correctly. NULL != 'x' returns NULL (not TRUE), which would break the comparison. NULL IS DISTINCT FROM 'x' returns TRUE.
• It excludes keys from the p_excluded_keys array: so you can filter out password_hash, view_count, and the like in a single pass.
• IMMUTABLE: the function has no side effects and the result only depends on the inputs. That lets PostgreSQL cache it.

The trigger using diff_jsonb

CREATE OR REPLACE FUNCTION audit.task_log_trigger()
RETURNS TRIGGER AS $$
DECLARE
    v_user_id BIGINT;
    v_request_id UUID;
    v_source TEXT;
    v_diff JSONB;
    -- Columns we NEVER want to audit (noise or sensitive)
    v_excluded_keys TEXT[] := ARRAY['updated_at', 'created_at'];
BEGIN
    v_user_id := NULLIF(current_setting('audit.user_id', true), '')::BIGINT;
    v_request_id := NULLIF(current_setting('audit.request_id', true), '')::UUID;
    v_source := NULLIF(current_setting('audit.source', true), '');

    -- Generate the diff according to the operation type
    CASE TG_OP
        WHEN 'INSERT' THEN
            v_diff := audit.diff_jsonb('{}'::jsonb, to_jsonb(NEW), v_excluded_keys);
        WHEN 'DELETE' THEN
            v_diff := audit.diff_jsonb(to_jsonb(OLD), '{}'::jsonb, v_excluded_keys);
        WHEN 'UPDATE' THEN
            v_diff := audit.diff_jsonb(to_jsonb(OLD), to_jsonb(NEW), v_excluded_keys);
    END CASE;

    -- If the UPDATE didn't change any meaningful column, don't audit
    IF TG_OP = 'UPDATE' AND v_diff = '{}'::jsonb THEN
        RETURN NULL;
    END IF;

    INSERT INTO audit.task_log (
        entity_id, action, changed_by, request_id, source, diff
    )
    VALUES (
        COALESCE(NEW.id, OLD.id),
        TG_OP,
        v_user_id,
        v_request_id,
        v_source,
        v_diff
    );

    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

New decisions:

• A CASE on TG_OP: PostgreSQL defines the special variable TG_OP with the type of operation that fired the trigger ('INSERT', 'UPDATE', 'DELETE').
• INSERT generates a diff against {}: every column shows up as [null, new_value]. Useful for reconstructing the original state at INSERT.
• DELETE generates a diff of OLD against {}: every column shows up as [value_that_existed, null]. Useful for reconstructing what was deleted.
• IF v_diff = '{}'::jsonb THEN RETURN NULL: if only excluded columns changed (updated_at, etc), the diff comes out empty and we don't audit. It avoids spam of "UPDATE with no meaningful changes."
• v_excluded_keys: the list lives in the trigger. To add exclusions, you modify the trigger. Another option is reading them from a configuration table, but for an audit log of one specific table, hard-coding is simpler and faster.

Solving problem #3: filtering out sensitive columns

We already have the mechanism (the p_excluded_keys array). We just have to use it correctly. For tasks there aren't extremely sensitive columns, but let's say for example we had an internal_notes column with private notes we do NOT want audited.

-- Modify the trigger to include the sensitive column in the exclusions
CREATE OR REPLACE FUNCTION audit.task_log_trigger()
RETURNS TRIGGER AS $$
DECLARE
    -- ...
    v_excluded_keys TEXT[] := ARRAY[
        'updated_at',        -- redundant with changed_at
        'created_at',        -- redundant in UPDATEs (it doesn't change)
        'internal_notes',    -- sensitive: must not go into the log
        'view_count'         -- noise: a volatile counter
    ];
BEGIN
    -- ... the rest is the same
END;
$$ LANGUAGE plpgsql;

For a users table with password_hash, the list would include 'password_hash', 'api_token', 'refresh_token', etc. The operational rule: any sensitive column goes in the trigger's list AND is documented in AUDIT-DECISIONS.md (from capsule 02).

Complete worked example

Let's set everything up from scratch: the schema, the tasks table, the audit log, the trigger, the FastAPI dependency, and end-to-end verification. You'll be able to copy this code into an empty project and run it.

Project setup

mkdir audit-trigger-demo && cd audit-trigger-demo
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn sqlalchemy[asyncio]==2.0.* asyncpg alembic pydantic-settings

# Bring up PostgreSQL 16+ with Docker
docker run -d --name audit-pg \
    -e POSTGRES_PASSWORD=postgres \
    -e POSTGRES_DB=audit_demo \
    -p 5432:5432 \
    postgres:16

Complete schema

-- migrations/001_initial.sql

-- A separate schema for audit
CREATE SCHEMA IF NOT EXISTS audit;

-- The domain table
CREATE TABLE tasks (
    id BIGSERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    description TEXT NULL,
    status TEXT NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'in_progress', 'closed', 'archived')),
    priority INTEGER NOT NULL DEFAULT 0,
    assignee_id BIGINT NULL,
    internal_notes TEXT NULL,  -- an example of a sensitive column (not audited)
    view_count BIGINT NOT NULL DEFAULT 0,  -- an example of a counter (not audited)
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- The audit log table
CREATE TABLE audit.task_log (
    id BIGSERIAL PRIMARY KEY,
    entity_id BIGINT NOT NULL,
    action TEXT NOT NULL CHECK (action IN ('INSERT', 'UPDATE', 'DELETE')),
    changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    changed_by BIGINT NULL,
    diff JSONB NOT NULL,
    request_id UUID NULL,
    source TEXT NULL CHECK (source IN ('api', 'cron', 'migration', 'manual') OR source IS NULL)
);

CREATE INDEX idx_audit_task_log_entity_time
    ON audit.task_log (entity_id, changed_at DESC);

CREATE INDEX idx_audit_task_log_changed_by_time
    ON audit.task_log (changed_by, changed_at DESC)
    WHERE changed_by IS NOT NULL;

CREATE INDEX idx_audit_task_log_diff_gin
    ON audit.task_log USING GIN (diff jsonb_path_ops);

-- The diff helper function
CREATE OR REPLACE FUNCTION audit.diff_jsonb(
    p_old JSONB,
    p_new JSONB,
    p_excluded_keys TEXT[] DEFAULT ARRAY[]::TEXT[]
)
RETURNS JSONB AS $$
DECLARE
    v_diff JSONB := '{}'::JSONB;
    v_key TEXT;
    v_old_val JSONB;
    v_new_val JSONB;
BEGIN
    -- Combine OLD's and NEW's keys (to cover the DELETE case where NEW is {})
    FOR v_key IN
        SELECT jsonb_object_keys(p_old)
        UNION
        SELECT jsonb_object_keys(p_new)
    LOOP
        IF v_key = ANY(p_excluded_keys) THEN
            CONTINUE;
        END IF;

        v_old_val := p_old -> v_key;
        v_new_val := p_new -> v_key;

        IF v_old_val IS DISTINCT FROM v_new_val THEN
            v_diff := v_diff || jsonb_build_object(
                v_key,
                jsonb_build_array(v_old_val, v_new_val)
            );
        END IF;
    END LOOP;

    RETURN v_diff;
END;
$$ LANGUAGE plpgsql IMMUTABLE;

-- The trigger function
CREATE OR REPLACE FUNCTION audit.task_log_trigger()
RETURNS TRIGGER AS $$
DECLARE
    v_user_id BIGINT;
    v_request_id UUID;
    v_source TEXT;
    v_diff JSONB;
    v_excluded_keys TEXT[] := ARRAY[
        'updated_at',
        'created_at',
        'internal_notes',
        'view_count'
    ];
BEGIN
    v_user_id := NULLIF(current_setting('audit.user_id', true), '')::BIGINT;
    v_request_id := NULLIF(current_setting('audit.request_id', true), '')::UUID;
    v_source := NULLIF(current_setting('audit.source', true), '');

    CASE TG_OP
        WHEN 'INSERT' THEN
            v_diff := audit.diff_jsonb('{}'::jsonb, to_jsonb(NEW), v_excluded_keys);
        WHEN 'DELETE' THEN
            v_diff := audit.diff_jsonb(to_jsonb(OLD), '{}'::jsonb, v_excluded_keys);
        WHEN 'UPDATE' THEN
            v_diff := audit.diff_jsonb(to_jsonb(OLD), to_jsonb(NEW), v_excluded_keys);
    END CASE;

    IF TG_OP = 'UPDATE' AND v_diff = '{}'::jsonb THEN
        RETURN NULL;
    END IF;

    INSERT INTO audit.task_log (
        entity_id, action, changed_by, request_id, source, diff
    )
    VALUES (
        COALESCE(NEW.id, OLD.id),
        TG_OP,
        v_user_id,
        v_request_id,
        v_source,
        v_diff
    );

    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- The trigger attached to the table
CREATE TRIGGER task_log_trigger
AFTER INSERT OR UPDATE OR DELETE ON tasks
FOR EACH ROW
EXECUTE FUNCTION audit.task_log_trigger();

# Apply the schema
psql postgresql://postgres:postgres@localhost/audit_demo -f migrations/001_initial.sql

The app's code

# app/db.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/audit_demo"

engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_session():
    async with AsyncSessionLocal() as session:
        yield session

# app/models.py
from datetime import datetime

from sqlalchemy import BigInteger, CheckConstraint, DateTime, Integer, String, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(Text, nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    status: Mapped[str] = mapped_column(String(20), nullable=False, default="open")
    priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    assignee_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    internal_notes: Mapped[str | None] = mapped_column(Text, nullable=True)
    view_count: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )

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

from fastapi import Depends, Header
from sqlalchemy import text
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 on the transaction.

    In production, x_user_id comes from the decoded JWT, not from a header
    trusting the client. Here we simplify for the demo.
    """
    if x_user_id is not None:
        await db.execute(text("SET LOCAL audit.user_id = :uid"), {"uid": str(x_user_id)})

    request_id = x_request_id or str(uuid4())
    await db.execute(text("SET LOCAL audit.request_id = :rid"), {"rid": request_id})
    await db.execute(text("SET LOCAL audit.source = 'api'"))

    return db

# app/main.py
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit_context import audit_context
from app.models import Task

app = FastAPI(title="Audit Trigger Demo")


class TaskCreate(BaseModel):
    title: str
    description: Optional[str] = None
    priority: int = 0


class TaskUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    priority: Optional[int] = None


@app.post("/tasks", status_code=201)
async def create_task(payload: TaskCreate, db: AsyncSession = Depends(audit_context)):
    task = Task(**payload.model_dump())
    db.add(task)
    await db.commit()
    await db.refresh(task)
    return {"id": task.id, "title": task.title, "status": task.status}


@app.put("/tasks/{task_id}")
async def update_task(
    task_id: int,
    payload: TaskUpdate,
    db: AsyncSession = Depends(audit_context),
):
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(404, "Task not found")

    for key, value in payload.model_dump(exclude_unset=True).items():
        setattr(task, key, value)

    await db.commit()
    await db.refresh(task)
    return {"id": task.id, "title": task.title, "status": task.status}


@app.delete("/tasks/{task_id}", status_code=204)
async def delete_task(task_id: int, db: AsyncSession = Depends(audit_context)):
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(404, "Task not found")
    await db.delete(task)
    await db.commit()

End-to-end verification

uvicorn app.main:app --reload &

# 1. Create a task as user 47
curl -X POST http://localhost:8000/tasks \
    -H "Content-Type: application/json" \
    -H "X-User-Id: 47" \
    -d '{"title": "Buy milk", "priority": 1}'
# {"id":1,"title":"Buy milk","status":"open"}

# 2. Modify the title
curl -X PUT http://localhost:8000/tasks/1 \
    -H "Content-Type: application/json" \
    -H "X-User-Id: 47" \
    -d '{"title": "Buy milk and bread"}'

# 3. Change the status as user 23
curl -X PUT http://localhost:8000/tasks/1 \
    -H "Content-Type: application/json" \
    -H "X-User-Id: 23" \
    -d '{"status": "closed"}'

# 4. Look at the audit log
psql postgresql://postgres:postgres@localhost/audit_demo \
    -c "SELECT id, entity_id, action, changed_by, source, diff FROM audit.task_log ORDER BY id;"

 id | entity_id | action | changed_by | source | diff
----+-----------+--------+------------+--------+----------------------------------------------------
  1 |         1 | INSERT |         47 | api    | {"id": [null, 1], "title": [null, "Buy milk"], "status": [null, "open"], "priority": [null, 1], "description": [null, null], "assignee_id": [null, null]}
  2 |         1 | UPDATE |         47 | api    | {"title": ["Buy milk", "Buy milk and bread"]}
  3 |         1 | UPDATE |         23 | api    | {"status": ["open", "closed"]}

Three rows, three correct changed_bys, three clean diffs. The internal_notes and view_count columns don't show up because they're excluded. Neither do created_at/updated_at. This is what the module promised: an audit log that's useful.

Traps and common mistakes

Mistake 1 (conceptual): using SET instead of SET LOCAL

Symptom: after user 47's first request, every following request on the same connection audits as user 47, even user 23's. The audit log shows the wrong user_id.

Why it happens: SET audit.user_id = '47' (without LOCAL) persists as long as the connection is alive. In a SQLAlchemy connection pool, connections get reused between requests. The next request inherits the previous one's setting. SET LOCAL only lasts until the COMMIT/ROLLBACK, which matches the request's lifecycle.

How to tell: look for anywhere in the code (a dependency, middleware, init) where SET is used without LOCAL. It also shows up as "the first request audits correctly, the rest don't."

How to fix it: always SET LOCAL in the audit context. A test that catches the bug:

async def test_audit_user_id_no_se_filtra_entre_requests(client):
    """User A creates a task, User B modifies it. The audit has to show two different users."""
    # User A creates
    res = await client.post("/tasks", json={"title": "X"}, headers={"X-User-Id": "47"})
    task_id = res.json()["id"]

    # User B modifies
    await client.put(f"/tasks/{task_id}", json={"title": "Y"}, headers={"X-User-Id": "23"})

    # Verify
    result = await db.execute(text("""
        SELECT changed_by FROM audit.task_log
        WHERE entity_id = :tid ORDER BY changed_at
    """), {"tid": task_id})
    users = [row[0] for row in result]
    assert users == [47, 23], f"User_ids aren't the expected ones: {users}"

Mistake 2 (practical): current_setting without missing_ok=true raises an error

Symptom: a test runs an INSERT on tasks without going through the dependency. The trigger raises an error: ERROR: unrecognized configuration parameter "audit.user_id". The INSERT fails.

Why it happens: current_setting('audit.user_id') (with no second argument) raises an error if the setting doesn't exist. In tests or jobs that don't use the dependency, the setting never gets established.

How to tell: look at the trigger's signature. If it uses current_setting('audit.user_id') without the , true, this bug shows up.

How to fix it: always use current_setting('audit.user_id', true). The second argument is "missing_ok": it returns an empty string if the setting doesn't exist, instead of raising an error. Combined with NULLIF(..., '') it results in NULL, which is what we want.

Mistake 3 (conceptual): the trigger fires BEFORE instead of AFTER

Symptom: the trigger uses to_jsonb(NEW) on INSERT but NEW.id is always NULL. The audit log ends up with entity_id = NULL.

Why it happens: a BEFORE INSERT trigger fires before the row actually gets inserted. Columns with BIGSERIAL (autoincrement) don't have a value yet; PostgreSQL assigns them at insert time. Only in AFTER INSERT is NEW.id populated.

How to tell: look at whether the trigger was declared with BEFORE or AFTER. If it uses BIGSERIAL or IDENTITY and needs the ID, it has to be AFTER.

How to fix it: always AFTER INSERT OR UPDATE OR DELETE for audit logs (unless you have a specific reason to use BEFORE, like modifying the value before inserting — a different case from auditing).

CREATE TRIGGER task_log_trigger
AFTER INSERT OR UPDATE OR DELETE ON tasks  -- AFTER, not BEFORE
FOR EACH ROW
EXECUTE FUNCTION audit.task_log_trigger();

Mistake 4 (operational): the audit log impacts INSERT performance

Symptom: after adding the trigger, INSERTs on tasks that used to take 0.5ms now take 8ms. A bulk insert migration that used to take 30 seconds now takes 6 minutes.

Why it happens: every INSERT now also runs an INSERT into audit.task_log, which:

• Doubles the write IO.
• Updates three indexes (entity_time, changed_by_time, diff_gin).
• The GIN index on JSONB is especially costly to maintain.

This is the expected overhead of an audit log with triggers (~5-10% typically, up to 15x for large bulk inserts with heavy indexes).

How to tell: benchmark before/after the trigger. If the overhead exceeds what's acceptable, evaluate:

How to fix it (options):

1. Accept the overhead. For apps with normal traffic, 5-10% is acceptable in exchange for the audit guarantee.

2. Remove audit log indexes that aren't used. The idx_audit_task_log_diff_gin is expensive and only useful if you run queries like "which changes touched status." If not, dropping it saves ~30% of the overhead.

3. For specific bulk operations, temporarily disable the trigger. This is advanced — it requires making sure the bulk doesn't need auditing:

BEGIN;
ALTER TABLE tasks DISABLE TRIGGER task_log_trigger;
-- massive bulk insert
COPY tasks FROM '/path/to/data.csv';
ALTER TABLE tasks ENABLE TRIGGER task_log_trigger;
COMMIT;

4. Use the event listener approach (capsule 06) if the trigger is the bottleneck.

5. For extreme cases, move the audit log to an async table via LISTEN/NOTIFY or a job that consumes events. Covered in capsule 07.

Mistake 5 (conceptual): assuming the trigger sees changes from BEFORE triggers

Symptom: your table has a BEFORE UPDATE trigger that sets updated_at = NOW(). The audit log shows updated_at changes on every UPDATE, even though the app didn't change it explicitly.

Why it happens: triggers run in a chain. The BEFORE UPDATE fires first and modifies NEW.updated_at. When the AFTER UPDATE (audit) fires afterward, it sees the NEW with the updated_at already changed.

How to tell: review the existing triggers on the table with \d+ tasks in psql. If there's a BEFORE trigger that modifies columns, this can happen.

How to fix it: make sure the columns modified by BEFORE triggers are in the audit's exclusion list. In our example, updated_at is already excluded, which is why it doesn't show up in the log.

Exercises

Exercise 1: implement the full pattern on a new table

Your app is going to add a comments table. Schema:

CREATE TABLE comments (
    id BIGSERIAL PRIMARY KEY,
    task_id BIGINT NOT NULL REFERENCES tasks(id),
    body TEXT NOT NULL,
    is_internal BOOLEAN NOT NULL DEFAULT FALSE,
    edited_count INTEGER NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Implement the complete audit log: the audit.comment_log table, the audit.comment_log_trigger function, and the attached trigger. Document which columns you exclude and why.

-- The audit log table
CREATE TABLE audit.comment_log (
    id BIGSERIAL PRIMARY KEY,
    entity_id BIGINT NOT NULL,
    action TEXT NOT NULL CHECK (action IN ('INSERT', 'UPDATE', 'DELETE')),
    changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    changed_by BIGINT NULL,
    diff JSONB NOT NULL,
    request_id UUID NULL,
    source TEXT NULL CHECK (source IN ('api', 'cron', 'migration', 'manual') OR source IS NULL)
);

CREATE INDEX idx_audit_comment_log_entity_time
    ON audit.comment_log (entity_id, changed_at DESC);

CREATE INDEX idx_audit_comment_log_changed_by_time
    ON audit.comment_log (changed_by, changed_at DESC)
    WHERE changed_by IS NOT NULL;

-- The trigger function
CREATE OR REPLACE FUNCTION audit.comment_log_trigger()
RETURNS TRIGGER AS $$
DECLARE
    v_user_id BIGINT;
    v_request_id UUID;
    v_source TEXT;
    v_diff JSONB;
    v_excluded_keys TEXT[] := ARRAY[
        'created_at',     -- redundant with changed_at
        'edited_count'    -- a counter; noise
    ];
BEGIN
    v_user_id := NULLIF(current_setting('audit.user_id', true), '')::BIGINT;
    v_request_id := NULLIF(current_setting('audit.request_id', true), '')::UUID;
    v_source := NULLIF(current_setting('audit.source', true), '');

    CASE TG_OP
        WHEN 'INSERT' THEN
            v_diff := audit.diff_jsonb('{}'::jsonb, to_jsonb(NEW), v_excluded_keys);
        WHEN 'DELETE' THEN
            v_diff := audit.diff_jsonb(to_jsonb(OLD), '{}'::jsonb, v_excluded_keys);
        WHEN 'UPDATE' THEN
            v_diff := audit.diff_jsonb(to_jsonb(OLD), to_jsonb(NEW), v_excluded_keys);
    END CASE;

    IF TG_OP = 'UPDATE' AND v_diff = '{}'::jsonb THEN
        RETURN NULL;
    END IF;

    INSERT INTO audit.comment_log (entity_id, action, changed_by, request_id, source, diff)
    VALUES (COALESCE(NEW.id, OLD.id), TG_OP, v_user_id, v_request_id, v_source, v_diff);

    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER comment_log_trigger
AFTER INSERT OR UPDATE OR DELETE ON comments
FOR EACH ROW
EXECUTE FUNCTION audit.comment_log_trigger();

## Table `comments`

**Audited:** task_id (change of ownership), body (content), is_internal (visibility flag)

**Excluded:**
- `created_at`: redundant with the log's `changed_at`.
- `edited_count`: a counter with no audit value.

**Decided:** 2026-04-15 by the Backend team.

Exercise 2: detect missing context with a test

Write an async pytest test that verifies:

a) When the audit_context dependency is used, the audit log captures the right user_id. b) When a direct INSERT is made in raw SQL (with no dependency), changed_by is NULL.

# tests/test_audit_context.py
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

pytestmark = pytest.mark.asyncio


async def test_audit_captura_user_id_via_dependency(client, session: AsyncSession):
    """When the dependency sets the context, the audit log reflects it."""
    res = await client.post(
        "/tasks",
        json={"title": "Test"},
        headers={"X-User-Id": "47"},
    )
    task_id = res.json()["id"]

    # Check the audit log
    result = await session.execute(text("""
        SELECT changed_by FROM audit.task_log
        WHERE entity_id = :tid AND action = 'INSERT'
    """), {"tid": task_id})
    assert result.scalar() == 47


async def test_insert_directo_sin_contexto_audita_con_null(session: AsyncSession):
    """A direct INSERT (not going through the dependency) audits changed_by = NULL."""
    # Insert with no SET LOCAL
    result = await session.execute(text("""
        INSERT INTO tasks (title) VALUES ('From script')
        RETURNING id
    """))
    task_id = result.scalar()
    await session.commit()

    # Check the audit log
    result = await session.execute(text("""
        SELECT changed_by FROM audit.task_log
        WHERE entity_id = :tid AND action = 'INSERT'
    """), {"tid": task_id})
    assert result.scalar() is None  # NULL because there was no context


async def test_insert_directo_con_set_local_manual_audita_correcto(session: AsyncSession):
    """A manual SET LOCAL in SQL works too."""
    await session.execute(text("SET LOCAL audit.user_id = '99'"))

    result = await session.execute(text("""
        INSERT INTO tasks (title) VALUES ('From manual')
        RETURNING id
    """))
    task_id = result.scalar()
    await session.commit()

    result = await session.execute(text("""
        SELECT changed_by FROM audit.task_log
        WHERE entity_id = :tid
    """), {"tid": task_id})
    assert result.scalar() == 99

Exercise 3: implement the "history of a task" query

Write the SQL query that returns a task's complete history: timestamp, action, user, and a readable diff. The query is the one a GET /tasks/{id}/history endpoint would run.

-- Base query: a task's complete history
SELECT
    al.id AS log_id,
    al.changed_at,
    al.action,
    al.changed_by,
    al.source,
    al.diff
FROM audit.task_log al
WHERE al.entity_id = :task_id
ORDER BY al.changed_at DESC
LIMIT 100;

-- Version with a join to resolve the user's name
SELECT
    al.id AS log_id,
    al.changed_at,
    al.action,
    al.changed_by,
    u.email AS changed_by_email,  -- assuming a users table
    al.source,
    al.diff
FROM audit.task_log al
LEFT JOIN users u ON u.id = al.changed_by
WHERE al.entity_id = :task_id
ORDER BY al.changed_at DESC
LIMIT 100;

-- A "human-friendly" version that extracts meaningful changes from the diff
SELECT
    al.changed_at,
    al.action,
    u.email AS changed_by_email,
    -- Extract each change as a separate row
    jsonb_object_keys(al.diff) AS column_changed,
    al.diff->jsonb_object_keys(al.diff)->0 AS old_value,
    al.diff->jsonb_object_keys(al.diff)->1 AS new_value
FROM audit.task_log al
LEFT JOIN users u ON u.id = al.changed_by
WHERE al.entity_id = :task_id
ORDER BY al.changed_at DESC, column_changed
LIMIT 200;

@app.get("/tasks/{task_id}/history")
async def task_history(task_id: int, db: AsyncSession = Depends(get_session)):
    result = await db.execute(text("""
        SELECT
            al.changed_at,
            al.action,
            al.changed_by,
            al.source,
            al.diff
        FROM audit.task_log al
        WHERE al.entity_id = :tid
        ORDER BY al.changed_at DESC
        LIMIT 100
    """), {"tid": task_id})

    return [
        {
            "changed_at": row.changed_at.isoformat(),
            "action": row.action,
            "changed_by": row.changed_by,
            "source": row.source,
            "changes": [
                {
                    "field": key,
                    "old": diff_value[0],
                    "new": diff_value[1],
                }
                for key, diff_value in row.diff.items()
            ],
        }
        for row in result
    ]

[
  {
    "changed_at": "2026-04-15T14:33:01Z",
    "action": "UPDATE",
    "changed_by": 23,
    "source": "api",
    "changes": [{"field": "status", "old": "in_progress", "new": "blocked"}]
  },
  {
    "changed_at": "2026-04-15T09:14:22Z",
    "action": "UPDATE",
    "changed_by": 47,
    "source": "api",
    "changes": [{"field": "status", "old": "open", "new": "in_progress"}]
  },
  {
    "changed_at": "2026-04-14T11:00:00Z",
    "action": "INSERT",
    "changed_by": 47,
    "source": "api",
    "changes": [{"field": "title", "old": null, "new": "Buy milk"}]
  }
]

Exercise 4: add extra metadata to the audit (a request_id that correlates with logs)

Your team uses middleware that adds an X-Request-Id to every HTTP request, which also gets logged to stdout (Loki/CloudWatch). You want to correlate the audit log with the application log using that ID.

Modify the audit_context dependency to extract the request_id from the middleware and propagate it to SET LOCAL audit.request_id. Then write the query that correlates both systems.

# app/middleware.py
import uuid
from starlette.middleware.base import BaseHTTPMiddleware


class RequestIdMiddleware(BaseHTTPMiddleware):
    """Assigns a request_id to each request and exposes it on request.state."""

    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-Id", str(uuid.uuid4()))
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-Id"] = request_id
        return response


# app/main.py
from fastapi import FastAPI
from app.middleware import RequestIdMiddleware

app = FastAPI()
app.add_middleware(RequestIdMiddleware)


# app/audit_context.py
from fastapi import Depends, Request
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_session


async def audit_context(
    request: Request,
    db: AsyncSession = Depends(get_session),
) -> AsyncSession:
    user_id = getattr(request.state, "user_id", None)
    request_id = request.state.request_id  # from the middleware

    if user_id is not None:
        await db.execute(text("SET LOCAL audit.user_id = :uid"), {"uid": str(user_id)})

    await db.execute(text("SET LOCAL audit.request_id = :rid"), {"rid": request_id})
    await db.execute(text("SET LOCAL audit.source = 'api'"))

    return db

-- Given a request_id (typically from the app's log), find every change it generated
SELECT
    al.changed_at,
    al.entity_id AS task_id,
    al.action,
    al.diff
FROM audit.task_log al
WHERE al.request_id = :request_id
ORDER BY al.changed_at;

-- If a single request generated >100 changes, there's probably a bug
SELECT request_id, COUNT(*) AS changes, MIN(changed_at), MAX(changed_at)
FROM audit.task_log
WHERE changed_at > NOW() - INTERVAL '1 hour'
GROUP BY request_id
HAVING COUNT(*) > 100
ORDER BY changes DESC;

Exercise 5: defend the trigger approach against "let's go with SQLAlchemy event listeners"

Your senior teammate proposes implementing audit with event.listens_for(Session, "after_flush") in SQLAlchemy instead of triggers. Argue why triggers are better for this case (TaskFlow's audit log, a monolithic API with managed PostgreSQL).

Summary and next step

In this capsule you learned:

• The trigger is a universal interceptor at the DB level: it guarantees that EVERY change (including raw SQL, jobs, migrations) gets audited. That's the key difference from event listeners.
• The SET LOCAL and current_setting detail is what separates a useless audit log (changed_by = NULL) from a useful one. The FastAPI dependency sets the context at the start of each request; the trigger reads it when it fires.
• The - operator between JSONBs doesn't generate a diff: you need a helper function like audit.diff_jsonb that iterates keys and compares with IS DISTINCT FROM.
• The {"column": [old, new]} format is directly transformable into human-friendly APIs and allows GIN indexing for analytical queries.
• Filtering out sensitive and noisy columns is the trigger's responsibility: the v_excluded_keys array controls what enters the log. Passwords, tokens, and volatile counters are always excluded.
• AFTER triggers, not BEFORE: BIGSERIAL and autoincrement columns are only populated after the INSERT.
• SET LOCAL (not SET): SET persists on the connection and contaminates subsequent requests in the pool. SET LOCAL only lasts until the COMMIT.
• current_setting('audit.user_id', true) with missing_ok: avoids errors when an INSERT comes from a script with no context.
• Acceptable overhead (5-10%) for typical apps: if you need more, the options include removing indexes from the log, temporarily disabling the trigger for bulk, or migrating to an asynchronous approach (capsule 07).

Before moving on you should be able to:

• Implement the complete pattern (schema + helper function + trigger + dependency) on a new table in under 30 minutes.
• Diagnose the changed_by = NULL bug and fix it.
• Defend the trigger approach against an event listener with concrete arguments.
• Write the GET /tasks/{id}/history endpoint that consumes audit.task_log.
• Detect the SET vs SET LOCAL bug with a test that verifies isolation between requests.

Next capsule — The history tables pattern. You're going to learn the complementary approach: instead of storing the "change" (an audit log), storing the "complete row with temporal marks" (a history table). You're going to understand when each one applies (the audit log for "who and why," the history table for "exactly how the data looked on March 15"), why they can coexist, and when the history table's write cost isn't justified. Capsule 04 lays the foundation; 05 goes to event sourcing and 06 to the Python approach; 07 covers retention. Each one is a different piece of the same problem: preserving history.

Resources

1. PostgreSQL Documentation — PL/pgSQL Trigger Functions — the official reference. Chapter 41.10 covers TG_OP, NEW, OLD, and the special variables you use in the trigger.
2. PostgreSQL Documentation — SET and SET LOCAL — the difference is critical. Required reading if you've never used custom GUCs.
3. PostgreSQL Documentation — current_setting() and set_config() — the pair of functions for reading/writing custom GUCs from code.
4. PostgreSQL Documentation — JSONB operators — the reference for ||, -, ?, ->, ->>, jsonb_object_keys. Useful when writing and debugging diff_jsonb.
5. Supabase — Audit logs with triggers — a real implementation in multi-tenant production. A direct inspiration for this capsule's pattern. Supabase's audit.record_version() code is a key reference.
6. pgAudit extension — an alternative for auditing at the query level (not just data changes). Useful to know it exists; it's typically complementary, not a substitute for the custom trigger.
7. SQLAlchemy 2.0 — text() and bind parameters — the reference for understanding why text("SET LOCAL audit.user_id = :uid"), {"uid": ...} prevents SQL injection.
8. Vlad Mihalcea — "How to extract change data events from PostgreSQL" — an advanced perspective (CDC with Debezium) that shows when the trigger-based audit log doesn't scale and you need a different approach. Useful as a horizon.

Module 3 — SQL Patterns for Production APIs Guide

Next capsule: The history tables pattern — when to store the complete row instead of the change.