Module 3: Audit Logs and History Tables

Module 3 deliverable: Audit Trail in TaskFlow (standalone)

Next →

What are you going to build and why?

You've reached the module's close. In capsules 02-07 you saw what to audit (capsule 02), three approaches in SQL (triggers, history tables, event sourcing) plus one in Python (an event listener), and the operational problem of retention and partitioning. Now comes the integration. You're going to build the complete pattern applied to a standalone tasks table: a PostgreSQL trigger that writes to audit.task_log, a FastAPI dependency that passes the user_id with SET LOCAL, monthly partitioning with automated partition creation, endpoints that consume the audit, and tests that verify the pattern works end-to-end.

The project is standalone: the tasks table exists on its own, with no RLS, no multi-tenancy, no cursor pagination from the guide. That's by design. Module 8 (the guide's capstone project) takes this same pattern and combines it with multi-tenancy (RLS from module 4), pagination (module 1), and the rest. Isolating the audit here lets you master it without the combined complexity.

You're going to come away with a GitHub repo that demonstrates the complete pattern: you can clone it, bring it up with docker compose up, try it with curl, verify the audit log with SQL queries, and show the result in an interview or a code review. It's portfolio-worthy in the sense that any senior dev recognizes the pattern on sight, and being able to defend it line by line is what separates "I read about auditing" from "I implemented auditing correctly."

Project objective

By the time you complete this project:

  • You'll have implemented an audit log with PostgreSQL triggers over a real tasks table, with all the module's decisions: filtering out excluded columns, generating the JSONB diff, capturing the user_id via SET LOCAL.
  • You'll have partitioned audit.task_log by month from day 1, with an automated function that creates future partitions.
  • You'll have exposed FastAPI endpoints that demonstrate the pattern end-to-end: CRUD on tasks that generates audit entries, an "history of a task" endpoint that queries the audit log.
  • You'll have written automated tests that verify: the changed_by is captured correctly, the JSONB diff excludes sensitive columns, the inserts go to the right partition, the history endpoint returns correct data.
  • You'll have documented the decisions in AUDIT-DECISIONS.md and BENCHMARKS.md: why triggers, which columns to exclude, what retention policy, what measured overhead the trigger has.

How it fits with what you learned

Module conceptWhere it's used in the project
Capsule 02: what to auditThe decision about which columns to exclude (internal_notes, view_count); the documentation in AUDIT-DECISIONS.md
Capsule 03: PostgreSQL triggersThe audit.task_log_trigger function + the audit.diff_jsonb helper function + the integration with the FastAPI dependency
Capsule 04: history tablesNOT used in this project (the audit log is enough). The decision is documented.
Capsule 05: event sourcingNOT used in this project (over-engineering for simple auditing). The decision is documented.
Capsule 06: SQLAlchemy listenersNOT used in this project (self-hosted PostgreSQL allows triggers). Documentation of when to migrate.
Capsule 07: partitioning + retentionThe monthly-partitioned schema, the create_partition_for_month function, the maintenance job

Think of the project as the consolidated pattern: you applied triggers (capsule 03), filtered with judgment (capsule 02), partitioned from day 1 (capsule 07). Capsules 04, 05, and 06 gave you alternatives you documented but didn't use — because for this case (standalone TaskFlow, self-hosted PostgreSQL), triggers + partitioning are the right answer.

Technical specifications

Stack

  • Language: Python 3.12+
  • Main framework: FastAPI 0.110+
  • ORM: SQLAlchemy 2.0+ (async)
  • DB driver: asyncpg
  • Database: PostgreSQL 16+ (needed for some partitioning features)
  • Migrations: Alembic 1.13+
  • Container: Docker + docker-compose for a reproducible environment
  • Tests: pytest + pytest-asyncio + httpx for an async client

Initial setup

# Clone the scaffold repo
git clone https://github.com/<your-org>/taskflow-audit-module
cd taskflow-audit-module

# Bring up PostgreSQL 16
docker compose up -d postgres

