Schema-per-tenant: cuándo y cómo

Descripción de la cápsula

Ya viste los modelos del extremo "compartición máxima" (shared schema con tenant_id, cápsula 03) y del medio (shared schema con RLS, cápsulas 04-05). Esta cápsula cubre el extremo opuesto: schema-per-tenant, donde cada tenant tiene su propio esquema PostgreSQL en la misma base de datos. Es la opción más pesada operacionalmente y la que más frecuentemente se elige por las razones equivocadas ("es más limpio", "es más seguro"). El objetivo de esta cápsula es darte el criterio para saber cuándo realmente vale la pena pagar el costo operacional — y cuándo es overengineering disfrazado de elegancia.

Vas a aprender a implementar schema-per-tenant en SQLAlchemy 2.0 async + FastAPI 0.110+: cómo se crea un schema nuevo en signup, cómo se setea search_path desde una FastAPI dependency (el equivalente al SET LOCAL app.tenant_id de RLS), cómo Alembic NO maneja N schemas por default y qué wrapper hay que escribir para aplicar migrations a todos, qué pasa cuando una migration falla a mitad de N schemas, y por qué cross-tenant queries (analytics globales) son sustancialmente más complicadas.

Al terminar tendrás criterio claro: vas a poder argumentar técnicamente por qué para 9 de cada 10 productos B2B SaaS schema-per-tenant es peor decisión que RLS, y vas a saber identificar el 1 de cada 10 caso donde sí es la respuesta correcta. Esa claridad es la última pieza del decision matrix del módulo.

Modelo mental: edificios separados, no pisos separados

Volvamos a la metáfora del edificio de oficinas que usaste en la cápsula 03.

Shared schema con tenant_id: un edificio, un piso, todos los empleados de todas las empresas comparten oficinas y estanterías. Las carpetas tienen una etiqueta de empresa pero conviven físicamente. Si alguien busca mal, puede tomar la carpeta equivocada.

Shared schema con RLS: mismo edificio y piso, pero hay un empleado de seguridad en cada estantería que verifica la credencial antes de entregar la carpeta. Las carpetas siguen físicamente juntas, pero el guardia previene errores.

Schema-per-tenant: edificios completamente separados, uno por empresa. Cada empresa tiene su llave del edificio. Imposible que un empleado de Acme entre al edificio de Globex (no tiene la llave). Pero ahora cada vez que se cambia algo arquitectónico (instalar aire acondicionado), hay que ir a CADA edificio uno por uno y aplicar el cambio.

La pregunta clave de schema-per-tenant es: "¿necesito edificios separados o me alcanza con guardias en estanterías?". Para 9 de cada 10 SaaS B2B la respuesta es "guardias" (RLS). Schema-per-tenant es la respuesta cuando hay un requisito explícito que demanda edificios separados — un contrato enterprise que dice "queremos saber que nuestros datos están en un esquema dedicado documentable", una regulación que demanda aislamiento físico, o una customización por tenant que requiere variaciones de schema.

Cuándo schema-per-tenant es la respuesta correcta

Solo en estos casos:

1. Contratos enterprise con SLA explícito de "esquema dedicado"

Algunas empresas grandes (banca, salud, gobierno) firman contratos donde una cláusula dice algo como:

"Los datos del Cliente se almacenarán en un esquema PostgreSQL dedicado, identificable por el nombre 'cliente_', con el cual el Cliente puede solicitar exportación completa mediante pg_dump en cualquier momento."

Si tu SaaS quiere cerrar deals con esos compradores, schema-per-tenant es la respuesta literal a esa cláusula. RLS no la satisface (los datos siguen físicamente mezclados aunque sean lógicamente aislados).

2. Regulación que demanda aislamiento físico documentable

HIPAA en healthcare, ciertos frameworks de PCI-DSS para finance, regulaciones gubernamentales de soberanía de datos. Estos auditores quieren ver:

• Que los datos de cada cliente estén en un esquema separado.
• Que la app no pueda acceder a más de un schema sin re-autenticar.
• Que los backups se hagan por schema (un cliente puede pedir su backup sin afectar otros).

RLS pasa muchos auditores pero NO todos. Schema-per-tenant pasa todos.

3. Customización por tenant que requiere schema distinto

Si Acme necesita una columna cost_center en tasks que Globex no tiene, schema-per-tenant lo permite naturalmente. Shared schema te obliga a agregar la columna a todos los tenants (NULL para los que no la usan), o a usar JSONB para campos extras (con su propio overhead de queries).

Caveat: la mayoría de "customización por tenant" se resuelve mejor con un sistema de campos custom dentro del schema compartido (tabla custom_fields que asocia tenant_id + field_name + value). Schema-per-tenant para customización es overkill salvo casos muy específicos.

4. Tenants con volúmenes radicalmente distintos

Si Acme tiene 100M filas en tasks y Globex tiene 1k, schema-per-tenant te permite tratarlos distinto: Acme con su propio plan de mantenimiento, índices custom, retention agresiva. En shared schema, el plan de mantenimiento aplica a la tabla entera.

