Shared schema with `tenant_id` and its pitfalls

Capsule overview

You already know the three multi-tenancy models and you know that for typical B2B SaaS the recommended default is a shared schema with RLS. But before learning RLS you need to experience firsthand the model RLS comes to complement: a shared schema with tenant_id and no DB-level protection. That's the most-used model in real production and also the one that has caused the most cross-tenant leaks in SaaS history.

This capsule teaches you to implement a shared schema correctly: how to model tenant_id in each table, how to index it so queries don't degrade at scale, how to encapsulate the filtering to reduce the probability of forgetting a WHERE, and what mitigation patterns exist (lint rules, base classes in SQLAlchemy, a code review checklist). You're going to implement a multi-tenant tasks model in SQLAlchemy 2.0 async + FastAPI 0.110+ and try it with two different tenants against PostgreSQL 16+.

By the end you'll be clear on why the "manual discipline in queries" approach has a natural ceiling — the formula is simple: a single forgotten WHERE tenant_id = a breach. That clarity is what motivates capsules 04 and 05, where you'll learn how PostgreSQL can apply the filter automatically with RLS so a bug in the code doesn't become a public incident.

Mental model: the ceiling of manual discipline

When you work with a shared schema with no DB-level protection, the isolation between tenants depends 100% on each individual query including WHERE tenant_id = .... Think of it like an office building where each client company has its floor, but the elevator doors have no key card — just a sign that says "please use your floor." As long as every employee respects the sign, the building works. The first employee who gets the wrong floor sees another company's documents.

In code, that "sign" is the dev's discipline. That "employee who makes a mistake" is the PR accepted in code review with a new query that forgot the filter. And unlike a physical building, where the mistake is visible and instantly correctable, in an API the mistake can go unnoticed for months until a customer reports "I'm seeing strange names in my dashboard."

The key question for this model isn't "how do we make sure nobody ever forgets?". It's: "how do we reduce the probability that someone forgets and, when they do, how do we limit the damage?". The answers are: encapsulation (queries that filter by default), tooling (lint rules that detect the missing pattern), code review (an explicit checklist), and eventually migrating to RLS when those mitigations are no longer enough.

Modeling tenant_id in SQLAlchemy 2.0

The first step is adding tenant_id to every table that has data belonging to a tenant. Global tables (shared catalogs like countries, currencies, subscription plans) do NOT carry tenant_id.

# app/db/models.py
from datetime import datetime
from sqlalchemy import BigInteger, ForeignKey, String, DateTime, Index, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


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")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )

    # CRITICAL: a composite index that starts with tenant_id.
    # Almost every query in the module is going to filter by tenant_id first.
    __table_args__ = (
        Index("ix_tasks_tenant_created", "tenant_id", "created_at"),
        Index("ix_tasks_tenant_status", "tenant_id", "status"),
    )

Three model decisions that matter:

1. tenant_id BIGINT NOT NULL, not nullable. A row with no tenant is an orphaned row, forbidden.
2. A ForeignKey with ondelete="RESTRICT" prevents deleting a tenant that still has associated rows. To "delete" a tenant there has to be an explicit process that cleans up its data first.
3. Composite indexes that start with tenant_id. Without these, queries that filter by tenant + another column scan the whole table. With these, PostgreSQL navigates straight to the tenant's rows.

The initial migration with Alembic

# Generate the migration from the model
alembic revision --autogenerate -m "add tenants and tasks tables"
alembic upgrade head

The generated migration includes the CREATE INDEXes automatically because they're in __table_args__.

Query filtering: the part where teams fail

You already have the schema. Now every query against tasks has to filter by tenant_id. The "obvious" way is to add it in every place:

# app/api/tasks.py — the obvious version, fragile
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_session
from app.db.models import Task, Tenant
from app.auth import get_current_tenant

router = APIRouter()