# Set up the virtualenv and deps
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt

# Apply the migrations
alembic upgrade head

# Verify the initial partitions got created
psql postgresql://postgres:postgres@localhost/taskflow \
    -c "\dt+ audit.*"

# Bring up FastAPI
uvicorn app.main:app --reload

Required functionality

1. The base tasks schema with a partitioned audit log

Schema:

-- 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,  -- a NON-audited column (sensitive)
    view_count BIGINT NOT NULL DEFAULT 0,  -- a NON-audited column (noise)
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- A trigger to auto-update updated_at
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER tasks_update_updated_at
BEFORE UPDATE ON tasks
FOR EACH ROW EXECUTE FUNCTION update_updated_at();

-- The audit schema
CREATE SCHEMA IF NOT EXISTS audit;

-- The partitioned audit table
CREATE TABLE audit.task_log (
    id BIGSERIAL,
    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),
    PRIMARY KEY (id, changed_at)
) PARTITION BY RANGE (changed_at);

-- The default partition (a safety net)
CREATE TABLE audit.task_log_default PARTITION OF audit.task_log DEFAULT;

-- Indexes on the parent table (they apply to the partitions)
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);

Expected behavior:

  • tasks can receive INSERT/UPDATE/DELETE normally.
  • Every operation fires the audit log trigger automatically.
  • The inserts into audit.task_log go to the partition corresponding to changed_at's month.

2. The diff helper function and the audit trigger

-- The helper function that generates the JSONB diff
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
    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;

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

3. The automatic partition creation function

CREATE OR REPLACE FUNCTION audit.create_partition_for_month(
    p_year INTEGER,
    p_month INTEGER
)
RETURNS TEXT AS $$
DECLARE
    v_partition_name TEXT;
    v_start_date DATE;
    v_end_date DATE;
BEGIN
    v_partition_name := format('audit.task_log_y%sm%s',
        p_year::TEXT, lpad(p_month::TEXT, 2, '0'));

    v_start_date := make_date(p_year, p_month, 1);
    v_end_date := v_start_date + INTERVAL '1 month';

    EXECUTE format(
        'CREATE TABLE IF NOT EXISTS %s PARTITION OF audit.task_log FOR VALUES FROM (%L) TO (%L)',
        v_partition_name, v_start_date, v_end_date
    );

    RETURN v_partition_name;
END;
$$ LANGUAGE plpgsql;

And a script that creates the initial partitions:

# scripts/create_initial_partitions.py
import asyncio
from datetime import date

from sqlalchemy import text

from app.db import AsyncSessionLocal


async def main():
    today = date.today()

    async with AsyncSessionLocal() as session:
        # Create the partition for last month, the current one, and the next 3
        for offset in range(-1, 4):
            year = today.year
            month = today.month + offset
            while month > 12: month -= 12; year += 1
            while month < 1: month += 12; year -= 1

            result = await session.execute(text(
                "SELECT audit.create_partition_for_month(:y, :m)"
            ), {"y": year, "m": month})
            print(f"Created: {result.scalar()}")

        await session.commit()


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

4. The FastAPI dependency for the audit context

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