Caveat: PostgreSQL 14+ con partitioning declarativo cubre muchos de estos casos sin requerir schemas separados. Considerá partitioning antes de schema-per-tenant.

5. Bring-your-own-DB como feature explícito

Algunos productos (Supabase, ciertas plataformas low-code) ofrecen "tienes tu propia base de datos PostgreSQL" como feature de venta. En esos casos, schema-per-tenant es el paso intermedio antes de DB-per-tenant. Pero esto es producto-específico, no patrón general.

Cuándo NO usar schema-per-tenant (la mayoría de casos)

• ❌ "Es más limpio conceptualmente." Limpieza no es métrica operacional.
• ❌ "Es más seguro que RLS." Solo es marginalmente más seguro y a costo operacional alto. RLS bien implementado satisface la mayoría de auditores.
• ❌ "Algún día podríamos tener un cliente que lo pida." No diseñes para hipótesis. Cuando un cliente lo pida, migrá ESE cliente específico a su schema (modelo híbrido).
• ❌ "Tenemos pocos tenants ahora, será fácil." Hoy son 30, mañana son 500. El costo operacional crece linealmente con la cantidad de schemas.
• ❌ "Las migrations son fáciles, solo aplicamos a más schemas." Esa simpleza desaparece cuando una migration falla a mitad o cuando la suma de tiempos te impide deployar al ritmo que necesitás.

Implementación con SQLAlchemy 2.0 async

Modelo de datos sin tenant_id

A diferencia de shared schema, las tablas en schema-per-tenant NO necesitan tenant_id. El aislamiento es físico: la tabla tenant_acme.tasks solo contiene tasks de Acme, fin del asunto.

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


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    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
    )
    # NO hay tenant_id. El aislamiento es por schema.

Hay una tabla tenants que vive en un schema "público" o "shared" (por convención public o shared):

class Tenant(Base):
    __tablename__ = "tenants"
    __table_args__ = {"schema": "public"}  # explícito

    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)
    schema_name: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)

schema_name apunta al nombre del schema PostgreSQL del tenant: tenant_acme, tenant_globex, etc.

Crear schema en signup

Cuando un nuevo cliente se registra, hay que:

1. Crear una entrada en public.tenants.
2. Crear el schema PostgreSQL con su nombre.
3. Aplicar todas las migrations actuales a ese schema (crear tablas, índices, etc.).

# app/services/tenant_provisioning.py
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import Tenant


async def provision_tenant(db: AsyncSession, slug: str, name: str) -> Tenant:
    """
    Crea un nuevo tenant: registro en public.tenants + schema PostgreSQL +
    aplicación de migrations.
    """
    schema_name = f"tenant_{slug}"

    # 1. Crear el registro en public.tenants
    tenant = Tenant(slug=slug, name=name, schema_name=schema_name)
    db.add(tenant)
    await db.flush()

    # 2. Crear el schema PostgreSQL.
    # CRÍTICO: usar identifier escaping. f-string aquí solo porque slug
    # ya está validado (alfanumérico + guiones bajos solamente).
    if not slug.replace("_", "").isalnum():
        raise ValueError(f"Invalid slug: {slug}")

    await db.execute(text(f'CREATE SCHEMA "{schema_name}"'))

    # 3. Crear las tablas del schema.
    # En producción esto se hace con Alembic (ver siguiente sección).
    # Para esta cápsula simplificamos con CREATE TABLE directo.
    await db.execute(text(f"""
        CREATE TABLE "{schema_name}".tasks (
            id BIGSERIAL PRIMARY KEY,
            title VARCHAR(200) NOT NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'open',
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
        );
    """))

    await db.execute(text(f"""
        CREATE INDEX ix_tasks_created ON "{schema_name}".tasks (created_at DESC);
    """))

    # 4. Otorgar permisos al rol app_user para el nuevo schema
    await db.execute(text(f'GRANT USAGE ON SCHEMA "{schema_name}" TO app_user'))
    await db.execute(text(f'GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA "{schema_name}" TO app_user'))
    await db.execute(text(f'GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA "{schema_name}" TO app_user'))

    await db.commit()
    return tenant

Validación crítica: slug debe ser sanitizado antes de interpolarlo en el SQL. Identifiers de PostgreSQL (nombres de schemas) no pueden parametrizarse — se interpolan literalmente. Si slug viene de input de usuario sin validación, hay riesgo de SQL injection.

Dependency que setea search_path

El equivalente en schema-per-tenant del SET LOCAL app.tenant_id de RLS es SET LOCAL search_path TO .... La dependency hace lo mismo conceptualmente: ANTES de que el endpoint corra, configura PostgreSQL para que las queries vayan al schema correcto.

# app/db/tenant_context.py
from typing import AsyncIterator
from fastapi import Depends, HTTPException, Request
from sqlalchemy import text, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import SessionLocal
from app.db.models import Tenant


