Module 4: Multi-Tenancy in PostgreSQL

Project: multi-tenant TaskFlow with RLS

What are you going to build and why?

You've reached the close of module 4. So far each capsule taught you a piece: the architectural decision between three models (capsule 02), a shared schema with its mitigations (03), RLS conceptually (04), RLS integrated with FastAPI + SQLAlchemy async (05), schema-per-tenant for enterprise cases (06), and the catalog of anti-patterns that end in a breach (07). In this capsule you're going to integrate everything into a working mini-project: TaskFlow, a SaaS task management API with three fictional tenants — Acme Corp, Globex, and Initech — that demonstrates multi-tenant isolation in production.

Module 4's TaskFlow is NOT the complete version of module 8's capstone project. It's the multi-tenant foundation: a shared schema with working RLS, an audit log that records each change's tenant_id, cursor pagination that respects the isolation, soft deletes with partial indexes per tenant, and a suite of "malicious" tests that verify the isolation against capsule 07's five anti-patterns. Module 8 is going to take this base and add optimistic locking, bulk operations, zero-downtime migrations, and the complete API with all the endpoints. What you build here gets reused directly.

This project demonstrates to an enterprise buyer (or a senior SaaS interviewer) that you know how to implement multi-tenancy correctly. The classic due-diligence question — "how do you guarantee that an employee of one customer company can't see another customer company's data?" — you answer by showing the Alembic migration with ENABLE ROW LEVEL SECURITY + FORCE ROW LEVEL SECURITY, the tenant isolation policy, the automated tests proving that an attacker with deliberately "malicious" queries can't read or write crossed data, and the MULTITENANCY.md documenting the chosen model and the quantitative criteria behind the decision.

Project objective

By completing module 4's TaskFlow you'll:

Implement a working multi-tenant SaaS mini-API (3 fictional tenants, 4 endpoints) with a shared schema + RLS + FORCE ROW LEVEL SECURITY, integrated with FastAPI 0.110+ and SQLAlchemy 2.0 async + asyncpg over PostgreSQL 16+.
Verify the isolation with a pytest-async suite covering capsule 07's five anti-patterns (a query with no filter, an IDOR via db.get, a query builder bug, a cron with no context, tenant_id in the payload).
Connect the patterns from the previous modules: cursor pagination from module 1 (decodable + stable under RLS), soft deletes from module 2 (partial indexes with tenant_id), the audit log from module 3 (a PostgreSQL trigger that captures current_setting('app.tenant_id')).
Document the architectural decision in MULTITENANCY.md with quantitative criteria from capsule 02's decision matrix.

How it fits with what you learned

Module 4 concept	Where it's used in TaskFlow
Capsule 02 — The multi-tenant decision matrix	`MULTITENANCY.md` documents the choice of shared schema + RLS with quantitative criteria
Capsule 03 — Shared schema with `tenant_id`	The SQLAlchemy model for `tasks` with `tenant_id` BIGINT NOT NULL + composite indexes
Capsule 04 — RLS fundamentals	The Alembic migration with `ENABLE` + `FORCE ROW LEVEL SECURITY` + a policy with `WITH CHECK`
Capsule 05 — RLS with FastAPI + SQLAlchemy async	The `get_tenant_session` dependency that runs `SET LOCAL` + `statement_cache_size=0`
Capsule 06 — Schema-per-tenant	The documented justification of why NOT to use schema-per-tenant in TaskFlow
Capsule 07 — Cross-tenant leak anti-patterns	"Malicious" tests that reproduce the five anti-patterns and verify RLS blocks them

And the connectors with the previous modules:

Previous concept	Where it's used in TaskFlow
Module 1 — Cursor pagination	`GET /tasks` with a cursor encoded in base64 (timestamp + id) — the cursor is only valid within the tenant
Module 2 — Soft deletes	The `tasks` table with `deleted_at` + a partial index `WHERE deleted_at IS NULL` that starts with `tenant_id`
Module 3 — Audit log	The `audit.task_history` table with a trigger that captures `tenant_id` from `current_setting`

Think of the project as TaskFlow's prototype: the minimum foundation that demonstrates the module's principles, ready for module 8 to add the remaining features.

Technical specifications

Stack

Language: Python 3.11+
Framework: FastAPI 0.110+
ORM: SQLAlchemy 2.0+ with async + asyncpg
Database: PostgreSQL 16+
Migrations: Alembic 1.13+
Tests: pytest 8+ with pytest-asyncio and httpx.AsyncClient
Auth (simplified): an X-Tenant-ID header (in production it would be a JWT with a tenant_id claim)

Initial setup

# Create the project
mkdir taskflow-m4
cd taskflow-m4
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install \
    "fastapi>=0.110" \
    "sqlalchemy[asyncio]>=2.0" \
    "asyncpg>=0.29" \
    "alembic>=1.13" \
    "uvicorn[standard]>=0.27" \
    "pydantic>=2.6" \
    "pytest>=8" \
    "pytest-asyncio>=0.23" \
    "httpx>=0.27"