from fastapi import Depends, Header, 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),
    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 would be extracted from the JWT, not from a header.
    For this demo project, we simplify.
    """
    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

5. CRUD endpoints + a history endpoint

# app/main.py
from typing import Optional, List

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

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

app = FastAPI(title="TaskFlow Audit Module")


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


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


class TaskOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    description: Optional[str]
    status: str
    priority: int
    assignee_id: Optional[int]


@app.post("/tasks", status_code=status.HTTP_201_CREATED, response_model=TaskOut)
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 task


@app.get("/tasks/{task_id}", response_model=TaskOut)
async def get_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")
    return task


@app.put("/tasks/{task_id}", response_model=TaskOut)
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 task


@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
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()


@app.get("/tasks/{task_id}/history")
async def task_history(
    task_id: int,
    limit: int = 100,
    db: AsyncSession = Depends(audit_context),
):
    """
    Returns a task's change history.
    Filters by changed_at over the last 12 months to take advantage of partition pruning.
    """
    result = await db.execute(text("""
        SELECT
            changed_at, action, changed_by, source, diff
        FROM audit.task_log
        WHERE entity_id = :tid
          AND changed_at >= NOW() - INTERVAL '12 months'
        ORDER BY changed_at DESC
        LIMIT :limit
    """), {"tid": task_id, "limit": limit})

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

6. Automated tests

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

pytestmark = pytest.mark.asyncio


async def test_insert_genera_audit_con_changed_by(client):
    res = await client.post(
        "/tasks",
        json={"title": "T1", "priority": 1},
        headers={"X-User-Id": "47"},
    )
    assert res.status_code == 201
    task_id = res.json()["id"]

    # Check the audit
    history = await client.get(f"/tasks/{task_id}/history")
    entries = history.json()
    assert len(entries) == 1
    assert entries[0]["action"] == "INSERT"
    assert entries[0]["changed_by"] == 47
    titles_in_diff = [c for c in entries[0]["changes"] if c["field"] == "title"]
    assert len(titles_in_diff) == 1
    assert titles_in_diff[0]["new"] == "T1"


async def test_update_genera_audit_solo_de_cambios(client):
    # Create
    res = await client.post(
        "/tasks", json={"title": "T1"}, headers={"X-User-Id": "47"}
    )
    task_id = res.json()["id"]

    # Update only the title
    await client.put(
        f"/tasks/{task_id}",
        json={"title": "T1 modified"},
        headers={"X-User-Id": "23"},
    )

    history = await client.get(f"/tasks/{task_id}/history")
    entries = history.json()

    update_entries = [e for e in entries if e["action"] == "UPDATE"]
    assert len(update_entries) == 1
    assert update_entries[0]["changed_by"] == 23

    # Only the title changed, no other columns
    fields_changed = [c["field"] for c in update_entries[0]["changes"]]
    assert "title" in fields_changed
    assert "priority" not in fields_changed
    assert "status" not in fields_changed


async def test_internal_notes_no_se_auditan(client, session):
    # Create with internal_notes
    await session.execute(text("SET LOCAL audit.user_id = '47'"))
    res = await session.execute(text("""
        INSERT INTO tasks (title, internal_notes)
        VALUES ('T1', 'Private admin note')
        RETURNING id
    """))
    task_id = res.scalar()
    await session.commit()

    # Verify the audit doesn't contain internal_notes
    result = await session.execute(text("""
        SELECT diff FROM audit.task_log WHERE entity_id = :tid
    """), {"tid": task_id})
    diff = result.scalar()

    assert "internal_notes" not in diff, "internal_notes leaked into the audit log"


async def test_view_count_no_se_audita(client, session):
    res = await client.post("/tasks", json={"title": "T1"}, headers={"X-User-Id": "47"})
    task_id = res.json()["id"]

    # UPDATE only view_count (it shouldn't get audited)
    await session.execute(text("""
        UPDATE tasks SET view_count = view_count + 1 WHERE id = :tid
    """), {"tid": task_id})
    await session.commit()

    # Verify: only the initial INSERT is there, there's no UPDATE in the audit
    result = await session.execute(text("""
        SELECT COUNT(*) FROM audit.task_log WHERE entity_id = :tid AND action = 'UPDATE'
    """), {"tid": task_id})
    assert result.scalar() == 0, "A view_count UPDATE must not be audited"


async def test_delete_genera_audit_con_diff_completo(client):
    res = await client.post(
        "/tasks", json={"title": "T1", "priority": 5}, headers={"X-User-Id": "47"}
    )
    task_id = res.json()["id"]

    await client.delete(f"/tasks/{task_id}", headers={"X-User-Id": "23"})

    history = await client.get(f"/tasks/{task_id}/history")
    entries = history.json()

    delete_entries = [e for e in entries if e["action"] == "DELETE"]
    assert len(delete_entries) == 1
    assert delete_entries[0]["changed_by"] == 23

    # On DELETE, the diff has [value, null] for each column
    title_change = next(c for c in delete_entries[0]["changes"] if c["field"] == "title")
    assert title_change["old"] == "T1"
    assert title_change["new"] is None


async def test_inserciones_van_a_particion_correcta(session):
    """Verifies the partitioning works: rows inserted today go to this month's partition."""
    from datetime import date

    today = date.today()
    expected_partition = f"task_log_y{today.year}m{today.month:02d}"

    # Insert
    res = await session.execute(text("""
        INSERT INTO tasks (title) VALUES ('Test partition') RETURNING id
    """))
    task_id = res.scalar()
    await session.commit()

    # Verify the audit log went to the right partition
    result = await session.execute(text(f"""
        SELECT COUNT(*) FROM audit.{expected_partition}
        WHERE entity_id = :tid
    """), {"tid": task_id})

    assert result.scalar() >= 1, f"The audit didn't go to partition {expected_partition}"