async def get_current_tenant(
    request: Request,
) -> Tenant:
    """Extrae tenant del request y lo busca en la DB."""
    tenant_id_str = request.headers.get("X-Tenant-ID")
    if not tenant_id_str:
        raise HTTPException(status_code=401, detail="Missing X-Tenant-ID")
    try:
        tenant_id = int(tenant_id_str)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid X-Tenant-ID")

    # Buscar tenant en public.tenants para obtener schema_name
    async with SessionLocal() as session:
        result = await session.execute(
            select(Tenant).where(Tenant.id == tenant_id)
        )
        tenant = result.scalar_one_or_none()
        if tenant is None:
            raise HTTPException(status_code=404, detail="Tenant not found")
        return tenant


async def get_tenant_session(
    tenant: Tenant = Depends(get_current_tenant),
) -> AsyncIterator[AsyncSession]:
    """Sesión con search_path configurado al schema del tenant."""
    async with SessionLocal() as session:
        async with session.begin():
            # CRÍTICO: search_path determina qué schema se usa.
            # tenant.schema_name viene de la DB, no de input del usuario,
            # así que está validado.
            await session.execute(
                text(f'SET LOCAL search_path TO "{tenant.schema_name}", public')
            )
            yield session

Tres detalles importantes:

1. SET LOCAL search_path, no SET search_path. Misma razón que en RLS (cápsula 05): con pooling, el setting persistente entre requests es bug.

2. Incluir public después del schema del tenant. El search_path TO "tenant_acme", public significa "primero buscá en tenant_acme, después en public". Esto permite que las queries para tablas del tenant (como tasks) vayan al schema del tenant, mientras las queries a tablas globales (como public.tenants) sigan funcionando.

3. Los schemas se interpolan literalmente porque vienen de la DB. tenant.schema_name ya está en public.tenants con un valor controlado por la app (no input directo del usuario). Es seguro.

Endpoint usando la dependency

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

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

router = APIRouter()


@router.get("/tasks")
async def list_tasks(db: AsyncSession = Depends(get_tenant_session)):
    # Esta query va a "tenant_acme.tasks" o "tenant_globex.tasks"
    # según el search_path. NO necesitamos tenant_id.
    result = await db.execute(
        select(Task).order_by(Task.created_at.desc()).limit(50)
    )
    return list(result.scalars().all())


@router.post("/tasks")
async def create_task(
    title: str,
    db: AsyncSession = Depends(get_tenant_session),
):
    task = Task(title=title)  # NO pasamos tenant_id
    db.add(task)
    await db.flush()
    await db.refresh(task)
    return task

Notar: el código del endpoint es más simple que con RLS. No hay tenant_id en queries, no hay setting que verificar. Esa es la "elegancia" de schema-per-tenant que seduce a los devs. Lo que NO se ve aquí es el costo operacional que viene a continuación.

El problema operacional: migrations a N schemas

Acá es donde schema-per-tenant deja de ser elegante y se vuelve pesado. Alembic, el sistema estándar de migrations de SQLAlchemy, no maneja schema-per-tenant out-of-the-box.

Lo que Alembic hace por default

Una migration típica:

# alembic/versions/abc123_add_priority_column.py
"""add priority column to tasks"""
from alembic import op
import sqlalchemy as sa


def upgrade():
    op.add_column(
        "tasks",
        sa.Column("priority", sa.Integer(), nullable=False, server_default="0"),
    )


def downgrade():
    op.drop_column("tasks", "priority")

Si ejecutás alembic upgrade head, esto agrega la columna a UNA tabla tasks. ¿En cuál schema? El que esté en search_path (default: public). Si tienes 500 schemas con tabla tasks, esto SOLO actualiza uno (probablemente el equivocado).

Wrapper para aplicar migrations a todos los schemas

Necesitás escribir lógica que itere sobre todos los tenants y aplique migrations a cada schema. Versión simplificada:

# scripts/migrate_all_tenants.py
import asyncio
import sys
from sqlalchemy import text, select
from alembic.config import Config
from alembic import command

from app.db.session import SessionLocal
from app.db.models import Tenant


async def get_all_tenant_schemas() -> list[str]:
    async with SessionLocal() as session:
        result = await session.execute(select(Tenant.schema_name))
        return [row[0] for row in result.all()]


def run_migration_for_schema(schema_name: str):
    """
    Ejecuta `alembic upgrade head` apuntando al schema específico.
    Alembic acepta una variable de entorno o config para search_path.
    """
    cfg = Config("alembic.ini")
    cfg.set_main_option("schema", schema_name)
    # Necesitás modificar env.py de Alembic para leer este setting
    # y aplicar SET search_path antes de las migrations.
    command.upgrade(cfg, "head")