# The initial structure
mkdir -p app/{api,db/repositories,jobs} alembic tests/security
touch app/__init__.py app/main.py

Local PostgreSQL with Docker

docker run -d \
    --name taskflow-pg \
    -e POSTGRES_PASSWORD=postgres \
    -e POSTGRES_DB=taskflow \
    -p 5432:5432 \
    postgres:16

# Wait 3 seconds for it to start
sleep 3

# Create the app's role (not superuser, not owner)
docker exec -i taskflow-pg psql -U postgres -d taskflow <<SQL
CREATE ROLE app_user LOGIN PASSWORD 'app_secure_password';
GRANT CONNECT ON DATABASE taskflow TO app_user;
GRANT USAGE ON SCHEMA public TO app_user;
SQL

Environment variables

# .env (don't commit; use .env.example in the repo)
DATABASE_URL=postgresql+asyncpg://app_user:app_secure_password@localhost:5432/taskflow
DATABASE_ADMIN_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/taskflow

DATABASE_URL is the app's role (limited, subject to RLS). DATABASE_ADMIN_URL is used only for running Alembic migrations and for the cron job demonstrating anti-pattern 4.

Required functionality

1. The multi-tenant data model

Tables:

tenants — the 3 fictional tenants (Acme, Globex, Initech). A "global" table, it carries no tenant_id.
tasks — tasks per tenant. It carries tenant_id BIGINT NOT NULL, soft delete (deleted_at), audit fields.
audit.task_history — the audit log with a PostgreSQL trigger that captures changes.

Spec of the Task model:

Field	Type	Notes
`id`	BIGINT PK	Auto-increment
`tenant_id`	BIGINT NOT NULL	FK to `tenants.id`, a composite index
`title`	VARCHAR(200) NOT NULL
`status`	VARCHAR(20) NOT NULL	`open`, `in_progress`, `done`
`priority`	INT NOT NULL DEFAULT 0
`created_at`	TIMESTAMPTZ NOT NULL	`server_default=func.now()`
`updated_at`	TIMESTAMPTZ NOT NULL	`server_default=func.now()` + a `BEFORE UPDATE` trigger
`deleted_at`	TIMESTAMPTZ NULL	NULL = active, NOT NULL = soft deleted

Required indexes:

-- Active reads ordered by created_at (cursor pagination)
CREATE INDEX ix_tasks_tenant_created_active
    ON tasks (tenant_id, created_at DESC, id DESC)
    WHERE deleted_at IS NULL;

-- Filtering by status within the tenant
CREATE INDEX ix_tasks_tenant_status_active
    ON tasks (tenant_id, status)
    WHERE deleted_at IS NULL;

Why partial indexes: most queries only read active tasks (deleted_at IS NULL). A partial index is 60-80% smaller than a full index, it lowers the maintenance cost on INSERT/UPDATE, and it speeds up the common queries. A direct connection with module 2.

2. The initial Alembic migration

Spec: a single migration that creates:

The tenants table with an initial seed of 3 tenants.
The tasks table with the partial indexes.
The audit schema with the task_history table.
A PostgreSQL trigger that captures changes on tasks and writes them to audit.task_history, including the tenant_id from current_setting('app.tenant_id').
Turning on RLS on tasks: ENABLE + FORCE + a tenant_isolation policy with USING + WITH CHECK.
Permissions for the app_user role: SELECT/INSERT/UPDATE/DELETE on tasks, USAGE on sequences.

Expected behavior when running alembic upgrade head:

The tables are created, the indexes are present, the trigger is active, the policy is created.
\d+ tasks in psql shows "Policies: tenant_isolation" and "Force RLS: yes".
The seed of 3 tenants is visible: SELECT * FROM tenants; (with the postgres role, not app_user).

3. The `POST /tasks` endpoint

Spec:

POST /tasks
Headers: X-Tenant-ID: <int>
Body: {"title": "string", "status": "open|in_progress|done", "priority": 0}
Response 201: {"id": 123, "title": "...", "status": "...", "created_at": "..."}

Expected behavior:

Creates a task attributed to the tenant from the X-Tenant-ID header.
If the body includes tenant_id, the endpoint IGNORES it (the Pydantic schema has no such field).
The audit log records the INSERT with the tenant_id from the session setting.
If there's no X-Tenant-ID header → 401.

4. The `GET /tasks` endpoint with cursor pagination

Spec:

GET /tasks?cursor=<base64>&limit=20
Headers: X-Tenant-ID: <int>
Response 200:
{
    "items": [...],
    "next_cursor": "base64string" | null
}

Expected behavior:

Lists NOT-deleted tasks (deleted_at IS NULL) of the header's tenant.
Ordered by created_at DESC, id DESC.
The cursor is encoded in base64 with (created_at_iso, id).
The decoded cursor is used as WHERE (created_at, id) < (cursor_created, cursor_id).
If there are no more results, next_cursor: null.
A cursor between tenants is invalid by construction: if tenant A generates a cursor and tenant B tries to use it, RLS filters and the decoded IDs don't show up.

5. The `GET /tasks/{id}` and `DELETE /tasks/{id}` endpoints (soft delete)

Spec:

GET /tasks/{id}    → 200 with the task | 404 if it doesn't exist OR belongs to another tenant
DELETE /tasks/{id} → 204 if soft-deleted | 404 if it doesn't exist OR belongs to another tenant

Expected behavior:

GET with another tenant's ID returns an indistinguishable 404 (no leak of the existence of IDs).
DELETE sets deleted_at = NOW(), it doesn't physically delete the row.
The audit log records the UPDATE with deleted_at.

6. The audit log with a PostgreSQL trigger

Spec of the trigger:

CREATE OR REPLACE FUNCTION audit.log_task_change() RETURNS TRIGGER AS $$
DECLARE
    tenant_ctx BIGINT;
BEGIN
    -- Capture the tenant from the session setting.
    -- If it isn't set, use the row's tenant_id (a fallback for admin jobs).
    BEGIN
        tenant_ctx := current_setting('app.tenant_id')::BIGINT;
    EXCEPTION WHEN OTHERS THEN
        tenant_ctx := COALESCE(NEW.tenant_id, OLD.tenant_id);
    END;

    INSERT INTO audit.task_history (
        task_id, tenant_id, operation, changed_by_role,
        old_data, new_data, changed_at
    ) VALUES (
        COALESCE(NEW.id, OLD.id),
        tenant_ctx,
        TG_OP,
        current_user,
        CASE WHEN TG_OP IN ('UPDATE', 'DELETE') THEN row_to_json(OLD) END,
        CASE WHEN TG_OP IN ('INSERT', 'UPDATE') THEN row_to_json(NEW) END,
        NOW()
    );
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

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

Why SECURITY DEFINER: the trigger runs with the audit schema owner's permissions, not app_user's. This lets it write to the audit schema without granting direct permissions to app_user (who could delete/modify audit logs if it had a direct GRANT).

7. The seed of three tenants and initial data

# scripts/seed.py
import asyncio
from sqlalchemy import text
from app.db.session import AdminSessionLocal
from app.db.models import Tenant, Task


async def seed():
    async with AdminSessionLocal() as db:
        # 3 tenants
        acme = Tenant(slug="acme", name="Acme Corp")
        globex = Tenant(slug="globex", name="Globex Corporation")
        initech = Tenant(slug="initech", name="Initech")
        db.add_all([acme, globex, initech])
        await db.flush()

        # Tasks per tenant — using the admin role with BYPASSRLS
        async with db.begin_nested():
            await db.execute(text("SET LOCAL row_security = off"))
            for tenant, count in [(acme, 50), (globex, 30), (initech, 10)]:
                for i in range(count):
                    db.add(Task(
                        tenant_id=tenant.id,
                        title=f"{tenant.slug.capitalize()} task #{i+1}",
                        status="open" if i % 3 != 0 else "in_progress",
                        priority=i % 5,
                    ))
        await db.commit()
        print(f"Seeded: Acme={acme.id}, Globex={globex.id}, Initech={initech.id}")


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

Validation and error handling

What has to be validated

