Shared schema con `tenant_id` y trampas

Descripción de la cápsula

Ya conoces los tres modelos de multi-tenancy y sabes que para B2B SaaS típico el default recomendado es shared schema con RLS. Pero antes de aprender RLS necesitas vivir en carne propia el modelo que RLS viene a complementar: shared schema con tenant_id sin protección a nivel DB. Ese es el modelo más usado en producción real y también el que más cross-tenant leaks ha causado en la historia del SaaS.

Esta cápsula te enseña a implementar shared schema correctamente: cómo modelar tenant_id en cada tabla, cómo indexar para que las queries no degraden a escala, cómo encapsular el filtrado para reducir la probabilidad de olvidar un WHERE, y qué patterns de mitigación existen (lint rules, base classes en SQLAlchemy, code review checklist). Vas a implementar un modelo de tasks multitenant en SQLAlchemy 2.0 async + FastAPI 0.110+ y vas a probarlo con dos tenants distintos contra PostgreSQL 16+.

Al terminar tendrás claro por qué el approach "disciplina manual en queries" tiene un techo natural — la fórmula es simple: un solo WHERE tenant_id olvidado = un breach. Esa claridad es la que motiva las cápsulas 04 y 05, donde aprenderás cómo PostgreSQL puede aplicar el filtro automáticamente con RLS para que el bug en código no se convierta en incidente público.

Modelo mental: el techo de la disciplina manual

Cuando trabajas con shared schema sin protección a nivel DB, el aislamiento entre tenants depende 100% de que cada query individual incluya WHERE tenant_id = .... Piénsalo como un edificio de oficinas donde cada empresa cliente tiene su piso, pero las puertas del ascensor no tienen tarjeta — solo un cartel que dice "por favor, usa tu piso". Mientras todos los empleados respeten el cartel, el edificio funciona. El primer empleado que se equivoque de piso ve documentos de otra empresa.

En código, ese "cartel" es la disciplina del dev. Ese "empleado que se equivoca" es el PR aceptado en code review con una query nueva que olvidó el filtro. Y a diferencia de un edificio físico, donde el error es visible y corregible al instante, en una API el error puede pasar desapercibido durante meses hasta que un cliente reporta "veo nombres extraños en mi dashboard".

La pregunta clave de este modelo no es "¿cómo lograr que nadie se olvide nunca?". Es: "¿cómo reducir la probabilidad de que alguien se olvide y, cuando se olvide, cómo limitar el daño?". Las respuestas son: encapsulación (queries que filtren por defecto), tooling (lint rules que detecten el patrón faltante), code review (checklist explícito) y eventualmente migración a RLS cuando estos mitigantes ya no alcancen.

Modelado de tenant_id en SQLAlchemy 2.0

El primer paso es agregar tenant_id a cada tabla que tenga datos pertenecientes a un tenant. Las tablas globales (catálogos compartidos como países, monedas, planes de subscripción) NO llevan 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
    )

    # CRÍTICO: índice compuesto que empieza por tenant_id.
    # Casi todas las queries del módulo van a filtrar por tenant_id primero.
    __table_args__ = (
        Index("ix_tasks_tenant_created", "tenant_id", "created_at"),
        Index("ix_tasks_tenant_status", "tenant_id", "status"),
    )

Tres decisiones del modelo que importan:

1. tenant_id BIGINT NOT NULL, no nullable. Una fila sin tenant es una fila huérfana, prohibida.
2. ForeignKey con ondelete="RESTRICT" previene borrar un tenant que aún tiene filas asociadas. Para "borrar" un tenant debe haber un proceso explícito que limpie sus datos primero.
3. Índices compuestos que empiezan por tenant_id. Sin estos, las queries que filtran por tenant + otra columna escanean toda la tabla. Con estos, PostgreSQL navega directo a las filas del tenant.

Migración inicial con Alembic

# Generar la migration desde el modelo
alembic revision --autogenerate -m "add tenants and tasks tables"
alembic upgrade head

La migration generada incluye los CREATE INDEX automáticamente porque están en __table_args__.