async def main():
    schemas = await get_all_tenant_schemas()
    print(f"Migrating {len(schemas)} tenant schemas...")

    failures = []
    for i, schema in enumerate(schemas, start=1):
        print(f"[{i}/{len(schemas)}] Migrating {schema}...")
        try:
            run_migration_for_schema(schema)
        except Exception as e:
            print(f"  FAILED: {e}")
            failures.append((schema, str(e)))

    if failures:
        print(f"\n{len(failures)} migrations failed:")
        for schema, error in failures:
            print(f"  - {schema}: {error}")
        sys.exit(1)
    print(f"\nAll {len(schemas)} schemas migrated successfully.")


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

Y modificar env.py de Alembic para que respete el setting schema:

# alembic/env.py (simplificado)
from alembic import context
from sqlalchemy import create_engine

config = context.config


def run_migrations_online():
    schema = config.get_main_option("schema", "public")

    connectable = create_engine(
        config.get_main_option("sqlalchemy.url"),
    )

    with connectable.connect() as connection:
        # Setear search_path antes de aplicar migrations
        connection.execute(text(f'SET search_path TO "{schema}", public'))

        context.configure(
            connection=connection,
            target_metadata=None,  # o tu Base.metadata
            include_schemas=False,
            version_table_schema=schema,  # cada schema tiene su propio version table
        )

        with context.begin_transaction():
            context.run_migrations()

Notar version_table_schema=schema: cada schema tiene su propia tabla alembic_version para tracker de migrations. Esto es necesario porque cada tenant puede estar en una versión distinta (si una migration falló para un tenant pero pasó para otros).

Lo que pasa cuando una migration falla

Imaginá: 500 tenants, una migration que tarda 30 segundos por tenant. Total estimado: 4 horas. A la mitad (tenant 250), una migration falla porque hay un constraint que ya existe en ese schema específico (caso edge).

Estado resultante:

• Schemas 1-249: migration aplicada, schema en versión nueva.
• Schemas 250: a mitad de migración, estado inconsistente (puede haber agregado la columna pero fallado en el índice).
• Schemas 251-500: NO migración aplicada, schema en versión vieja.

Tu app productiva ahora:

• Asume la versión nueva del schema en código (espera la columna nueva).
• Funciona para tenants 1-249.
• Funciona PARCIALMENTE para tenant 250 (depende de qué pasó exactamente).
• Falla para tenants 251-500 (la columna nueva no existe).

Recovery:

1. Investigar por qué falló tenant 250 (revisar logs, diagnosticar).
2. Limpiar el estado inconsistente del tenant 250 manualmente (rollback parcial).
3. Continuar la migración para tenants 251-500.
4. Mientras tanto, los tenants 251-500 reciben errores 500.

Esto es schema-per-tenant a 500 tenants. A 50 es manejable. A 5000 es un trabajo full-time. A 50000 es bloqueante.

Comparación con RLS

Con shared schema + RLS, la misma migration:

alembic upgrade head
# Output: 1 migration applied (8 segundos).

Una sola tabla, una sola operación, una sola transacción. Si falla, falla atómicamente y se hace rollback. No hay estado inconsistente entre tenants. Todos ven la versión vieja o todos ven la nueva.

Por eso el costo operacional de schema-per-tenant es la métrica real, no la "limpieza conceptual".

Cross-tenant queries: lo que pierdes

En shared schema (con o sin RLS), una query "global" como "cuántas tasks se crearon hoy across all tenants" es trivial:

-- En shared schema, con bypass RLS para admin
SELECT COUNT(*) FROM tasks WHERE created_at > NOW() - INTERVAL '1 day';

En schema-per-tenant, hay que iterar todos los schemas y unir resultados:

-- En schema-per-tenant, query "global"
SELECT COUNT(*) FROM (
    SELECT id FROM tenant_acme.tasks WHERE created_at > NOW() - INTERVAL '1 day'
    UNION ALL
    SELECT id FROM tenant_globex.tasks WHERE created_at > NOW() - INTERVAL '1 day'
    UNION ALL
    SELECT id FROM tenant_initech.tasks WHERE created_at > NOW() - INTERVAL '1 day'
    -- ... 500 UNION ALL más
) sub;

A 500 schemas, esta query es enorme y lenta. Hay workarounds:

• Generar el SQL dinámicamente desde la lista de tenants. Funciona pero el query es masivo.
• Tabla de agregaciones que se mantiene con triggers o jobs. Cuesta complejidad.
• Tabla materialised view por schema que se refresca periódicamente. Cuesta storage.

Ninguno es tan simple como el SELECT COUNT(*) FROM tasks de shared schema. Si tu producto tiene dashboards globales (para admins, para analytics internos, para reporting), schema-per-tenant los hace 10x más complejos.

¿Por qué importa esto en el trabajo real?

1. Es la decisión que más equipos lamentan en retrospectiva. Equipos que eligen schema-per-tenant por "elegancia" frecuentemente migran de vuelta a RLS después de 2 años de dolor operacional. La migración de regreso es proyecto de meses.