The X-Tenant-ID header is present on every /tasks* endpoint (401 if it's missing).
X-Tenant-ID has to be a valid integer (400 if it doesn't parse).
The tenant_id referenced by the header has to exist in tenants (404 if it doesn't — optional, it can also be handled as an "inactive tenant").
title is required on POST /tasks, max 200 chars.
status has to be one of: open, in_progress, done.
cursor is decodable (400 if it isn't valid base64).
limit between 1 and 100 (default 20).

Errors that have to be handled

HTTPException(401): X-Tenant-ID missing.
HTTPException(400): X-Tenant-ID isn't an integer, an invalid cursor, an invalid body.
HTTPException(404): the task doesn't exist or belongs to another tenant (indistinguishable).
HTTPException(500): an unexpected DB error. Logged.
asyncpg.exceptions.PostgresError wrapped into a 500 without leaking the message (so as not to reveal policy names or internal columns).

Minimal implementation example

Below you have the runnable working skeleton. It is NOT the complete solution — it's the base on top of which you have to implement what's missing to satisfy the required functionality.

`app/db/session.py`

import os
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

DATABASE_URL = os.environ["DATABASE_URL"]
DATABASE_ADMIN_URL = os.environ.get("DATABASE_ADMIN_URL", DATABASE_URL)

# The app's engine — a role with RLS applied
engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    connect_args={
        "statement_cache_size": 0,  # CRITICAL: prevents prepared statement leaks
    },
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# The admin engine — only for migrations and cross-tenant jobs
admin_engine = create_async_engine(
    DATABASE_ADMIN_URL,
    echo=False,
    connect_args={"statement_cache_size": 0},
)
AdminSessionLocal = async_sessionmaker(admin_engine, expire_on_commit=False)

`app/db/models.py`

from datetime import datetime
from sqlalchemy import (
    BigInteger, ForeignKey, String, DateTime, Integer, Index, func, text,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Tenant(Base):
    __tablename__ = "tenants"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    slug: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    tenant_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey("tenants.id", ondelete="RESTRICT"),
        nullable=False,
    )
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    status: Mapped[str] = mapped_column(String(20), nullable=False, default="open")
    priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True,
    )

    __table_args__ = (
        Index(
            "ix_tasks_tenant_created_active",
            "tenant_id", text("created_at DESC"), text("id DESC"),
            postgresql_where=text("deleted_at IS NULL"),
        ),
        Index(
            "ix_tasks_tenant_status_active",
            "tenant_id", "status",
            postgresql_where=text("deleted_at IS NULL"),
        ),
    )

`app/db/tenant_context.py`

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

from app.db.session import SessionLocal


async def get_current_tenant_id(request: Request) -> int:
    tenant_id_str = request.headers.get("X-Tenant-ID")
    if not tenant_id_str:
        raise HTTPException(status_code=401, detail="Missing X-Tenant-ID header")
    try:
        return int(tenant_id_str)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid X-Tenant-ID header")


async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant_id),
) -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        async with session.begin():
            # SET LOCAL only lasts the current transaction.
            # We validate that tenant_id is an int (get_current_tenant_id does it).
            await session.execute(
                text(f"SET LOCAL app.tenant_id = '{int(tenant_id)}'")
            )
            yield session

`app/api/tasks.py` (a skeleton — you have to complete it)

import base64
import json
from datetime import datetime
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy import select, and_, tuple_
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.tenant_context import get_tenant_session
from app.db.models import Task

router = APIRouter()


class TaskCreate(BaseModel):
    # NOTE: do NOT include tenant_id here. Anti-pattern 5.
    title: str = Field(..., max_length=200)
    status: str = Field(default="open", pattern="^(open|in_progress|done)$")
    priority: int = Field(default=0, ge=0, le=10)


class TaskOut(BaseModel):
    id: int
    title: str
    status: str
    priority: int
    created_at: datetime

    class Config:
        from_attributes = True


class TaskListResponse(BaseModel):
    items: list[TaskOut]
    next_cursor: str | None


def encode_cursor(created_at: datetime, task_id: int) -> str:
    payload = {"c": created_at.isoformat(), "i": task_id}
    return base64.urlsafe_b64encode(
        json.dumps(payload).encode()
    ).decode()


def decode_cursor(cursor: str) -> tuple[datetime, int]:
    try:
        payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
        return datetime.fromisoformat(payload["c"]), payload["i"]
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid cursor")


@router.post("/tasks", status_code=201, response_model=TaskOut)
async def create_task(
    payload: TaskCreate,
    db: AsyncSession = Depends(get_tenant_session),
):
    # tenant_id comes from the SET LOCAL — RLS applies it via WITH CHECK.
    # Your code has to insert the row WITHOUT specifying tenant_id explicitly
    # ... BUT the column is NOT NULL. The solution: read current_setting when inserting.
    # TODO: implement the correct insert.
    raise NotImplementedError("Implement create_task")


@router.get("/tasks", response_model=TaskListResponse)
async def list_tasks(
    cursor: str | None = None,
    limit: int = Query(default=20, ge=1, le=100),
    db: AsyncSession = Depends(get_tenant_session),
):
    stmt = (
        select(Task)
        .where(Task.deleted_at.is_(None))
        .order_by(Task.created_at.desc(), Task.id.desc())
        .limit(limit + 1)  # +1 to detect whether there's a next page
    )

    if cursor:
        cursor_created, cursor_id = decode_cursor(cursor)
        stmt = stmt.where(
            tuple_(Task.created_at, Task.id) < (cursor_created, cursor_id)
        )

    result = await db.execute(stmt)
    rows = list(result.scalars().all())

    has_next = len(rows) > limit
    items = rows[:limit]
    next_cursor = (
        encode_cursor(items[-1].created_at, items[-1].id) if has_next else None
    )

    return TaskListResponse(
        items=[TaskOut.model_validate(t) for t in items],
        next_cursor=next_cursor,
    )


@router.get("/tasks/{task_id}", response_model=TaskOut)
async def get_task(
    task_id: int,
    db: AsyncSession = Depends(get_tenant_session),
):
    # RLS filters by tenant — another tenant's task returns None here.
    task = await db.get(Task, task_id)
    if task is None or task.deleted_at is not None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.delete("/tasks/{task_id}", status_code=204)
async def delete_task(
    task_id: int,
    db: AsyncSession = Depends(get_tenant_session),
):
    # TODO: implement the soft delete (UPDATE deleted_at = NOW())
    raise NotImplementedError("Implement the soft delete")

`app/main.py`

from fastapi import FastAPI
from app.api import tasks

app = FastAPI(title="TaskFlow M4 — Multitenant Foundation")
app.include_router(tasks.router, prefix="/api")


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

To run it locally

export DATABASE_URL="postgresql+asyncpg://app_user:app_secure_password@localhost:5432/taskflow"
export DATABASE_ADMIN_URL="postgresql+asyncpg://postgres:postgres@localhost:5432/taskflow"

# 1. Run the migrations
alembic upgrade head

# 2. Seed the tenants
python scripts/seed.py

# 3. Bring up the API
uvicorn app.main:app --reload

# Try it manually with curl
curl -H "X-Tenant-ID: 1" http://localhost:8000/api/tasks
# Returns only Acme's tasks

curl -H "X-Tenant-ID: 2" http://localhost:8000/api/tasks
# Returns only Globex's tasks

What you have:

The project structure, the models, the tenant context dependency.
The GET /tasks (working cursor pagination) and GET /tasks/{id} endpoints complete.

What you have to implement:

The Alembic migration with RLS + the audit trigger.
POST /tasks inserting correctly while respecting the policy's WITH CHECK.
DELETE /tasks/{id} with a soft delete + verification that the audit log captures it.
The suite of "malicious" tests (the rubric section below).
MULTITENANCY.md documenting the architectural decision.
A cron job demonstrating anti-pattern 4 + its corrected version.

Evaluation rubric (self-check)

Total: 100 points. Pass: ≥70. Distinction ("EXEMPLARY"): ≥90.

Correct multi-tenant configuration (25 points)

(5 pts) An Alembic migration with ENABLE ROW LEVEL SECURITY + FORCE ROW LEVEL SECURITY on tasks.
(5 pts) A tenant_isolation policy with USING AND WITH CHECK. Verifiable with \d+ tasks in psql.
(5 pts) A non-owner app_user role with minimal permissions (SELECT/INSERT/UPDATE/DELETE on tasks, USAGE on sequences).
(5 pts) connect_args={"statement_cache_size": 0} configured in the async engine (verifiable by reviewing app/db/session.py).
(5 pts) The get_tenant_session dependency runs SET LOCAL inside an explicit session.begin().

"Malicious" isolation tests (30 points)

Each test has to run against two real tenants (Acme and Globex) in the same Postgres and verify the isolation is respected.

(5 pts) Anti-pattern 1 test: GET /tasks from tenant A doesn't return tenant B's tasks. Even if the base query had no tenant_id filter, RLS blocks it.
(5 pts) Anti-pattern 2 test (IDOR): GET /tasks/{id} with a valid tenant B ID from a tenant A session returns a 404 (not a 200, not a 403, not a 500). Indistinguishable.
(5 pts) Anti-pattern 5 test: an attempt at POST /tasks with a tenant_id in the body gets rejected or the field gets ignored (verify the task gets created in the header's tenant, not the body's).
(5 pts) A direct SQL test: a connection as app_user with SET LOCAL app.tenant_id = '1' runs SELECT * FROM tasks WHERE 1=1 and returns only tenant 1's tasks.
(5 pts) A cross-tenant INSERT test: a direct attempt at INSERT INTO tasks (tenant_id, ...) VALUES (999, ...) from a tenant 1 session fails with a policy violation.
(5 pts) A cross-tenant cursor test: a cursor generated by tenant A, used by tenant B, doesn't return tenant A's rows. The cursor is isolated by construction.

A working audit log (15 points)

(5 pts) The trg_task_audit trigger is active on tasks. Verifiable with \d+ tasks showing "Triggers".
(5 pts) An INSERT on tasks from the endpoint produces a row in audit.task_history with the correct tenant_id from the session setting.
(5 pts) An UPDATE (including a soft delete) produces a row in the audit with old_data and new_data populated.

Connection with the previous modules (15 points)

(5 pts) Cursor pagination (module 1): GET /tasks with ?cursor= works correctly, you page to the end, next_cursor: null on the last page.
(5 pts) Soft delete with a partial index (module 2): DELETE /tasks/{id} sets deleted_at, EXPLAIN shows an Index Scan over ix_tasks_tenant_created_active on GET /tasks.
(5 pts) Audit log (module 3): the trigger captures current_setting('app.tenant_id') correctly.

Documentation (10 points)

(5 pts) MULTITENANCY.md documents: the chosen model (shared schema + RLS), the quantitative criteria from the decision matrix (expected number of tenants, isolation requirement, operational cost), why schema-per-tenant was NOT chosen.
(5 pts) README.md with runnable setup instructions (Docker, migrations, seed, tests).

A safe cron job (5 points)

(3 pts) An implementation of the "weekly summary" cron job following the correct pattern: the admin role only to enumerate tenants, SET LOCAL app.tenant_id per iteration, isolated errors.
(2 pts) A test verifying the cron job doesn't send emails with crossed data between tenants.

Extra credit (optional, up to +10 pts)

(+3 pts) A documented bypass pattern for admin: an admin_app role with BYPASSRLS + a GET /admin/tasks/cross-tenant endpoint with separate auth that lists tasks from every tenant (with a documented justification).
(+3 pts) A documented benchmark: the latency of GET /tasks with and without RLS enabled. Expected: <15% overhead.
(+2 pts) A regression test verifying FORCE ROW LEVEL SECURITY is enabled (it fails if somebody disables it in a future migration).
(+2 pts) A visual decision tree (PNG or ASCII) in MULTITENANCY.md showing when to choose each model.

Example "malicious" tests

Below, the critical tests solved as a reference. The student has to complete and add the others from the rubric.

# tests/security/test_isolation.py
import pytest
from httpx import AsyncClient
from sqlalchemy import text

from app.main import app
from app.db.session import SessionLocal
from app.db.models import Tenant, Task


@pytest.fixture
async def setup_two_tenants(admin_db):
    """Creates two tenants with tasks in each one."""
    acme = Tenant(slug="acme-iso", name="Acme Isolation Test")
    globex = Tenant(slug="globex-iso", name="Globex Isolation Test")
    admin_db.add_all([acme, globex])
    await admin_db.flush()

    async with admin_db.begin_nested():
        await admin_db.execute(text("SET LOCAL row_security = off"))
        admin_db.add_all([
            Task(tenant_id=acme.id, title="Acme Task A1"),
            Task(tenant_id=acme.id, title="Acme Task A2"),
            Task(tenant_id=globex.id, title="Globex Task G1"),
        ])
    await admin_db.commit()
    return {"acme": acme, "globex": globex}


@pytest.mark.asyncio
async def test_list_tasks_isolated_per_tenant(setup_two_tenants):
    """Anti-pattern 1: GET /tasks must not cross tenants."""
    data = setup_two_tenants

    async with AsyncClient(app=app, base_url="http://test") as client:
        # Acme sees only its tasks
        r = await client.get(
            "/api/tasks",
            headers={"X-Tenant-ID": str(data["acme"].id)},
        )
        assert r.status_code == 200
        titles = [t["title"] for t in r.json()["items"]]
        assert "Acme Task A1" in titles
        assert "Globex Task G1" not in titles, (
            f"BREACH: Acme saw a Globex task. Titles: {titles}"
        )


@pytest.mark.asyncio
async def test_idor_returns_404(setup_two_tenants, admin_db):
    """Anti-pattern 2: requesting another tenant's task has to be a 404."""
    data = setup_two_tenants

    # Capture the ID of Globex's task
    result = await admin_db.execute(
        text("SET LOCAL row_security = off; "
             "SELECT id FROM tasks WHERE tenant_id = :tid LIMIT 1"),
        {"tid": data["globex"].id},
    )
    globex_task_id = result.scalar_one()

    async with AsyncClient(app=app, base_url="http://test") as client:
        r = await client.get(
            f"/api/tasks/{globex_task_id}",
            headers={"X-Tenant-ID": str(data["acme"].id)},
        )
        assert r.status_code == 404, (
            f"BREACH: Acme accessed Globex's task {globex_task_id}. "
            f"Status={r.status_code}, body={r.json()}"
        )


@pytest.mark.asyncio
async def test_insert_cross_tenant_blocked():
    """RLS WITH CHECK has to block an INSERT with someone else's tenant_id."""
    async with SessionLocal() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '1'"))

            # An attempt to insert with tenant_id = 999 (not the setting's)
            with pytest.raises(Exception) as exc_info:
                await session.execute(
                    text(
                        "INSERT INTO tasks (tenant_id, title, status) "
                        "VALUES (999, 'malicious', 'open')"
                    )
                )
            assert "row-level security policy" in str(exc_info.value).lower(), (
                f"Expected a policy violation, got: {exc_info.value}"
            )


@pytest.mark.asyncio
async def test_select_with_malicious_where_filtered_by_rls():
    """A query with WHERE 1=1 from a tenant 1 session returns only tenant 1's tasks."""
    async with SessionLocal() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '1'"))

            result = await session.execute(
                text("SELECT DISTINCT tenant_id FROM tasks WHERE 1=1")
            )
            tenant_ids = {row[0] for row in result}
            assert tenant_ids <= {1}, (
                f"BREACH: a query with WHERE 1=1 returned tasks from tenants: {tenant_ids}"
            )

Common mistakes in this project

Mistake 1: the app connects as `postgres` (the owner) and RLS doesn't apply

Symptom: the "malicious" tests pass locally but a manual curl with X-Tenant-ID: 1 returns tasks from all three tenants. Total confusion.

Why it happens: the DATABASE_URL points at the postgres user from copy-pasting a tutorial. PostgreSQL doesn't apply RLS to the owner (unless FORCE ROW LEVEL SECURITY is enabled).

How to tell: run SHOW USER; in a psql session connected with your .env's DATABASE_URL. If it says postgres (or the role that owns tasks), that's the bug.

How to fix it: create the separate app_user role, connect the app with that role. Turn on FORCE ROW LEVEL SECURITY as defense in depth in case somebody uses the wrong role in the future.

Mistake 2: `SET LOCAL` run outside an explicit transaction

Symptom: the endpoint's queries work fine locally but return 0 rows or blow up in production with PgBouncer.

Why it happens: SET LOCAL only applies to the current transaction. If the dependency sets the context but the endpoint operates in autocommit, the setting doesn't apply to the endpoint's queries.

How to tell: review app/db/tenant_context.py. Is the yield session INSIDE an async with session.begin():? If it's outside, that's the bug.

How to fix it: nest the yield inside session.begin(). The endpoint's whole operation stays in an explicit transaction.

Mistake 3: forgetting `WITH CHECK` in the policy

Symptom: the SELECT tests pass, but a cross-tenant INSERT test also passes (when it should fail).

Why it happens: USING only applies to SELECT/UPDATE/DELETE. Without WITH CHECK, the INSERTs don't get validated — inserting rows with any tenant_id is allowed.

How to tell: \d+ tasks in psql. Look for the "Policies" section. It has to say something like tenant_isolation FOR ALL USING (...) WITH CHECK (...).

How to fix it: alter the policy to include WITH CHECK. Capsule 04 shows the exact syntax.

Mistake 4: a pagination cursor that doesn't include `tenant_id`

Symptom: confusion about whether the cursor "leaks" cross-tenant information.

Why it happens: the dev worries that the cursor (which contains a task's id) could be used by another tenant to access that task.

How to tell: the rubric's "cross-tenant cursor" test. If it passes (no leak), the model is correct. If it fails, there's a bug.

How to fix it: the cursor on its own is just a (created_at, id) pair. RLS filters when the query runs. Tenant B using A's cursor only sees B's rows with (created_at, id) < cursor. It isn't a leak. Document this reasoning in MULTITENANCY.md so a reviewer understands it in code review.

Mistake 5: the audit log trigger doesn't capture `tenant_id`

Symptom: rows in audit.task_history show up with tenant_id = NULL or all with the same wrong value.

Why it happens: the trigger uses current_setting('app.tenant_id') without handling the "not set" case. Or the seed runs without setting the context and the rows end up with no tenant.

How to tell: SELECT tenant_id, COUNT(*) FROM audit.task_history GROUP BY 1; should show three different tenant_ids with coherent counts.

How to fix it: wrap current_setting in a BEGIN/EXCEPTION to fall back to the row's tenant_id (COALESCE(NEW.tenant_id, OLD.tenant_id)). This covers the admin seed case that sets row_security = off.

Mistake 6: the tests pass on local SQLite

Symptom: the team configures the tests with SQLite "so they're fast." The tests pass. But RLS is a PostgreSQL feature — SQLite has no policies.

Why it happens: copy-pasting a common pattern from non-multi-tenant projects where SQLite is a good substitute.

How to tell: review tests/conftest.py. If there's a sqlite:/// in some engine, that's the bug.

How to fix it: the isolation tests MUST run against PostgreSQL (ideally the same Docker as production). Use pytest-postgresql or a dedicated test container. The tests' speed doesn't justify losing the guarantee.

What to do if you get stuck

The setup doesn't work: verify PostgreSQL is up (docker ps), that the app_user role exists (\du in psql as postgres), that the taskflow DB exists.
The Alembic migration fails: review capsule 04 for the exact syntax of ENABLE/FORCE/CREATE POLICY. Make sure Alembic's op.execute() isn't inside a with op.batch_alter_table(...).
The "malicious" tests fail unexpectedly: verify with \d+ tasks in psql that the policies are enabled. Verify the role with SHOW USER;.
SET LOCAL doesn't work: check that the dependency does async with session.begin() and that the yield is inside.
The audit log doesn't fill up: verify the trigger got created with \df audit.log_task_change. Verify the audit schema's permissions (the trigger has to be able to INSERT).
Cursor pagination returns duplicate rows: the ordering has to be deterministic — include id as the second criterion when created_at can have ties.

Resources for the project

PostgreSQL 16 — Row Security Policies — the official reference, especially the FORCE ROW LEVEL SECURITY section and its interaction with roles.
Supabase — Row Level Security — a pragmatic guide from the RLS provider most used in modern SaaS.
SQLAlchemy 2.0 — Async ORM — the official reference for the async ORM used.
FastAPI — Dependencies — the pattern used in get_tenant_session.
Alembic — Operation Reference — for writing the migration with op.execute() to create policies and triggers.
Crunchy Data — Row Level Security for Tenants — a practical guide with an emphasis on FORCE ROW LEVEL SECURITY and role design.
pganalyze — Best practices for Postgres RLS — a technical analysis with performance benchmarks for policies.
Brandur — Postgres-only stacks at scale — a perspective on defense in depth and each layer's limits.

What comes next

Module 4's TaskFlow leaves you with the working multi-tenant foundation: a shared schema with RLS, an audit log that captures the tenant, cursor pagination respecting the isolation, soft deletes with partial indexes, "malicious" tests that verify the isolation against the documented anti-patterns.

Module 5 closes the "Live Operations" block with zero-downtime migrations. You're going to learn the expand-contract pattern with real Alembic: how to add a NOT NULL column to tasks (for example, priority with a default) with no downtime, in 3 deploys (nullable → backfill → NOT NULL). In multi-tenant, this is even more critical — a downtime affects all three tenants simultaneously. The conceptual transition is direct: you already know how to isolate tenants; now you learn to evolve the schema without taking down the DB the three tenants share.

After module 5 come optimistic locking (module 6), bulk operations (module 7), and module 8's capstone project: the complete TaskFlow, where you extend what you built here with all the remaining features — the missing endpoints, optimistic locking on PUT /tasks/{id}, a bulk import with COPY, a zero-downtime migration run live with wrk running in the background measuring that no request failed.

Before moving on to module 5 you should have:

Module 4's TaskFlow running locally with alembic upgrade head + python scripts/seed.py + uvicorn.
The 6 "malicious" tests passing (at least the 5 required by the rubric).
MULTITENANCY.md written with quantitative criteria.
A manual run with curl demonstrating: X-Tenant-ID: 1 sees only Acme's tasks, X-Tenant-ID: 2 sees only Globex's tasks.
An audit log with rows correctly attributed to each tenant.

When you have that, you've closed module 4 with real mastery of multi-tenant isolation. Module 5 is going to add the ability to evolve that system without interrupting the service.

Module 4 — SQL Patterns for Production APIs Guide

Module 4 close. Next module: Zero-Downtime Migrations — how you evolve the schema of the DB the three tenants share without a single request failing.

← PreviousCross-tenant leak: anti-patterns Next →Module 5: Zero-Downtime Migrations

Project: multi-tenant TaskFlow with RLS

What are you going to build and why?

Project objective

How it fits with what you learned

Technical specifications

Stack

Initial setup

Local PostgreSQL with Docker

Environment variables

Required functionality

1. The multi-tenant data model

2. The initial Alembic migration

3. The POST /tasks endpoint

4. The GET /tasks endpoint with cursor pagination

5. The GET /tasks/{id} and DELETE /tasks/{id} endpoints (soft delete)

6. The audit log with a PostgreSQL trigger

7. The seed of three tenants and initial data

Validation and error handling

What has to be validated

Errors that have to be handled

Minimal implementation example

app/db/session.py

app/db/models.py

app/db/tenant_context.py

app/api/tasks.py (a skeleton — you have to complete it)

app/main.py

To run it locally

Evaluation rubric (self-check)

Correct multi-tenant configuration (25 points)

"Malicious" isolation tests (30 points)

A working audit log (15 points)

Connection with the previous modules (15 points)

Documentation (10 points)

A safe cron job (5 points)

Extra credit (optional, up to +10 pts)

Example "malicious" tests

Common mistakes in this project

Mistake 1: the app connects as postgres (the owner) and RLS doesn't apply

Mistake 2: SET LOCAL run outside an explicit transaction

Mistake 3: forgetting WITH CHECK in the policy

Mistake 4: a pagination cursor that doesn't include tenant_id

Mistake 5: the audit log trigger doesn't capture tenant_id

Mistake 6: the tests pass on local SQLite

What to do if you get stuck

Resources for the project

What comes next

3. The `POST /tasks` endpoint

4. The `GET /tasks` endpoint with cursor pagination

5. The `GET /tasks/{id}` and `DELETE /tasks/{id}` endpoints (soft delete)

`app/db/session.py`

`app/db/models.py`

`app/db/tenant_context.py`

`app/api/tasks.py` (a skeleton — you have to complete it)

`app/main.py`

Mistake 1: the app connects as `postgres` (the owner) and RLS doesn't apply

Mistake 2: `SET LOCAL` run outside an explicit transaction

Mistake 3: forgetting `WITH CHECK` in the policy

Mistake 4: a pagination cursor that doesn't include `tenant_id`

Mistake 5: the audit log trigger doesn't capture `tenant_id`