Filtrado de queries: la parte donde fallan los equipos

Ya tienes el schema. Ahora cada query a tasks debe filtrar por tenant_id. La forma "obvia" es agregarlo en cada lugar:

# app/api/tasks.py — versión obvia, frágil
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)  # CRÍTICO
        .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)  # CRÍTICO también acá
    )
    task = result.scalar_one_or_none()
    if task is None:
        # Sin tenant_id en el WHERE, esto retornaría tasks de OTROS tenants
        return {"error": "not found"}
    return task

El problema con esta forma es lo que ya viste en la cápsula 01: una sola query nueva que olvide ese .where(Task.tenant_id == ...) abre un leak. Y a medida que el equipo crece, la probabilidad de que alguien lo olvide aumenta.

Patrón de mitigación 1: encapsular en una clase de repositorio

En lugar de queries crudas dispersas en endpoints, encapsulas el acceso en un repository que recibe tenant_id siempre:

# 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  # filtrado siempre por este 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 — usando el 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

Lo que ganas: la única forma de saltarse el filtro es escribir SQL crudo en lugar de usar el repository. Eso es detectable en code review (todos saben que Task se accede solo por TaskRepository).

Lo que NO ganas: seguridad real. Un dev nuevo puede escribir select(Task) directo en un endpoint y la app lo permite. El repository es convención, no obligación.

Patrón de mitigación 2: prohibir queries directas con lint

Puedes escribir una regla de lint custom (con ruff o flake8) que detecte select(Task) fuera del archivo del repository y falle el CI. Es factible pero requiere mantener la regla actualizada cada vez que agregas un modelo.

Una alternativa más pragmática: agregar una validación a CI con grep simple que falle si encuentra select(Task fuera de archivos permitidos:

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

# Buscar select(Task) o select(Project), etc. fuera de la carpeta de repos
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

Lo agregas al pipeline de CI. Falla el build si alguien escribe una query directa.

Patrón de mitigación 3: mixin que requiera filtrado

Algunos equipos usan un mixin de SQLAlchemy que sobrescribe el query default. Es seductor pero frágil — los mixins de query requieren disciplina extra y muchos casos rompen la abstracción. La cápsula 04 muestra por qué RLS es la solución correcta a este problema y no un mixin.

Ejemplo trabajado: dos tenants en la misma DB

Vamos a probar todo el setup con dos tenants concretos (Acme y Globex) y verificar que el filtrado funciona — y también qué pasa cuando se olvida el filtro.

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 — crea dos tenants y tareas para cada uno
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()

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

        # Tasks de Globex
        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())

Ejecuta:

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

Query correcta vs query con bug