2. Es la decisión correcta cuando un contrato lo pide explícitamente. Si llegas a tener un cliente enterprise grande con esta cláusula, schema-per-tenant es la respuesta literal. No hay atajos.

3. Conocer el costo operacional te permite negociar mejor. Cuando un cliente pide "esquema dedicado", puedes decir: "nuestro modelo es shared con RLS y aislamiento garantizado por DB. Si necesitás esquema dedicado documentable, podemos migrar SOLO tu cuenta a un esquema separado por un fee adicional que cubre el costo operacional". Es un upsell defendible.

4. Es lo que distingue a un dev senior de un junior en arquitectura. Junior elige por elegancia. Senior elige por costo total de ownership a 3 años. Conocer este trade-off es señal de seniority.

Trampas y errores comunes

Error 1 (conceptual): elegir por "limpieza" sin medir costo operacional

Síntoma: equipo elige schema-per-tenant porque "es lo más limpio". A los 12 meses tienen 150 tenants y los deploys tardan horas. Cada migration es un mini-proyecto.

Por qué pasa: "limpieza conceptual" es seductora. El código del endpoint es realmente más simple. Lo que NO se ve hasta después es el costo operacional.

Cómo distinguirlo: preguntá: "¿estás dispuesto a dedicar 30% del tiempo de un dev senior a operations de schemas en 18 meses? Si la respuesta es no, no elijas schema-per-tenant".

Cómo corregir: RLS por default. Schema-per-tenant solo cuando un contrato/regulación lo demanda explícitamente.

Error 2 (operacional): asumir que Alembic "simplemente funciona" con N schemas

Síntoma: equipo activa schema-per-tenant, ejecuta alembic upgrade head y descubre que solo se aplicó al schema public. Los schemas de tenants quedaron sin migrar.

Por qué pasa: Alembic está diseñado para una migración por DB, no por schema. No itera schemas automáticamente.

Cómo distinguirlo: revisar alembic_version en cada schema. Si solo public.alembic_version existe (no tenant_acme.alembic_version, etc.), la configuración es incompleta.

Cómo corregir: escribir el wrapper de migrate_all_tenants.py que viste en esta cápsula. Modificar env.py para respetar el parámetro de schema. Documentar el proceso en runbooks.

Error 3 (operacional): no manejar el caso de migration fallida a mitad

Síntoma: una migration falla en el tenant 250 de 500. El equipo no tiene script de recovery, no sabe en qué estado quedó cada schema. Tiene que investigar uno por uno.

Por qué pasa: se asume que las migrations son atómicas a nivel global cuando en realidad cada schema es independiente.

Cómo distinguirlo: el wrapper de migrations debe registrar qué tenants se migraron exitosamente. Si no registra, en caso de falla no puedes saber dónde retomar.

Cómo corregir: el wrapper debe:

1. Registrar progreso (en una tabla auxiliar o log estructurado).
2. Si falla, reportar exactamente qué tenants quedaron sin migrar.
3. Permitir reintentar desde donde falló (no desde cero).
4. Tener un comando de "rollback all": revertir tenants ya migrados si la mayoría falló.

Error 4 (operacional): permisos olvidados al crear schema nuevo

Síntoma: se crea un tenant nuevo via provision_tenant. La app falla con permission denied for relation tasks en ese tenant específico. Los tenants existentes funcionan.

Por qué pasa: los GRANT se aplicaron solo al rol cuando se creó el primer tenant. Para nuevos tenants, hay que volver a aplicar GRANTs (o configurar ALTER DEFAULT PRIVILEGES).

Cómo distinguirlo: ejecutar \dn+ tenant_nuevo en psql. Verificar permisos del rol app_user.

Cómo corregir: la función provision_tenant debe incluir los GRANT explícitos para el rol app_user (lo viste en el código). Y también ALTER DEFAULT PRIVILEGES por si en el futuro se agregan tablas:

ALTER DEFAULT PRIVILEGES IN SCHEMA "tenant_acme"
    GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO app_user;

Error 5 (conceptual): mezclar schema-per-tenant con RLS

Síntoma: equipo activa schema-per-tenant Y RLS sobre las tablas de cada schema "para más seguridad". Resultado: complejidad doble, debugging horrible, performance degradada por policies que no aportan nada.

Por qué pasa: "más capas de seguridad es mejor" como falacia. Schema-per-tenant ya da aislamiento físico, RLS encima es redundante.

Cómo distinguirlo: revisar las tablas de los schemas de tenants. Si tienen RLS activa con policies que filtran por algo distinto a tenant_id (porque las tablas no tienen tenant_id), algo está raro.

Cómo corregir: elegí UN modelo. O schema-per-tenant SIN RLS, o shared schema CON RLS. No mezclar.

Error 6 (operacional): no monitorear disk usage por schema

Síntoma: un tenant específico crece descontroladamente. La DB llega al 90% de disco. Nadie sabía qué tenant era el responsable hasta investigar manualmente.

Por qué pasa: schema-per-tenant te da granularidad de uso por tenant, pero PostgreSQL no la expone en métricas estándar — hay que consultarla activamente.