@router.get("/tasks")
async def list_tasks(
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    result = await db.execute(
        select(Task)
        .where(Task.tenant_id == current_tenant.id)  # CRITICAL
        .order_by(Task.created_at.desc())
        .limit(50)
    )
    return result.scalars().all()


@router.get("/tasks/{task_id}")
async def get_task(
    task_id: int,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    result = await db.execute(
        select(Task)
        .where(Task.id == task_id)
        .where(Task.tenant_id == current_tenant.id)  # CRITICAL here too
    )
    task = result.scalar_one_or_none()
    if task is None:
        # Without tenant_id in the WHERE, this would return tasks from OTHER tenants
        return {"error": "not found"}
    return task

The problem with this form is what you already saw in capsule 01: a single new query that forgets that .where(Task.tenant_id == ...) opens a leak. And as the team grows, the probability of someone forgetting it goes up.

Mitigation pattern 1: encapsulate in a repository class

Instead of raw queries scattered across endpoints, you encapsulate the access in a repository that always receives tenant_id:

# app/db/repositories/task_repo.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import Task


class TaskRepository:
    def __init__(self, db: AsyncSession, tenant_id: int):
        self.db = db
        self.tenant_id = tenant_id  # always filtered by this tenant

    async def list(self, limit: int = 50) -> list[Task]:
        result = await self.db.execute(
            select(Task)
            .where(Task.tenant_id == self.tenant_id)
            .order_by(Task.created_at.desc())
            .limit(limit)
        )
        return list(result.scalars().all())

    async def get(self, task_id: int) -> Task | None:
        result = await self.db.execute(
            select(Task)
            .where(Task.id == task_id)
            .where(Task.tenant_id == self.tenant_id)
        )
        return result.scalar_one_or_none()

    async def create(self, title: str, status: str = "open") -> Task:
        task = Task(tenant_id=self.tenant_id, title=title, status=status)
        self.db.add(task)
        await self.db.flush()
        return task

# app/api/tasks.py — using the repository
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_session
from app.db.repositories.task_repo import TaskRepository
from app.db.models import Tenant
from app.auth import get_current_tenant

router = APIRouter()


def get_task_repo(
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
) -> TaskRepository:
    return TaskRepository(db, tenant_id=current_tenant.id)


@router.get("/tasks")
async def list_tasks(repo: TaskRepository = Depends(get_task_repo)):
    return await repo.list()


@router.get("/tasks/{task_id}")
async def get_task(task_id: int, repo: TaskRepository = Depends(get_task_repo)):
    task = await repo.get(task_id)
    if task is None:
        return {"error": "not found"}
    return task

What you gain: the only way to skip the filter is to write raw SQL instead of using the repository. That's detectable in code review (everybody knows Task is accessed only through TaskRepository).

What you do NOT gain: real security. A new dev can write select(Task) directly in an endpoint and the app allows it. The repository is a convention, not an obligation.

Mitigation pattern 2: forbid direct queries with a lint rule

You can write a custom lint rule (with ruff or flake8) that detects select(Task) outside the repository's file and fails CI. It's feasible but it requires keeping the rule updated every time you add a model.

A more pragmatic alternative: add a CI check with a simple grep that fails if it finds select(Task outside the allowed files:

# scripts/check_no_direct_task_queries.sh
#!/bin/bash
set -e

# Look for select(Task) or select(Project), etc. outside the repositories folder
violations=$(grep -rn "select(\(Task\|Project\|Note\)" \
    --include="*.py" \
    --exclude-dir=repositories \
    app/ tests/ || true)

if [ -n "$violations" ]; then
    echo "Direct queries to multitenant models detected outside repositories:"
    echo "$violations"
    exit 1
fi

You add it to the CI pipeline. It fails the build if someone writes a direct query.

Mitigation pattern 3: a mixin that requires filtering

Some teams use a SQLAlchemy mixin that overrides the default query. It's seductive but fragile — query mixins require extra discipline and many cases break the abstraction. Capsule 04 shows why RLS is the right solution to this problem and not a mixin.

Worked example: two tenants in the same DB

Let's try the whole setup with two concrete tenants (Acme and Globex) and verify the filtering works — and also what happens when the filter gets forgotten.

Setup

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

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

engine = create_async_engine(DATABASE_URL, echo=False)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncSession:
    async with SessionLocal() as session:
        yield session

# scripts/seed.py — creates two tenants and tasks for each one
import asyncio
from app.db.session import SessionLocal
from app.db.models import Tenant, Task


async def seed():
    async with SessionLocal() as db:
        # Tenants
        acme = Tenant(slug="acme", name="Acme Corp")
        globex = Tenant(slug="globex", name="Globex")
        db.add_all([acme, globex])
        await db.flush()

        # Acme's tasks
        db.add_all([
            Task(tenant_id=acme.id, title="Acme task 1"),
            Task(tenant_id=acme.id, title="Acme task 2"),
        ])

        # Globex's tasks
        db.add_all([
            Task(tenant_id=globex.id, title="Globex task 1"),
            Task(tenant_id=globex.id, title="Globex task 2"),
            Task(tenant_id=globex.id, title="Globex task 3"),
        ])

        await db.commit()
        print(f"Seeded: Acme id={acme.id}, Globex id={globex.id}")


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

Run it:

python scripts/seed.py
# Output: Seeded: Acme id=1, Globex id=2

The correct query vs the buggy query

# scripts/demo_filtering.py
import asyncio
from sqlalchemy import select
from app.db.session import SessionLocal
from app.db.models import Task
from app.db.repositories.task_repo import TaskRepository


async def demo():
    async with SessionLocal() as db:
        # CASE 1: using the repository — correct filtering
        acme_repo = TaskRepository(db, tenant_id=1)
        acme_tasks = await acme_repo.list()
        print(f"\nAcme via repository: {len(acme_tasks)} tasks")
        for t in acme_tasks:
            print(f"  - {t.title}")

        # CASE 2: a direct query forgetting the filter — BUG
        result = await db.execute(select(Task).order_by(Task.created_at))
        all_tasks = result.scalars().all()
        print(f"\nQuery with no filter (BUG): {len(all_tasks)} tasks")
        for t in all_tasks:
            print(f"  - tenant_id={t.tenant_id}: {t.title}")


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

Run it:

python scripts/demo_filtering.py

Expected output:

Acme via repository: 2 tasks
  - Acme task 1
  - Acme task 2

Query with no filter (BUG): 5 tasks
  - tenant_id=1: Acme task 1
  - tenant_id=1: Acme task 2
  - tenant_id=2: Globex task 1
  - tenant_id=2: Globex task 2
  - tenant_id=2: Globex task 3

This is exactly the cross-tenant leak. The "buggy" query brought back tasks from both tenants. If that query were behind the GET /tasks endpoint that Acme's user called, Acme would have just seen Globex's tasks in their dashboard.

PostgreSQL has no way of knowing that query was "wrong." To PostgreSQL, that query is valid and efficient. The filtering depends 100% on the application code. That's the model's natural ceiling.

Correct indexing: the detail that kills performance

A common trap in a shared schema is not indexing correctly for multi-tenant queries. The rule:

Almost every index has to start with tenant_id.

Reason: PostgreSQL navigates B-tree indexes from the leftmost column. If the index is (created_at) and your query is WHERE tenant_id = 1 ORDER BY created_at, PostgreSQL can use the index for the ORDER BY but it has to scan rows from every tenant before filtering. If the index is (tenant_id, created_at), PostgreSQL navigates straight to tenant 1's rows already ordered by created_at.

-- ❌ WRONG: an index on created_at only
CREATE INDEX idx_tasks_created ON tasks (created_at);

-- ✅ RIGHT: a composite index starting with tenant_id
CREATE INDEX idx_tasks_tenant_created ON tasks (tenant_id, created_at DESC);

-- ✅ RIGHT: for queries by status within a tenant
CREATE INDEX idx_tasks_tenant_status ON tasks (tenant_id, status);

An exception: globally unique indexes (e.g. email in a users table if emails are unique across the whole platform) don't lead with tenant_id. But those cases are rare — it's more common for the unique to be per tenant: UNIQUE (tenant_id, email).

Verification with EXPLAIN

EXPLAIN ANALYZE
SELECT * FROM tasks
WHERE tenant_id = 1
ORDER BY created_at DESC
LIMIT 50;

With the right index you'll see something like:

Limit  (cost=0.42..8.45 rows=50 width=...)
  ->  Index Scan using idx_tasks_tenant_created on tasks
      Index Cond: (tenant_id = 1)

Without the index you'll see Seq Scan on tasks (a full table scan). At 100k rows that takes ~30ms. At 5M rows it takes 1.5s. At 50M it takes 15s. Performance degrades linearly with the table's size, not with the tenant's size.

Why does this matter in real work?

1. It's the most-used model in the industry. GitHub, Slack, Linear, most B2B and consumer SaaS started with a shared schema and tenant_id. Knowing this model deeply is the foundation.

2. It's where the most bugs reach production. Almost every publicly documented cross-tenant leak (GitHub Octopus 2021, several smaller incidents at scaleups) started with a lapse in a query. If you understand the model, you anticipate the bug.

3. The mitigation patterns you learn here also get used with RLS. The repository pattern, the custom lint, the code review checklist — all of that stays useful even when you add RLS. RLS is the final safety net, but the earlier layers still count.

4. The decision to migrate to RLS depends on understanding this model's limits. If you never implemented a shared schema without RLS, you can't argue "RLS solves this concrete problem." Capsule 04 is going to build on the fact that you've already seen it.

Traps and common mistakes

Mistake 1 (conceptual): thinking "filtering manually" is safe if "the team is disciplined"

Symptom: a team of 4 senior devs that trusts its discipline. "We've gone 2 years without a single filtering bug, we don't need RLS." A new dev arrives, opens a PR, and the bug gets in.

Why it happens: discipline isn't transmitted by osmosis. Each new dev has to learn it. The bigger the team, the more likely someone skips it. It's a problem of accumulated probability, not of individual capability.

How to tell: ask "if we hire 5 new devs in a month, what's the probability that none of them forgets the filter in a new query in their first quarter?". If the honest answer is "not very high," the model no longer scales.

How to fix it: combine discipline with tooling (lint in CI) and, when the team grows, add RLS as the safety net. Capsules 04 and 05 cover how.

Mistake 2 (technical): indexes without tenant_id first

Symptom: queries that filter by tenant + another column degrade exponentially with the table's size. EXPLAIN ANALYZE shows a Seq Scan instead of an Index Scan.

Why it happens: people follow the reflex "index by created_at because I order by created_at" without thinking that the query filters by tenant_id first. PostgreSQL can't use an index on created_at to filter by tenant_id first.

How to tell: run EXPLAIN ANALYZE on your app's common queries. If you see a Seq Scan, the composite index is missing.

How to fix it: create composite indexes that start with tenant_id. Replace the "single column" indexes with composite ones when possible (an index on (tenant_id, X) also serves queries that filter only by tenant_id, so you don't need an extra index on tenant_id alone).

Mistake 3 (operational): tenant_id as INT instead of BIGINT

Symptom: after 2 years the tenant_id reaches 2 billion (an INT overflow) and the app starts failing with integer out of range. Migrating INT to BIGINT on a table with billions of rows is operationally expensive (it requires rewriting the whole table).

Why it happens: people underestimate how much tenant_id grows. In a consumer SaaS, every signup creates an entry — 2 billion signups seems impossible until it happens.

How to tell: review the tenant_id definition in your schema. If it's INTEGER, consider migrating to BIGINT as soon as possible (while the table is still small).

How to fix it: from day one, use BIGINT for tenant_id (and for PKs in general). The extra space is negligible (8 bytes vs 4) compared with the cost of migrating later.

Mistake 4 (conceptual): assuming a JOIN with tenants protects automatically

Symptom: someone argues "if I do JOIN tasks ON tasks.tenant_id = current_user.tenant_id, I'm already filtering." But the JOIN requires the code to build the condition correctly — if the code doesn't include it, the JOIN doesn't appear.

Why it happens: it confuses the final SQL with the code that generates it. The SQL can have filtering, but if the ORM/query doesn't add it, it doesn't appear.

How to tell: look at the SQL SQLAlchemy generates with echo=True. If the query doesn't mention tenant_id in a WHERE or JOIN ON, it isn't filtered.

How to fix it: encapsulation in a repository prevents this. Direct queries are where the bugs show up.

Mistake 5 (operational): using CASCADE on tenant_id

Symptom: someone defines ForeignKey("tenants.id", ondelete="CASCADE") thinking "if I delete the tenant, everything gets deleted." Later a cron job accidentally deletes a tenant and deletes millions of tasks. Recovery requires a restore from backup.

Why it happens: "deleting a tenant" is usually an explicit, monitored process, not a casual operation. CASCADE turns an accidental deletion into a silent catastrophe.

How to tell: review all your FKs to tenants.id. If they're CASCADE, that's an alarm signal.

How to fix it: use ondelete="RESTRICT" (the default is restrict in PostgreSQL if you don't specify). If you want to delete a tenant, write the explicit script that first cleans up the data and then deletes the tenant.

Exercises

Exercise 1: detect queries with a filtering bug in a code review

This PR comes to you. Review it and list every query that has a tenant filtering bug.

# app/api/projects.py
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import Project, Tenant
from app.db.session import get_session
from app.auth import get_current_tenant

router = APIRouter()


@router.get("/projects")
async def list_projects(
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    result = await db.execute(
        select(Project)
        .where(Project.tenant_id == current_tenant.id)
        .order_by(Project.created_at.desc())
    )
    return result.scalars().all()


@router.get("/projects/{project_id}")
async def get_project(
    project_id: int,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    result = await db.execute(select(Project).where(Project.id == project_id))
    return result.scalar_one_or_none()


@router.get("/projects/search")
async def search_projects(
    q: str,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    result = await db.execute(
        select(Project).where(Project.name.ilike(f"%{q}%"))
    )
    return result.scalars().all()


@router.delete("/projects/{project_id}")
async def delete_project(
    project_id: int,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    project = await db.get(Project, project_id)
    if project:
        await db.delete(project)
        await db.commit()
    return {"deleted": True}

# get_project
result = await db.execute(
    select(Project)
    .where(Project.id == project_id)
    .where(Project.tenant_id == current_tenant.id)
)

# search_projects
result = await db.execute(
    select(Project)
    .where(Project.tenant_id == current_tenant.id)
    .where(Project.name.ilike(f"%{q}%"))
)

# delete_project
result = await db.execute(
    select(Project)
    .where(Project.id == project_id)
    .where(Project.tenant_id == current_tenant.id)
)
project = result.scalar_one_or_none()
if project:
    await db.delete(project)
    await db.commit()

Exercise 2: refactor Exercise 1's code into the repository pattern

Take the delete_project endpoint from Exercise 1 (with its bug) and refactor the code to use a ProjectRepository that receives tenant_id in its constructor. Then write the endpoint using the repo.

# app/db/repositories/project_repo.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import Project


class ProjectRepository:
    def __init__(self, db: AsyncSession, tenant_id: int):
        self.db = db
        self.tenant_id = tenant_id

    async def get(self, project_id: int) -> Project | None:
        result = await self.db.execute(
            select(Project)
            .where(Project.id == project_id)
            .where(Project.tenant_id == self.tenant_id)
        )
        return result.scalar_one_or_none()

    async def delete(self, project_id: int) -> bool:
        project = await self.get(project_id)
        if project is None:
            return False
        await self.db.delete(project)
        await self.db.commit()
        return True

# app/api/projects.py
from fastapi import APIRouter, Depends, HTTPException

from app.db.repositories.project_repo import ProjectRepository
from app.db.models import Tenant
from app.db.session import get_session
from app.auth import get_current_tenant
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()


def get_project_repo(
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
) -> ProjectRepository:
    return ProjectRepository(db, tenant_id=current_tenant.id)


@router.delete("/projects/{project_id}")
async def delete_project(
    project_id: int,
    repo: ProjectRepository = Depends(get_project_repo),
):
    deleted = await repo.delete(project_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Project not found")
    return {"deleted": True}

Exercise 3: identify the right index for a given query

For each of these queries (all on a multi-tenant tasks table with millions of rows), identify the optimal composite index.

a) SELECT * FROM tasks WHERE tenant_id = $1 AND status = 'open' b) SELECT * FROM tasks WHERE tenant_id = $1 ORDER BY created_at DESC LIMIT 20 c) SELECT * FROM tasks WHERE tenant_id = $1 AND assigned_to = $2 ORDER BY due_date ASC d) SELECT COUNT(*) FROM tasks WHERE tenant_id = $1 AND status = 'open' AND created_at > NOW() - INTERVAL '7 days'

Exercise 4: write a test that catches a cross-tenant leak

Write an async pytest test that:

1. Creates two tenants (Acme and Globex) with tasks in each one.
2. Uses the TaskRepository with tenant_id=acme.id.
3. Verifies that repo.list() returns only Acme's tasks.
4. Verifies that repo.get(globex_task_id) returns None (it doesn't expose Globex's tasks even if you know the ID).

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

from app.db.models import Tenant, Task
from app.db.repositories.task_repo import TaskRepository


@pytest.mark.asyncio
async def test_repository_filters_by_tenant(db: AsyncSession):
    # Setup: two tenants with tasks
    acme = Tenant(slug="acme-test", name="Acme Test")
    globex = Tenant(slug="globex-test", name="Globex Test")
    db.add_all([acme, globex])
    await db.flush()

    acme_task = Task(tenant_id=acme.id, title="Acme task A")
    globex_task = Task(tenant_id=globex.id, title="Globex task G")
    db.add_all([acme_task, globex_task])
    await db.flush()

    # Acme's repository
    acme_repo = TaskRepository(db, tenant_id=acme.id)

    # 1. list() returns only Acme's tasks
    acme_tasks = await acme_repo.list()
    assert len(acme_tasks) == 1
    assert acme_tasks[0].title == "Acme task A"

    # 2. get(another_tenants_id) returns None even though the ID exists
    leaked = await acme_repo.get(globex_task.id)
    assert leaked is None, (
        f"BREACH: Acme repository returned task {globex_task.id} from Globex"
    )

    # 3. get(own_id) does return
    own = await acme_repo.get(acme_task.id)
    assert own is not None
    assert own.title == "Acme task A"

Exercise 5: argue to your lead why a shared schema without RLS no longer scales

Your team has been at it 6 months with 80 customers and a team of 4 Python devs. 3 new devs are coming next quarter. Your lead says "a shared schema with tenant_id has been working fine for us, let's keep going." Articulate 3 arguments for migrating to RLS before the team grows.

Exercise 6: spot the badly-placed index

A new dev proposed this migration. What's wrong with it? How would you fix it?

# alembic/versions/abc123_add_tasks_indexes.py
"""add tasks indexes"""
from alembic import op


def upgrade():
    op.create_index("ix_tasks_status", "tasks", ["status"])
    op.create_index("ix_tasks_created", "tasks", ["created_at"])
    op.create_index("ix_tasks_assigned", "tasks", ["assigned_to"])


def downgrade():
    op.drop_index("ix_tasks_assigned")
    op.drop_index("ix_tasks_created")
    op.drop_index("ix_tasks_status")

# alembic/versions/abc123_add_tasks_indexes.py
"""add tasks indexes"""
from alembic import op


def upgrade():
    # The main index: tenant + status (for status filters within the tenant)
    op.create_index(
        "ix_tasks_tenant_status",
        "tasks",
        ["tenant_id", "status"],
    )
    # A secondary index: tenant + created_at (for ordered listings)
    op.create_index(
        "ix_tasks_tenant_created",
        "tasks",
        ["tenant_id", "created_at"],
    )
    # An index for assignments: tenant + assigned_to
    op.create_index(
        "ix_tasks_tenant_assigned",
        "tasks",
        ["tenant_id", "assigned_to"],
    )


def downgrade():
    op.drop_index("ix_tasks_tenant_assigned")
    op.drop_index("ix_tasks_tenant_created")
    op.drop_index("ix_tasks_tenant_status")

Summary and next step

In this capsule you learned:

• A shared schema with tenant_id is the simplest and most-used model: a single DB, a single table per entity, and a tenant_id column to identify each row's owner.
• The filtering depends 100% on the code's discipline. Every query has to include WHERE tenant_id = ... or the bug becomes a cross-tenant leak.
• The mitigation patterns reduce the risk but don't eliminate it: the repository pattern (encapsulates the filtering), a custom lint (detects direct queries), a code review checklist (a human checks).
• Correct indexing requires composite indexes that start with tenant_id. Without this, common queries degrade exponentially with the table's size.
• The model has a natural ceiling: a single forgotten WHERE tenant_id = a breach. The accumulated probability goes up with the team's size and the number of queries.
• That's why this model needs a DB-level safety net when the team grows. That net is Row-Level Security (RLS).

Before moving on you should be able to:

• Implement a correct multi-tenant SQLAlchemy model with tenant_id and the appropriate composite indexes.
• Encapsulate queries in a repository that receives tenant_id in its constructor.
• Detect queries with a filtering bug in a code review.
• Argue why a shared schema without RLS doesn't scale with the team's size.
• Identify badly-placed indexes in a proposed migration.

Next capsule — Row-Level Security: fundamentals. You're going to learn the mechanism PostgreSQL offers to apply the tenant_id filter automatically, without depending on the code adding it. You're going to understand how RLS gets turned on, how policies get written, what happens with the owner role (FORCE ROW LEVEL SECURITY), and why we have to repeat the most important caveat: RLS here is used only for multi-tenancy, NOT for auth/RBAC (mixing them leads to debugging hell). It's the capsule that closes today's model's ceiling with a DB-level safety net.

Resources

1. PostgreSQL — CREATE INDEX reference — the syntax and options for indexes, especially composite ones.
2. Use The Index, Luke! — Concatenated Indexes — the classic explanation of why the column order in a composite index matters. Required reading.
3. Crunchy Data — Designing Your Postgres Database for Multi-Tenancy — practical shared schema patterns with tenant_id.
4. SQLAlchemy 2.0 — Async ORM tutorial — the official reference for the async ORM used in this capsule.
5. FastAPI — Dependencies with sub-dependencies — the pattern used in get_task_repo that connects auth + a DB session + the repository.
6. GitHub Engineering — How we designed our database for multi-tenancy — a reference for how large platforms implement tenant_id at scale (search the blog).
7. Brandur — Postgres-only stacks at scale — a perspective on PostgreSQL's practical limits in multi-tenant at high scale.

Module 4 — SQL Patterns for Production APIs Guide

Next capsule: Row-Level Security: fundamentals — the mechanism that closes manual discipline's ceiling with a database-level guarantee.