# 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:
        # CASO 1: usando el repository — filtrado correcto
        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}")

        # CASO 2: query directa olvidando el filtro — BUG
        result = await db.execute(select(Task).order_by(Task.created_at))
        all_tasks = result.scalars().all()
        print(f"\nQuery sin filtro (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())

Ejecuta:

python scripts/demo_filtering.py

Output esperado:

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

Query sin filtro (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

Esto es exactamente el cross-tenant leak. La query "del bug" trajo tasks de los dos tenants. Si esa query estuviera detrás del endpoint GET /tasks que el usuario de Acme llamó, Acme acabaría de ver tasks de Globex en su dashboard.

PostgreSQL no tiene forma de saber que esa query estaba "mal". Para PostgreSQL, esa query es válida y eficiente. El filtrado depende 100% del código de aplicación. Ese es el techo natural del modelo.

Indexado correcto: el detalle que mata performance

Una trampa común en shared schema es no indexar correctamente para queries multi-tenant. La regla:

Casi todos los índices deben empezar por tenant_id.

Razón: PostgreSQL navega B-tree indexes desde la columna izquierda. Si el índice es (created_at) y tu query es WHERE tenant_id = 1 ORDER BY created_at, PostgreSQL puede usar el índice para el ORDER BY pero tiene que scanear filas de todos los tenants antes de filtrar. Si el índice es (tenant_id, created_at), PostgreSQL navega directo a las filas de tenant 1 ya ordenadas por created_at.

-- ❌ MAL: índice solo por created_at
CREATE INDEX idx_tasks_created ON tasks (created_at);

-- ✅ BIEN: índice compuesto empezando por tenant_id
CREATE INDEX idx_tasks_tenant_created ON tasks (tenant_id, created_at DESC);

-- ✅ BIEN: para queries por status dentro de un tenant
CREATE INDEX idx_tasks_tenant_status ON tasks (tenant_id, status);

Excepción: índices únicos a nivel global (ej: email en una tabla de users si emails son únicos en toda la plataforma) no llevan tenant_id primero. Pero esos casos son raros — más común es que el unique sea por tenant: UNIQUE (tenant_id, email).

Verificación con EXPLAIN

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

Con índice correcto verás algo como:

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

Sin índice verás Seq Scan on tasks (escaneo completo de la tabla). Eso a 100k filas tarda ~30ms. A 5M filas tarda 1.5s. A 50M tarda 15s. La performance degrada linealmente con el tamaño de la tabla, no con el tamaño del tenant.

¿Por qué importa esto en el trabajo real?

1. Es el modelo más usado en la industria. GitHub, Slack, Linear, la mayoría de SaaS B2B y consumer empezaron con shared schema y tenant_id. Conocer este modelo a fondo es la base.

2. Es donde más bugs pasan a producción. Casi todos los cross-tenant leaks documentados públicamente (GitHub Octopus 2021, varios incidentes menores en scaleups) empezaron con un olvido en una query. Si entiendes el modelo, anticipas el bug.

3. Los patterns de mitigación que aprendes acá se usan también con RLS. El repository pattern, el lint custom, el code review checklist — todo eso sigue siendo útil aún cuando agregues RLS. RLS es la red de seguridad final, pero los layers anteriores siguen valiendo.

4. La decisión de migrar a RLS depende de entender los límites de este modelo. Si nunca implementaste shared schema sin RLS, no puedes argumentar "RLS resuelve este problema concreto". La cápsula 04 va a basarse en que ya lo viste.

Trampas y errores comunes

Error 1 (conceptual): pensar que "filtrar manualmente" es seguro si "el equipo es disciplinado"

Síntoma: equipo de 4 devs senior que confía en su disciplina. "Llevamos 2 años sin un solo bug de filtrado, no necesitamos RLS." Llega un dev nuevo, hace un PR, y el bug entra.

Por qué pasa: la disciplina no se transmite por ósmosis. Cada dev nuevo tiene que aprenderla. Cuanto más grande el equipo, más probabilidad de que alguien se la salte. Es problema de probabilidad acumulada, no de capacidad individual.

Cómo distinguirlo: pregunta "si mañana contratamos 5 devs nuevos en un mes, ¿cuál es la probabilidad de que ninguno olvide el filtro en una query nueva en su primer trimestre?". Si la respuesta honesta es "no muy alta", el modelo ya no escala.

Cómo corregir: combinar disciplina con tooling (lint en CI) y, cuando el equipo crezca, agregar RLS como red de seguridad. Las cápsulas 04 y 05 cubren cómo.

Error 2 (técnico): índices sin tenant_id al inicio

Síntoma: queries que filtran por tenant + otra columna degradan exponencialmente con el tamaño de la tabla. EXPLAIN ANALYZE muestra Seq Scan en lugar de Index Scan.

Por qué pasa: se sigue el reflejo "indexar por created_at porque ordeno por created_at" sin pensar que la query primero filtra por tenant_id. PostgreSQL no puede usar un índice por created_at para filtrar primero por tenant_id.

Cómo distinguirlo: ejecuta EXPLAIN ANALYZE en queries comunes de tu app. Si ves Seq Scan, falta el índice compuesto.

Cómo corregir: crea índices compuestos que empiecen por tenant_id. Reemplaza los índices "single column" por compuestos cuando sea posible (un índice (tenant_id, X) también sirve para queries que filtran solo por tenant_id, no necesitas un índice extra solo por tenant_id).

Error 3 (operacional): tenant_id como INT en lugar de BIGINT

Síntoma: después de 2 años el tenant_id llega a 2 mil millones (overflow de INT) y la app empieza a fallar con integer out of range. Migrar INT a BIGINT en una tabla con miles de millones de filas es operacionalmente caro (requiere reescribir toda la tabla).

Por qué pasa: se subestima cuánto crece tenant_id. En un consumer SaaS, cada signup crea una entrada — 2 mil millones de signups parece imposible hasta que pasa.

Cómo distinguirlo: revisa la definición de tenant_id en tu schema. Si es INTEGER, considera migrar a BIGINT cuanto antes (mientras la tabla aún sea chica).

Cómo corregir: desde el día uno, usa BIGINT para tenant_id (y para PKs en general). El espacio extra es negligible (8 bytes vs 4) comparado con el costo de migrar después.

Error 4 (conceptual): asumir que JOIN con tenants protege automáticamente

Síntoma: alguien argumenta "si hago JOIN tasks ON tasks.tenant_id = current_user.tenant_id, ya estoy filtrando". Pero el JOIN requiere que el código construya correctamente la condición — si el código no la incluye, el JOIN no aparece.

Por qué pasa: confunde el SQL final con el código que lo genera. El SQL puede tener filtrado, pero si el ORM/query no lo agrega, no aparece.

Cómo distinguirlo: mira el SQL generado por SQLAlchemy con echo=True. Si la query no menciona tenant_id en WHERE o JOIN ON, está sin filtrar.

Cómo corregir: la encapsulación en repository previene esto. Las queries directas son donde aparecen los bugs.

Error 5 (operacional): usar CASCADE en tenant_id

Síntoma: alguien define ForeignKey("tenants.id", ondelete="CASCADE") pensando "si borro el tenant, se borra todo". Después un cron job borra accidentalmente un tenant y borra millones de tareas. Recovery requiere restore desde backup.

Por qué pasa: "borrar un tenant" suele ser proceso explícito y monitoreado, no operación casual. CASCADE convierte un borrado accidental en catástrofe silenciosa.

Cómo distinguirlo: revisa todos tus FKs a tenants.id. Si son CASCADE, es señal de alarma.

Cómo corregir: usa ondelete="RESTRICT" (el default es restrict en PostgreSQL si no especificas). Si quieres borrar un tenant, escribe el script explícito que primero limpia datos y después borra el tenant.

Ejercicios

Ejercicio 1: detectar queries con bug de filtrado en code review

Te llega este PR. Revísalo y lista todas las queries que tienen bug de filtrado por tenant.

# 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()

Ejercicio 2: refactorizar el código del Ejercicio 1 a repository pattern

Toma el endpoint delete_project del Ejercicio 1 (con su bug) y refactoriza el código para usar un ProjectRepository que reciba tenant_id en su constructor. Después escribe el endpoint usando el 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}

Ejercicio 3: identificar el índice correcto para una query dada

Para cada una de estas queries (todas en una tabla tasks multitenant con millones de filas), identifica el índice compuesto óptimo.

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'

Ejercicio 4: escribir un test que detecte cross-tenant leak

Escribe un test pytest async que:

1. Cree dos tenants (Acme y Globex) con tasks en cada uno.
2. Use el TaskRepository con tenant_id=acme.id.
3. Verifique que repo.list() retorna solo las tasks de Acme.
4. Verifique que repo.get(globex_task_id) retorna None (no expone tasks de Globex aunque conozcas el 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: dos tenants con 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()

    # Repository de Acme
    acme_repo = TaskRepository(db, tenant_id=acme.id)

    # 1. list() retorna solo tasks de Acme
    acme_tasks = await acme_repo.list()
    assert len(acme_tasks) == 1
    assert acme_tasks[0].title == "Acme task A"

    # 2. get(id_de_otro_tenant) retorna None aunque el ID exista
    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(id_propio) sí retorna
    own = await acme_repo.get(acme_task.id)
    assert own is not None
    assert own.title == "Acme task A"

Ejercicio 5: argumentar a tu lead por qué shared schema sin RLS ya no escala

Tu equipo tiene 6 meses con 80 clientes y un equipo de 4 devs Python. Vienen 3 devs nuevos el próximo trimestre. Tu lead dice "shared schema con tenant_id nos viene funcionando bien, sigamos así". Articula 3 argumentos para migrar a RLS antes de que el equipo crezca.

Ejercicio 6: detectar índice mal puesto

Esta migración la propuso un dev nuevo. ¿Qué tiene de mal? ¿Cómo lo corregirías?

# 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():
    # Índice principal: tenant + status (para filtros por estado dentro del tenant)
    op.create_index(
        "ix_tasks_tenant_status",
        "tasks",
        ["tenant_id", "status"],
    )
    # Índice secundario: tenant + created_at (para listings ordenados)
    op.create_index(
        "ix_tasks_tenant_created",
        "tasks",
        ["tenant_id", "created_at"],
    )
    # Índice para asignaciones: 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")

Resumen y siguiente paso

En esta cápsula aprendiste:

• Shared schema con tenant_id es el modelo más simple y más usado: una sola DB, una sola tabla por entidad, columna tenant_id para distinguir el dueño de cada fila.
• El filtrado depende 100% de la disciplina del código. Cada query debe incluir WHERE tenant_id = ... o el bug se vuelve cross-tenant leak.
• Los patterns de mitigación reducen el riesgo pero no lo eliminan: repository pattern (encapsula filtrado), lint custom (detecta queries directas), code review checklist (humano revisa).
• El indexado correcto requiere índices compuestos que empiecen por tenant_id. Sin esto, queries comunes degradan exponencialmente con el tamaño de la tabla.
• El modelo tiene un techo natural: un solo WHERE tenant_id olvidado = breach. La probabilidad acumulada aumenta con el tamaño del equipo y la cantidad de queries.
• Por eso este modelo necesita una red de seguridad a nivel DB cuando el equipo crece. Esa red es Row-Level Security (RLS).

Antes de avanzar deberías poder:

• Implementar un modelo SQLAlchemy multitenant correcto con tenant_id y los índices compuestos apropiados.
• Encapsular queries en un repository que reciba tenant_id en su constructor.
• Detectar queries con bug de filtrado en code review.
• Argumentar por qué shared schema sin RLS no escala con el tamaño del equipo.
• Identificar índices mal puestos en una migración propuesta.

Siguiente cápsula — Row-Level Security: fundamentos. Vas a aprender el mecanismo que PostgreSQL ofrece para aplicar el filtro por tenant_id automáticamente, sin depender de que el código lo agregue. Vas a entender cómo se activa RLS, cómo se escriben policies, qué pasa con el role owner (FORCE ROW LEVEL SECURITY), y por qué hay que repetir el caveat más importante: RLS aquí se usa solo para multi-tenancy, NO para auth/RBAC (mezclarlos lleva a debug infierno). Es la cápsula que cierra el techo del modelo de hoy con una red de seguridad a nivel DB.

Recursos

1. PostgreSQL — CREATE INDEX reference — sintaxis y opciones de índices, especialmente índices compuestos.
2. Use The Index, Luke! — Concatenated Indexes — explicación clásica de por qué el orden de columnas en un índice compuesto importa. Lectura obligatoria.
3. Crunchy Data — Designing Your Postgres Database for Multi-Tenancy — patrones prácticos de shared schema con tenant_id.
4. SQLAlchemy 2.0 — Async ORM tutorial — referencia oficial del ORM async usado en esta cápsula.
5. FastAPI — Dependencies with sub-dependencies — patrón usado en get_task_repo que conecta auth + DB session + repository.
6. GitHub Engineering — How we designed our database for multi-tenancy — referencia de cómo grandes plataformas implementan tenant_id en escala (búsqueda en el blog).
7. Brandur — Postgres-only stacks at scale — perspectiva sobre los límites prácticos de PostgreSQL en multitenant a escala alta.

Módulo 4 — SQL Patterns for Production APIs Guide

Siguiente cápsula: Row-Level Security: fundamentos — el mecanismo que cierra el techo de la disciplina manual con una garantía a nivel de base de datos.