Cómo distinguirlo: SELECT pg_size_pretty(pg_total_relation_size(schemaname || '.' || tablename)) por schema. Si no monitoreas esto, no sabes.

Cómo corregir: dashboard que muestre disk usage por schema. Alertas cuando un schema crezca más allá de un umbral. Reporting mensual a ventas (cliente que crece mucho = oportunidad de upsell).

Ejercicios

Ejercicio 1: decidir entre RLS y schema-per-tenant para un caso

Para cada escenario, decide qué modelo recomendarías y justifica con tres razones cuantificables:

a) Producto B2B SaaS de gestión de proyectos. 600 clientes activos, ninguno con cláusula de "schema dedicado". Equipo de 8 devs. Compradores piden "garantía de aislamiento" pero aceptan respuesta técnica.

b) Producto de healthcare para hospitales. 25 clientes activos, contratos HIPAA con cláusula explícita "datos de pacientes en esquema separado por institución". Auditores externos revisan anualmente.

c) Producto consumer-like (cuentas individuales). 8 millones de usuarios. Equipo de 30 devs. Sin contratos enterprise.

d) Plataforma multi-marca de retail. 12 marcas grandes (cada una con su propio dominio, branding, catálogo de productos diferente). Algunas marcas piden customización profunda del modelo de datos.

Ejercicio 2: implementar provision_tenant con manejo de errores

La función provision_tenant que viste tiene un bug: si falla a mitad (ej: el CREATE TABLE falla pero CREATE SCHEMA ya pasó), queda un schema vacío sin tablas y un registro en public.tenants apuntando a ese schema. Reescribí la función para que sea atómica: o todo se crea, o nada se crea.

async def provision_tenant(db: AsyncSession, slug: str, name: str) -> Tenant:
    """
    Provisiona un nuevo tenant atómicamente: si cualquier paso falla,
    se hace rollback completo (no quedan schemas huérfanos).
    """
    if not slug.replace("_", "").isalnum():
        raise ValueError(f"Invalid slug: {slug}")

    schema_name = f"tenant_{slug}"

    # CRÍTICO: hacemos todo dentro de una sola transacción.
    # CREATE SCHEMA y CREATE TABLE soportan transacciones en PostgreSQL.
    # Si algún paso falla, COMMIT no se ejecuta y todo se revierte.
    try:
        async with db.begin():
            # 1. Verificar que el schema no exista ya
            result = await db.execute(text("""
                SELECT 1 FROM information_schema.schemata
                WHERE schema_name = :schema_name
            """), {"schema_name": schema_name})
            if result.scalar_one_or_none() is not None:
                raise ValueError(f"Schema {schema_name} already exists")

            # 2. Verificar que el slug no esté tomado
            result = await db.execute(
                select(Tenant).where(Tenant.slug == slug)
            )
            if result.scalar_one_or_none() is not None:
                raise ValueError(f"Tenant slug {slug} already exists")

            # 3. Crear registro en public.tenants
            tenant = Tenant(slug=slug, name=name, schema_name=schema_name)
            db.add(tenant)
            await db.flush()

            # 4. Crear schema PostgreSQL
            await db.execute(text(f'CREATE SCHEMA "{schema_name}"'))

            # 5. Crear tablas en el schema
            await db.execute(text(f"""
                CREATE TABLE "{schema_name}".tasks (
                    id BIGSERIAL PRIMARY KEY,
                    title VARCHAR(200) NOT NULL,
                    status VARCHAR(20) NOT NULL DEFAULT 'open',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )
            """))

            await db.execute(text(f"""
                CREATE INDEX ix_tasks_created
                ON "{schema_name}".tasks (created_at DESC)
            """))

            # 6. Permisos
            await db.execute(text(
                f'GRANT USAGE ON SCHEMA "{schema_name}" TO app_user'
            ))
            await db.execute(text(
                f'GRANT SELECT, INSERT, UPDATE, DELETE '
                f'ON ALL TABLES IN SCHEMA "{schema_name}" TO app_user'
            ))
            await db.execute(text(
                f'GRANT USAGE, SELECT ON ALL SEQUENCES '
                f'IN SCHEMA "{schema_name}" TO app_user'
            ))

            # Default privileges para futuras tablas en el schema
            await db.execute(text(
                f'ALTER DEFAULT PRIVILEGES IN SCHEMA "{schema_name}" '
                f'GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO app_user'
            ))

            # Si llegamos acá sin excepción, el async with hace commit.
            return tenant

    except Exception as e:
        # El rollback ya pasó automáticamente.
        # Acá solo logueamos y re-raise.
        # Importante: NO hacer cleanup manual (DROP SCHEMA) porque
        # el rollback ya revirtió todo.
        raise RuntimeError(f"Failed to provision tenant {slug}: {e}") from e

Ejercicio 3: modificar env.py de Alembic para schema dinámico