async def test_endpoint_history_devuelve_orden_correcto(client):
    res = await client.post("/tasks", json={"title": "v1"}, headers={"X-User-Id": "47"})
    task_id = res.json()["id"]

    await client.put(
        f"/tasks/{task_id}", json={"title": "v2"}, headers={"X-User-Id": "47"}
    )
    await client.put(
        f"/tasks/{task_id}", json={"title": "v3"}, headers={"X-User-Id": "47"}
    )

    history = await client.get(f"/tasks/{task_id}/history")
    entries = history.json()

    # Most recent first
    assert len(entries) == 3
    assert entries[0]["action"] == "UPDATE"
    assert entries[-1]["action"] == "INSERT"

Validation and error handling

What has to be validated

  • title can't be empty on CREATE.
  • status only accepts values from the CHECK constraint (open, in_progress, closed, archived).
  • priority has to be an integer (Pydantic validates it).
  • The X-User-Id header is optional but if it's present it has to be an integer.

Errors that have to be handled

  • Task not found: GET/PUT/DELETE /tasks/{id} with a nonexistent id → HTTP 404.
  • Invalid status: PUT /tasks/{id} with status="foo" → HTTP 422 (Pydantic).
  • Internal server error: if the audit trigger fails for some reason (e.g. a full default partition and a constraint), HTTP 500. The whole transaction rolls back (audit + change). Log the error.

Minimal implementation example

This is the project's skeleton. You extend it to cover all the required functionality. It isn't the complete solution.

taskflow-audit-module/
├── app/
│   ├── __init__.py
│   ├── db.py                  # engine + AsyncSession
│   ├── models.py              # Task model
│   ├── audit_context.py       # the dependency with SET LOCAL
│   └── main.py                # FastAPI app + endpoints
├── alembic/
│   ├── env.py
│   └── versions/
│       └── 001_initial_schema.py
├── scripts/
│   ├── create_initial_partitions.py
│   └── verify_audit.py
├── jobs/
│   └── maintain_partitions.py  # a monthly cron
├── tests/
│   ├── conftest.py            # fixtures: session, client
│   └── test_audit.py
├── docker-compose.yml          # postgres 16
├── Dockerfile
├── requirements.txt
├── README.md
├── AUDIT-DECISIONS.md          # documentation of the what-to-audit decisions
└── BENCHMARKS.md               # overhead measurements

# app/main.py — minimal skeleton
from fastapi import FastAPI

app = FastAPI(title="TaskFlow Audit Module")


@app.get("/health")
async def health():
    return {"status": "ok"}

# TODO: implement the CRUD endpoints + history (see "Required functionality")
# docker-compose.yml — minimal skeleton
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: taskflow
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