Modificá el archivo alembic/env.py para que respete una variable target_schema configurada al ejecutar Alembic. La variable debe poder pasarse via:

1. Argumento de línea de comandos: alembic upgrade head -x schema=tenant_acme.
2. Variable de entorno: ALEMBIC_SCHEMA=tenant_acme alembic upgrade head.

El env.py debe setear search_path antes de aplicar las migrations y usar version_table_schema para que cada schema tenga su propio tracker.

# alembic/env.py
import os
from logging.config import fileConfig

from sqlalchemy import engine_from_config, pool, text
from alembic import context

config = context.config
fileConfig(config.config_file_name)

# Importar tus modelos
from app.db.models import Base
target_metadata = Base.metadata


def get_target_schema() -> str:
    """
    Determina el schema target en este orden:
    1. -x schema=... en la línea de comandos
    2. ALEMBIC_SCHEMA en variables de entorno
    3. 'public' como default
    """
    # 1. Argumento -x
    x_args = context.get_x_argument(as_dictionary=True)
    if "schema" in x_args:
        return x_args["schema"]

    # 2. Variable de entorno
    env_schema = os.getenv("ALEMBIC_SCHEMA")
    if env_schema:
        return env_schema

    # 3. Default
    return "public"


def run_migrations_online():
    target_schema = get_target_schema()
    print(f"Running migrations for schema: {target_schema}")

    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        # Setear search_path al schema target ANTES de cualquier query.
        # Las migrations van a operar sobre tablas en este schema.
        connection.execute(text(f'SET search_path TO "{target_schema}", public'))

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            include_schemas=False,
            # Cada schema tiene su propio alembic_version.
            # Esto permite que distintos tenants estén en versiones distintas.
            version_table_schema=target_schema,
        )

        with context.begin_transaction():
            context.run_migrations()


def run_migrations_offline():
    """No soportamos modo offline en este setup."""
    raise NotImplementedError(
        "Offline migrations not supported in schema-per-tenant setup. "
        "Use 'alembic upgrade head -x schema=...' for online mode."
    )


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

# Migrar el schema public (default)
alembic upgrade head

# Migrar un tenant específico via -x
alembic upgrade head -x schema=tenant_acme

# Migrar via variable de entorno
ALEMBIC_SCHEMA=tenant_globex alembic upgrade head

# Migrar TODOS los tenants (desde el script wrapper)
python scripts/migrate_all_tenants.py

Ejercicio 4: predecir el costo de una migración a 500 tenants

Tu producto tiene 500 tenants en schema-per-tenant. Necesitás agregar tres cosas a tasks:

1. Una columna priority INTEGER NOT NULL DEFAULT 0 (rápido: ~5 segundos por schema).
2. Un índice CREATE INDEX ON tasks (assigned_to) (medio: ~15 segundos por schema con 100k filas).
3. Un constraint CHECK (priority >= 0 AND priority <= 10) (rápido: ~3 segundos por schema).

Calculá:

a) Tiempo total si se ejecutan en serie (un schema a la vez). b) Tiempo total si se ejecutan con paralelismo de 5 schemas a la vez. c) ¿Qué riesgo aparece con paralelismo? d) Compará con shared schema + RLS (mismo cambio).

Ejercicio 5: diseñar el modelo híbrido

Tu producto está en shared schema con RLS. Tenés 200 tenants. Llega un cliente enterprise (Megacorp) que firma un contrato grande pero exige una cláusula: "los datos de Megacorp deben estar en un esquema PostgreSQL dedicado, identificable como 'tenant_megacorp'".

NO quieres migrar TODOS tus tenants a schema-per-tenant (sería overkill para 199 tenants chicos). Querés un modelo híbrido: 199 tenants en shared schema con RLS + Megacorp en su schema dedicado.

Diseñá el approach:

1. ¿Cómo decide la app si va al schema compartido o al schema de Megacorp?
2. ¿Cómo se modifica la dependency de tenant para soportar ambos modos?
3. ¿Cómo se manejan migrations (las del shared schema vs las del schema de Megacorp)?

class Tenant(Base):
    __tablename__ = "tenants"
    __table_args__ = {"schema": "public"}

    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)
    isolation_mode: Mapped[str] = mapped_column(
        String(20),  # 'shared' o 'dedicated_schema'
        nullable=False,
        default="shared",
    )
    dedicated_schema_name: Mapped[str | None] = mapped_column(
        String(100),
        nullable=True,
        unique=True,
    )

async def get_tenant_session(
    tenant: Tenant = Depends(get_current_tenant),
) -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        async with session.begin():
            if tenant.isolation_mode == "shared":
                # Modo RLS: setear contexto del tenant
                await session.execute(
                    text(f"SET LOCAL app.tenant_id = {tenant.id}")
                )
            elif tenant.isolation_mode == "dedicated_schema":
                # Modo schema-per-tenant: setear search_path
                await session.execute(
                    text(f'SET LOCAL search_path TO "{tenant.dedicated_schema_name}", public')
                )
            else:
                raise ValueError(f"Unknown isolation_mode: {tenant.isolation_mode}")

            yield session

# scripts/migrate_all.py
async def migrate_all():
    # 1. Migrar shared schema
    print("Migrating shared schema (public)...")
    subprocess.run(["alembic", "upgrade", "head"], check=True)

    # 2. Migrar dedicated schemas (1 por ahora: Megacorp)
    async with SessionLocal() as session:
        result = await session.execute(
            select(Tenant).where(Tenant.isolation_mode == "dedicated_schema")
        )
        dedicated_tenants = result.scalars().all()

    for tenant in dedicated_tenants:
        print(f"Migrating dedicated schema {tenant.dedicated_schema_name}...")
        subprocess.run([
            "alembic", "upgrade", "head",
            "-x", f"schema={tenant.dedicated_schema_name}"
        ], check=True)

Ejercicio 6: argumentar a favor o en contra de schema-per-tenant para tu producto

Tu lead te pregunta: "Estamos por arrancar un nuevo producto B2B SaaS. 0 clientes hoy, esperamos 100-500 en el primer año. Compradores son empresas medianas, sin contratos enterprise grandes esperados en el corto plazo. ¿Schema-per-tenant para empezar 'limpio'?"

Articulá tu recomendación con 3 argumentos claros y propuesta concreta.

Resumen y siguiente paso

En esta cápsula aprendiste:

• Schema-per-tenant da aislamiento físico (cada tenant en su propio schema PostgreSQL) con código de aplicación más simple (sin tenant_id en queries).
• Cuándo es la respuesta correcta: contratos enterprise con cláusula de "schema dedicado", regulación que demanda aislamiento físico (HIPAA, PCI), customización profunda por tenant, o tenants con volúmenes radicalmente distintos.
• Cuándo NO usar: "es más limpio", "es más seguro", "podríamos tener un cliente que lo pida algún día". 9 de 10 productos B2B SaaS no necesitan schema-per-tenant.
• Implementación: modelos sin tenant_id, dependency con SET LOCAL search_path, función provision_tenant atómica que crea schema + tablas + permisos.
• El costo operacional real: Alembic NO maneja N schemas out-of-the-box (necesitás wrapper), migrations de N schemas tardan N veces más, fallas a mitad dejan estado inconsistente.
• Cross-tenant queries son pesadas: la query global trivial de shared schema se vuelve UNION ALL sobre N schemas en schema-per-tenant.
• Modelo híbrido: mayoría de tenants en shared schema + RLS, solo enterprise dedicados en su schema. Lo mejor de ambos cuando hay demanda explícita.

Antes de avanzar deberías poder:

• Decidir entre RLS y schema-per-tenant para un caso real con justificación cuantitativa.
• Implementar provision_tenant que cree schema + tablas + permisos atómicamente.
• Modificar Alembic env.py para soportar migrations por schema.
• Anticipar el tiempo total de una migration a N schemas y los riesgos del paralelismo.
• Diseñar el modelo híbrido (shared con RLS + dedicated para clientes específicos).
• Argumentar a favor o en contra de schema-per-tenant en una decisión arquitectónica real.

Siguiente cápsula — Cross-tenant leak: anti-patterns. Ya conocés los tres modelos en profundidad. Ahora vas a aprender a anticipar y diagnosticar los anti-patterns que terminan en cross-tenant leaks reales: queries SQLAlchemy sin filtro por tenant_id (el bug clásico), JOINs olvidados, cron jobs con role equivocado, código de seed que filtra entre tenants, response objects que exponen datos del otro tenant por error de mapping, y más. Vas a ver el código exacto del bug, cómo terminan en breach, y por qué RLS los previene incluso cuando hay bug en código. Es la cápsula que sintetiza todo el módulo en una checklist práctica de defensa.

Recursos

1. PostgreSQL — CREATE SCHEMA — referencia oficial de schemas.
2. PostgreSQL — search_path reference — el setting clave de schema-per-tenant.
3. Alembic — Operating with multi-schema environments — cookbook oficial sobre migrations multi-schema.
4. Crunchy Data — Multi-tenant solutions in Postgres — análisis pragmático que compara schema-per-tenant con otras opciones.
5. Citus Data — Schema-based multi-tenancy — perspectiva de cuándo schema-per-tenant escala y cuándo no.
6. AWS — Silo, Pool, and Bridge models — terminología AWS para los modelos (silo = schema-per-tenant, pool = shared schema, bridge = híbrido).
7. Hasura — Schema-based multi-tenancy — implementación específica de Hasura, útil para entender el modelo en producción.
8. Brandur — Postgres-only stacks at scale — perspectiva sobre cuándo PostgreSQL llega a sus límites con muchos schemas.

Módulo 4 — SQL Patterns for Production APIs Guide

Siguiente cápsula: Cross-tenant leak: anti-patterns — los errores más comunes que terminan en breach y cómo cada uno se previene.