This skeleton:

  • ✅ Is runnable (you can run docker compose up and uvicorn).
  • ✅ Shows the expected repo structure.
  • ❌ Does NOT include the required functionality (that's your job).

Evaluation rubric (self-check)

Functionality (50 points)

  • (10 pts) The PostgreSQL trigger implemented correctly. It captures INSERT/UPDATE/DELETE, writes to audit.task_log with a JSONB diff.
  • (10 pts) SET LOCAL audit.user_id integrated. The changed_by in the log reflects the HTTP request's X-User-Id.
  • (10 pts) Monthly partitioning implemented. The audit.task_log table partitioned, the create_partition_for_month function, the default partition present.
  • (5 pts) A correct audit.diff_jsonb function. It generates the diff in the {"col": [old, new]} format, and excludes keys from the excluded_keys array.
  • (5 pts) The CRUD endpoints working. POST/GET/PUT/DELETE on /tasks.
  • (5 pts) The /tasks/{id}/history endpoint working. It returns the history with parsed changes, ordered by date desc.
  • (5 pts) The excluded columns (internal_notes, view_count, created_at, updated_at) do NOT appear in the log.

Tests (20 points)

  • (5 pts) A test verifies the changed_by is captured correctly.
  • (5 pts) A test verifies the excluded columns don't show up in the log.
  • (5 pts) A test verifies an UPDATE of an excluded column doesn't generate an audit entry.
  • (5 pts) A test verifies the inserts go to the current month's correct partition.

Code quality (15 points)

  • (5 pts) Consistent type hints throughout the Python code.
  • (5 pts) Migrations with Alembic, not loose SQL scripts.
  • (5 pts) A clear project structure (separating the model, the context, the endpoints).

Documentation (15 points)

  • (5 pts) README.md with clear setup and usage instructions.
  • (5 pts) AUDIT-DECISIONS.md documents which columns are audited and which aren't, with reasons.
  • (5 pts) BENCHMARKS.md documents the trigger's measured overhead (INSERT with vs without the trigger).

Extra credit (optional, up to +10 pts)

  • (+3 pts) A runnable jobs/maintain_partitions.py job that creates future partitions.
  • (+3 pts) Integration with a logging system (Loki, CloudWatch, simple structured stdout) that correlates the audit's request_id with the app's logs.
  • (+4 pts) An implementation of the archive to S3 (localstack for development, real S3 for production) with post-upload verification.

Total: 100 points Pass: ≥70 points. Exemplary: ≥90 points + extras.

Common mistakes in this project

Mistake 1: changed_by is always NULL

Why it happens: the audit_context dependency isn't being applied, or the SET LOCAL doesn't run before the INSERT/UPDATE.

How to fix it: verify the endpoint uses Depends(audit_context) (not Depends(get_session)). Verify the text("SET LOCAL...") runs in the dependency. A test that fails:

async def test_changed_by_se_captura(client):
    res = await client.post("/tasks", json={"title": "T1"}, headers={"X-User-Id": "47"})
    history = await client.get(f"/tasks/{res.json()['id']}/history")
    assert history.json()[0]["changed_by"] == 47

Mistake 2: the inserts go to the default partition

Why it happens: the partitions for the current month aren't created. PostgreSQL puts the inserts in the default.

How to fix it: run python scripts/create_initial_partitions.py before trying the app. Verify:

SELECT COUNT(*) FROM audit.task_log_default;
-- It should be 0 in normal operation

Mistake 3: the trigger fires twice (duplicate audit)

Why it happens: the migration ran twice, or uvicorn's hot-reload re-runs init scripts.

How to fix it: make sure the trigger is created with CREATE OR REPLACE TRIGGER or that it gets cleanly DROPped+CREATEd. Verify:

SELECT tgname FROM pg_trigger WHERE tgrelid = 'tasks'::regclass;
-- There should be only one audit trigger (plus the updated_at one)

Mistake 4: the /tasks/{id}/history endpoint returns an unparsed diff

Why it happens: the code returns the JSONB without transforming it into the {"field": ..., "old": ..., "new": ...} format.

How to fix it: use the module example's transformation:

"changes": [
    {"field": k, "old": v[0], "new": v[1]}
    for k, v in row.diff.items()
],

Mistake 5: intermittent tests from contamination between sessions

Why it happens: the tests don't isolate the sessions correctly, or the SET LOCAL "sticks" between tests.

How to fix it: make sure each test gets a fresh session via a fixture, and that the transaction rolls back at the end of the test:

@pytest_asyncio.fixture
async def session():
    async with AsyncSessionLocal() as s:
        yield s
        await s.rollback()

Mistake 6: forgetting to disable the trigger in bulk operations

Why it happens: a migration or seed script inserts thousands of rows, each one fires the trigger, the audit log explodes.

How to fix it: for bulk inserts where the audit is optional, disable it temporarily:

async def bulk_seed(session, count=10000):
    await session.execute(text("ALTER TABLE tasks DISABLE TRIGGER task_log_trigger"))
    # ... bulk insert
    await session.execute(text("ALTER TABLE tasks ENABLE TRIGGER task_log_trigger"))

Document the bypass in AUDIT-DECISIONS.md.

What to do if you get stuck

  • The trigger doesn't fire: verify with \dft+ audit.* in psql that the function exists, and \d+ tasks that the trigger is active.
  • Partitioning throws an error on INSERT: verify a partition exists for the current month with \dt audit.task_log_*.
  • Tests fail with "schema audit does not exist": the migration wasn't applied. Run alembic upgrade head.
  • The diff's JSONB is enormous: review v_excluded_keys in the trigger, make sure created_at, updated_at, internal_notes, view_count are excluded.
  • The history endpoint is slow: verify the idx_audit_task_log_entity_time index exists. EXPLAIN the endpoint's query.

Resources for the project

  1. PostgreSQL Documentation — PL/pgSQL Trigger Functions — the reference for the trigger.
  2. PostgreSQL Documentation — Table Partitioning — the reference for partitioning.
  3. SQLAlchemy 2.0 — Async — the reference for the async setup.
  4. FastAPI — Dependencies — the reference for understanding Depends(audit_context).
  5. Alembic — Tutorial — the reference for writing the migrations.
  6. pytest-asyncio — Documentation — the reference for async tests with fixtures.
  7. The module's capsules: all of them (capsules 02-07) are reference. If you get stuck on a specific decision, review the corresponding capsule.

What comes next

This project closes module 3 (defensive modeling). What you built gets extended in module 4 and eventually in the capstone project of module 8:

In module 4 (Multi-Tenancy with RLS): you're going to learn to isolate tenants in PostgreSQL. The audit log you built gets enriched with tenant_id: every row of audit.task_log also records which tenant generated the change. The integration is elegant: on top of SET LOCAL audit.user_id, you're going to set SET LOCAL audit.tenant_id. The trigger reads both. RLS also simplifies the audit log's queries — you automatically only see your tenant's logs.

In module 8 (the complete TaskFlow API): this audit pattern gets combined with all the module's others. Cursor pagination on /tasks, soft delete from module 2, multi-tenant RLS from module 4, optimistic locking from module 6, bulk import from module 7. The audit log captures every change — the pattern you built here scales with the added complexity.

Before moving on, make sure your project:

  • ✅ Passes the automated tests (pytest).
  • ✅ The /tasks/{id}/history endpoint returns a clean, ordered audit.
  • ✅ The excluded columns (internal_notes, view_count) don't appear in any audit entry.
  • ✅ The current month's partitions exist and receive the inserts (verify with SELECT COUNT(*) FROM audit.task_log_yYYYYmMM).
  • ✅ AUDIT-DECISIONS.md documents the decisions made.
  • ✅ BENCHMARKS.md shows the trigger's overhead (INSERT with the trigger vs without).

If all that works, you're capable of implementing an audit log with PostgreSQL triggers in real production. Module 4 awaits.

Module 3 — SQL Patterns for Production APIs Guide

Next module: Multi-Tenancy in PostgreSQL — RLS as the isolation mechanism between tenants.

← PreviousRetention and partitioning of audit logsNext →Multi-Tenancy in PostgreSQL