Schema-per-tenant: when and how

Capsule overview

You already saw the models at the "maximum sharing" extreme (a shared schema with tenant_id, capsule 03) and in the middle (a shared schema with RLS, capsules 04-05). This capsule covers the opposite extreme: schema-per-tenant, where each tenant has its own PostgreSQL schema in the same database. It's the heaviest option operationally and the one most frequently chosen for the wrong reasons ("it's cleaner," "it's more secure"). The goal of this capsule is to give you the judgment to know when it's really worth paying the operational cost — and when it's over-engineering disguised as elegance.

You're going to learn to implement schema-per-tenant in SQLAlchemy 2.0 async + FastAPI 0.110+: how a new schema gets created at signup, how search_path gets set from a FastAPI dependency (the equivalent of RLS's SET LOCAL app.tenant_id), how Alembic does NOT handle N schemas by default and what wrapper you have to write to apply migrations to all of them, what happens when a migration fails halfway through N schemas, and why cross-tenant queries (global analytics) get substantially more complicated.

By the end you'll have clear judgment: you'll be able to argue technically why for 9 out of 10 B2B SaaS products schema-per-tenant is a worse decision than RLS, and you'll know how to identify the 1 in 10 case where it is the right answer. That clarity is the last piece of the module's decision matrix.

Mental model: separate buildings, not separate floors

Let's go back to the office building metaphor you used in capsule 03.

Shared schema with tenant_id: one building, one floor, all the employees of all the companies share offices and shelves. The folders have a company label but they live physically together. If somebody searches badly, they can take the wrong folder.

Shared schema with RLS: the same building and floor, but there's a security guard at each shelf who checks the credential before handing over the folder. The folders are still physically together, but the guard prevents mistakes.

Schema-per-tenant: completely separate buildings, one per company. Each company has its building key. It's impossible for an Acme employee to enter Globex's building (they don't have the key). But now every time something architectural changes (installing air conditioning), you have to go to EACH building one by one and apply the change.

Schema-per-tenant's key question is: "do I need separate buildings or are guards at the shelves enough?". For 9 out of 10 B2B SaaS the answer is "guards" (RLS). Schema-per-tenant is the answer when there's an explicit requirement demanding separate buildings — an enterprise contract that says "we want to know our data is in a documentable dedicated schema," a regulation demanding physical isolation, or a per-tenant customization requiring schema variations.

When schema-per-tenant is the right answer

Only in these cases:

1. Enterprise contracts with an explicit "dedicated schema" SLA

Some large companies (banking, health, government) sign contracts where a clause says something like:

"The Customer's data will be stored in a dedicated PostgreSQL schema, identifiable by the name 'customer_', from which the Customer may request a full export via pg_dump at any time."

If your SaaS wants to close deals with those buyers, schema-per-tenant is the literal answer to that clause. RLS doesn't satisfy it (the data is still physically mixed even though it's logically isolated).

2. Regulation that demands documentable physical isolation

HIPAA in healthcare, certain PCI-DSS frameworks for finance, government data sovereignty regulations. These auditors want to see:

• That each customer's data is in a separate schema.
• That the app can't access more than one schema without re-authenticating.
• That backups are done per schema (a customer can request their backup without affecting others).

RLS passes many auditors but NOT all. Schema-per-tenant passes all of them.

3. Per-tenant customization that requires a different schema

If Acme needs a cost_center column in tasks that Globex doesn't have, schema-per-tenant allows it naturally. A shared schema forces you to add the column for every tenant (NULL for the ones that don't use it), or to use JSONB for extra fields (with its own query overhead).

Caveat: most "per-tenant customization" is better solved with a custom fields system inside the shared schema (a custom_fields table associating tenant_id + field_name + value). Schema-per-tenant for customization is overkill except in very specific cases.

4. Tenants with radically different volumes

If Acme has 100M rows in tasks and Globex has 1k, schema-per-tenant lets you treat them differently: Acme with its own maintenance plan, custom indexes, aggressive retention. In a shared schema, the maintenance plan applies to the whole table.

Caveat: PostgreSQL 14+ with declarative partitioning covers many of these cases without requiring separate schemas. Consider partitioning before schema-per-tenant.

5. Bring-your-own-DB as an explicit feature

Some products (Supabase, certain low-code platforms) offer "you have your own PostgreSQL database" as a sales feature. In those cases, schema-per-tenant is the intermediate step before DB-per-tenant. But this is product-specific, not a general pattern.

When NOT to use schema-per-tenant (most cases)

• ❌ "It's conceptually cleaner." Cleanliness isn't an operational metric.
• ❌ "It's more secure than RLS." It's only marginally more secure and at a high operational cost. Well-implemented RLS satisfies most auditors.
• ❌ "Someday we might have a customer who asks for it." Don't design for hypotheticals. When a customer asks for it, migrate THAT specific customer to their schema (a hybrid model).
• ❌ "We have few tenants now, it'll be easy." Today it's 30, tomorrow it's 500. The operational cost grows linearly with the number of schemas.
• ❌ "Migrations are easy, we just apply them to more schemas." That simplicity vanishes when a migration fails halfway or when the sum of the times keeps you from deploying at the pace you need.

Implementation with SQLAlchemy 2.0 async

A data model with no tenant_id

Unlike a shared schema, the tables in schema-per-tenant do NOT need tenant_id. The isolation is physical: the tenant_acme.tasks table only contains Acme's tasks, end of story.

# 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
    )
    # There's NO tenant_id. The isolation is by schema.

There's a tenants table that lives in a "public" or "shared" schema (by convention public or shared):

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

    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 points at the tenant's PostgreSQL schema name: tenant_acme, tenant_globex, etc.

Creating the schema at signup

When a new customer signs up, you have to:

1. Create an entry in public.tenants.
2. Create the PostgreSQL schema with its name.
3. Apply all the current migrations to that schema (create tables, indexes, 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:
    """
    Creates a new tenant: a record in public.tenants + a PostgreSQL schema +
    applying the migrations.
    """
    schema_name = f"tenant_{slug}"

    # 1. Create the record in public.tenants
    tenant = Tenant(slug=slug, name=name, schema_name=schema_name)
    db.add(tenant)
    await db.flush()

    # 2. Create the PostgreSQL schema.
    # CRITICAL: use identifier escaping. The f-string here only because the slug
    # is already validated (alphanumeric + underscores only).
    if not slug.replace("_", "").isalnum():
        raise ValueError(f"Invalid slug: {slug}")

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

    # 3. Create the schema's tables.
    # In production this is done with Alembic (see the next section).
    # For this capsule we simplify with a direct CREATE TABLE.
    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. Grant permissions to the app_user role for the new 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

A critical validation: slug has to be sanitized before interpolating it into the SQL. PostgreSQL identifiers (schema names) can't be parameterized — they get interpolated literally. If slug comes from user input with no validation, there's a SQL injection risk.

The dependency that sets search_path

Schema-per-tenant's equivalent of RLS's SET LOCAL app.tenant_id is SET LOCAL search_path TO .... The dependency does the same thing conceptually: BEFORE the endpoint runs, it configures PostgreSQL so the queries go to the right schema.

# 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:
    """Extracts the tenant from the request and looks it up in the 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")

    # Look the tenant up in public.tenants to get 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]:
    """A session with search_path configured to the tenant's schema."""
    async with SessionLocal() as session:
        async with session.begin():
            # CRITICAL: search_path determines which schema gets used.
            # tenant.schema_name comes from the DB, not from user input,
            # so it's validated.
            await session.execute(
                text(f'SET LOCAL search_path TO "{tenant.schema_name}", public')
            )
            yield session

Three important details:

1. SET LOCAL search_path, not SET search_path. The same reason as in RLS (capsule 05): with pooling, a setting that persists between requests is a bug.

2. Include public after the tenant's schema. The search_path TO "tenant_acme", public means "look in tenant_acme first, then in public." This lets queries for the tenant's tables (like tasks) go to the tenant's schema, while queries against global tables (like public.tenants) keep working.

3. The schemas get interpolated literally because they come from the DB. tenant.schema_name is already in public.tenants with a value controlled by the app (not direct user input). It's safe.

The endpoint using the 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)):
    # This query goes to "tenant_acme.tasks" or "tenant_globex.tasks"
    # depending on the search_path. We do NOT need 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)  # We do NOT pass a tenant_id
    db.add(task)
    await db.flush()
    await db.refresh(task)
    return task

Notice: the endpoint's code is simpler than with RLS. There's no tenant_id in the queries, no setting to verify. That's schema-per-tenant's "elegance" that seduces devs. What you do NOT see here is the operational cost that comes next.

The operational problem: migrations to N schemas

This is where schema-per-tenant stops being elegant and gets heavy. Alembic, SQLAlchemy's standard migration system, doesn't handle schema-per-tenant out of the box.

What Alembic does by default

A typical migration:

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

If you run alembic upgrade head, this adds the column to ONE tasks table. In which schema? Whichever is in the search_path (default: public). If you have 500 schemas with a tasks table, this only updates one (probably the wrong one).

A wrapper to apply migrations to every schema

You need to write logic that iterates over every tenant and applies the migrations to each schema. A simplified version:

# 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):
    """
    Runs `alembic upgrade head` pointing at the specific schema.
    Alembic accepts an environment variable or config for search_path.
    """
    cfg = Config("alembic.ini")
    cfg.set_main_option("schema", schema_name)
    # You need to modify Alembic's env.py to read this setting
    # and apply SET search_path before the 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())

And modify Alembic's env.py so it respects the schema setting:

# alembic/env.py (simplified)
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:
        # Set search_path before applying the migrations
        connection.execute(text(f'SET search_path TO "{schema}", public'))

        context.configure(
            connection=connection,
            target_metadata=None,  # or your Base.metadata
            include_schemas=False,
            version_table_schema=schema,  # each schema has its own version table
        )

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

Notice version_table_schema=schema: each schema has its own alembic_version table for tracking migrations. This is necessary because each tenant can be on a different version (if a migration failed for one tenant but passed for the others).

What happens when a migration fails

Imagine: 500 tenants, a migration that takes 30 seconds per tenant. Estimated total: 4 hours. Halfway through (tenant 250), a migration fails because a constraint already exists in that specific schema (an edge case).

The resulting state:

• Schemas 1-249: the migration applied, the schema on the new version.
• Schema 250: halfway through the migration, an inconsistent state (it may have added the column but failed on the index).
• Schemas 251-500: the migration NOT applied, the schema on the old version.

Your production app now:

• Assumes the schema's new version in code (it expects the new column).
• Works for tenants 1-249.
• Works PARTIALLY for tenant 250 (it depends what exactly happened).
• Fails for tenants 251-500 (the new column doesn't exist).

Recovery:

1. Investigate why tenant 250 failed (review logs, diagnose).
2. Clean up tenant 250's inconsistent state manually (a partial rollback).
3. Continue the migration for tenants 251-500.
4. Meanwhile, tenants 251-500 get 500 errors.

This is schema-per-tenant at 500 tenants. At 50 it's manageable. At 5000 it's a full-time job. At 50000 it's blocking.

Comparison with RLS

With a shared schema + RLS, the same migration:

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

One table, one operation, one transaction. If it fails, it fails atomically and rolls back. There's no inconsistent state between tenants. Everyone sees the old version or everyone sees the new one.

That's why schema-per-tenant's operational cost is the real metric, not the "conceptual cleanliness."

Cross-tenant queries: what you lose

In a shared schema (with or without RLS), a "global" query like "how many tasks were created today across all tenants" is trivial:

-- In a shared schema, with an RLS bypass for admin
SELECT COUNT(*) FROM tasks WHERE created_at > NOW() - INTERVAL '1 day';

In schema-per-tenant, you have to iterate every schema and union the results:

-- In schema-per-tenant, a "global" query
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 more UNION ALLs
) sub;

At 500 schemas, this query is enormous and slow. There are workarounds:

• Generate the SQL dynamically from the list of tenants. It works but the query is massive.
• An aggregations table maintained with triggers or jobs. It costs complexity.
• A materialized view table per schema refreshed periodically. It costs storage.

None of them is as simple as a shared schema's SELECT COUNT(*) FROM tasks. If your product has global dashboards (for admins, for internal analytics, for reporting), schema-per-tenant makes them 10x more complex.

Why does this matter in real work?

1. It's the decision most teams regret in hindsight. Teams that choose schema-per-tenant for "elegance" frequently migrate back to RLS after 2 years of operational pain. The migration back is a months-long project.

2. It's the right decision when a contract explicitly asks for it. If you end up with a large enterprise customer with this clause, schema-per-tenant is the literal answer. There are no shortcuts.

3. Knowing the operational cost lets you negotiate better. When a customer asks for a "dedicated schema," you can say: "our model is shared with RLS and DB-guaranteed isolation. If you need a documentable dedicated schema, we can migrate ONLY your account to a separate schema for an additional fee that covers the operational cost." It's a defensible upsell.

4. It's what distinguishes a senior dev from a junior in architecture. A junior chooses by elegance. A senior chooses by total cost of ownership over 3 years. Knowing this trade-off is a signal of seniority.

Traps and common mistakes

Mistake 1 (conceptual): choosing by "cleanliness" without measuring the operational cost

Symptom: the team chooses schema-per-tenant because "it's the cleanest." Twelve months later they have 150 tenants and the deploys take hours. Every migration is a mini-project.

Why it happens: "conceptual cleanliness" is seductive. The endpoint's code really is simpler. What you do NOT see until later is the operational cost.

How to tell: ask: "are you willing to dedicate 30% of a senior dev's time to schema operations in 18 months? If the answer is no, don't choose schema-per-tenant."

How to fix it: RLS by default. Schema-per-tenant only when a contract/regulation explicitly demands it.

Mistake 2 (operational): assuming Alembic "just works" with N schemas

Symptom: the team turns on schema-per-tenant, runs alembic upgrade head and discovers it only got applied to the public schema. The tenant schemas were left unmigrated.

Why it happens: Alembic is designed for one migration per DB, not per schema. It doesn't iterate schemas automatically.

How to tell: check alembic_version in each schema. If only public.alembic_version exists (not tenant_acme.alembic_version, etc.), the configuration is incomplete.

How to fix it: write the migrate_all_tenants.py wrapper you saw in this capsule. Modify env.py to respect the schema parameter. Document the process in runbooks.

Mistake 3 (operational): not handling the case of a migration failing halfway

Symptom: a migration fails on tenant 250 of 500. The team has no recovery script, doesn't know what state each schema is in. It has to investigate one by one.

Why it happens: people assume the migrations are atomic globally when in reality each schema is independent.

How to tell: the migrations wrapper has to record which tenants migrated successfully. If it doesn't record it, in case of failure you can't know where to resume.

How to fix it: the wrapper has to:

1. Record progress (in an auxiliary table or a structured log).
2. If it fails, report exactly which tenants were left unmigrated.
3. Allow retrying from where it failed (not from scratch).
4. Have a "rollback all" command: revert the tenants already migrated if most of them failed.

Mistake 4 (operational): forgotten permissions when creating a new schema

Symptom: a new tenant gets created via provision_tenant. The app fails with permission denied for relation tasks in that specific tenant. The existing tenants work.

Why it happens: the GRANTs were applied to the role only when the first tenant got created. For new tenants, you have to reapply the GRANTs (or configure ALTER DEFAULT PRIVILEGES).

How to tell: run \dn+ new_tenant in psql. Check the app_user role's permissions.

How to fix it: the provision_tenant function has to include the explicit GRANTs for the app_user role (you saw it in the code). And also ALTER DEFAULT PRIVILEGES in case tables get added in the future:

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

Mistake 5 (conceptual): mixing schema-per-tenant with RLS

Symptom: the team turns on schema-per-tenant AND RLS on each schema's tables "for more security." Result: double the complexity, horrible debugging, degraded performance from policies that add nothing.

Why it happens: "more layers of security is better" as a fallacy. Schema-per-tenant already gives physical isolation, RLS on top is redundant.

How to tell: review the tables in the tenant schemas. If they have RLS enabled with policies filtering on something other than tenant_id (because the tables don't have tenant_id), something's off.

How to fix it: pick ONE model. Either schema-per-tenant WITHOUT RLS, or a shared schema WITH RLS. Don't mix them.

Mistake 6 (operational): not monitoring disk usage per schema

Symptom: one specific tenant grows out of control. The DB hits 90% disk. Nobody knew which tenant was responsible until investigating manually.

Why it happens: schema-per-tenant gives you per-tenant usage granularity, but PostgreSQL doesn't expose it in standard metrics — you have to query it actively.

How to tell: SELECT pg_size_pretty(pg_total_relation_size(schemaname || '.' || tablename)) per schema. If you don't monitor this, you don't know.

How to fix it: a dashboard showing disk usage per schema. Alerts when a schema grows past a threshold. Monthly reporting to sales (a customer growing a lot = an upsell opportunity).

Exercises

Exercise 1: decide between RLS and schema-per-tenant for a case

For each scenario, decide which model you'd recommend and justify it with three quantifiable reasons:

a) A B2B SaaS project management product. 600 active customers, none with a "dedicated schema" clause. A team of 8 devs. Buyers ask for an "isolation guarantee" but accept a technical answer.

b) A healthcare product for hospitals. 25 active customers, HIPAA contracts with an explicit clause "patient data in a separate schema per institution." External auditors review annually.

c) A consumer-like product (individual accounts). 8 million users. A team of 30 devs. No enterprise contracts.

d) A multi-brand retail platform. 12 large brands (each with its own domain, branding, and a different product catalog). Some brands ask for deep customization of the data model.

Exercise 2: implement provision_tenant with error handling

The provision_tenant function you saw has a bug: if it fails halfway (e.g. the CREATE TABLE fails but CREATE SCHEMA already went through), an empty schema with no tables is left behind along with a record in public.tenants pointing at that schema. Rewrite the function so it's atomic: either everything gets created, or nothing does.

async def provision_tenant(db: AsyncSession, slug: str, name: str) -> Tenant:
    """
    Provisions a new tenant atomically: if any step fails,
    a complete rollback happens (no orphaned schemas remain).
    """
    if not slug.replace("_", "").isalnum():
        raise ValueError(f"Invalid slug: {slug}")

    schema_name = f"tenant_{slug}"

    # CRITICAL: we do everything inside a single transaction.
    # CREATE SCHEMA and CREATE TABLE support transactions in PostgreSQL.
    # If any step fails, the COMMIT doesn't run and everything is reverted.
    try:
        async with db.begin():
            # 1. Verify the schema doesn't already exist
            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. Verify the slug isn't taken
            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. Create the record in public.tenants
            tenant = Tenant(slug=slug, name=name, schema_name=schema_name)
            db.add(tenant)
            await db.flush()

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

            # 5. Create the tables in the 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. Permissions
            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 for future tables in the schema
            await db.execute(text(
                f'ALTER DEFAULT PRIVILEGES IN SCHEMA "{schema_name}" '
                f'GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO app_user'
            ))

            # If we get here with no exception, the async with commits.
            return tenant

    except Exception as e:
        # The rollback already happened automatically.
        # Here we only log and re-raise.
        # Important: do NOT do a manual cleanup (DROP SCHEMA) because
        # the rollback already reverted everything.
        raise RuntimeError(f"Failed to provision tenant {slug}: {e}") from e

Exercise 3: modify Alembic's env.py for a dynamic schema

Modify the alembic/env.py file so it respects a target_schema variable configured when running Alembic. The variable has to be passable via:

1. A command-line argument: alembic upgrade head -x schema=tenant_acme.
2. An environment variable: ALEMBIC_SCHEMA=tenant_acme alembic upgrade head.

The env.py has to set search_path before applying the migrations and use version_table_schema so each schema has its own 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)

# Import your models
from app.db.models import Base
target_metadata = Base.metadata


def get_target_schema() -> str:
    """
    Determines the target schema in this order:
    1. -x schema=... on the command line
    2. ALEMBIC_SCHEMA in the environment variables
    3. 'public' as the default
    """
    # 1. The -x argument
    x_args = context.get_x_argument(as_dictionary=True)
    if "schema" in x_args:
        return x_args["schema"]

    # 2. The environment variable
    env_schema = os.getenv("ALEMBIC_SCHEMA")
    if env_schema:
        return env_schema

    # 3. The 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:
        # Set search_path to the target schema BEFORE any query.
        # The migrations are going to operate on tables in this schema.
        connection.execute(text(f'SET search_path TO "{target_schema}", public'))

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            include_schemas=False,
            # Each schema has its own alembic_version.
            # This lets different tenants be on different versions.
            version_table_schema=target_schema,
        )

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


def run_migrations_offline():
    """We don't support offline mode in this 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()

# Migrate the public schema (default)
alembic upgrade head

# Migrate a specific tenant via -x
alembic upgrade head -x schema=tenant_acme

# Migrate via an environment variable
ALEMBIC_SCHEMA=tenant_globex alembic upgrade head

# Migrate ALL the tenants (from the wrapper script)
python scripts/migrate_all_tenants.py

Exercise 4: predict the cost of a migration to 500 tenants

Your product has 500 tenants in schema-per-tenant. You need to add three things to tasks:

1. A priority INTEGER NOT NULL DEFAULT 0 column (fast: ~5 seconds per schema).
2. An index CREATE INDEX ON tasks (assigned_to) (medium: ~15 seconds per schema with 100k rows).
3. A constraint CHECK (priority >= 0 AND priority <= 10) (fast: ~3 seconds per schema).

Compute:

a) The total time if they run serially (one schema at a time). b) The total time if they run with a parallelism of 5 schemas at a time. c) What risk shows up with parallelism? d) Compare with a shared schema + RLS (the same change).

Exercise 5: design the hybrid model

Your product is on a shared schema with RLS. You have 200 tenants. An enterprise customer arrives (Megacorp) who signs a big contract but demands a clause: "Megacorp's data must be in a dedicated PostgreSQL schema, identifiable as 'tenant_megacorp'."

You do NOT want to migrate ALL your tenants to schema-per-tenant (that would be overkill for 199 small tenants). You want a hybrid model: 199 tenants on the shared schema with RLS + Megacorp in its dedicated schema.

Design the approach:

1. How does the app decide whether to go to the shared schema or Megacorp's schema?
2. How does the tenant dependency get modified to support both modes?
3. How do the migrations get handled (the shared schema's vs Megacorp's schema's)?

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' or '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":
                # RLS mode: set the tenant's context
                await session.execute(
                    text(f"SET LOCAL app.tenant_id = {tenant.id}")
                )
            elif tenant.isolation_mode == "dedicated_schema":
                # Schema-per-tenant mode: set 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. Migrate the shared schema
    print("Migrating shared schema (public)...")
    subprocess.run(["alembic", "upgrade", "head"], check=True)

    # 2. Migrate the dedicated schemas (1 for now: 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)

Exercise 6: argue for or against schema-per-tenant for your product

Your lead asks you: "We're about to start a new B2B SaaS product. 0 customers today, we expect 100-500 in the first year. The buyers are mid-sized companies, with no large enterprise contracts expected in the short term. Schema-per-tenant to start 'clean'?"

Articulate your recommendation with 3 clear arguments and a concrete proposal.

Summary and next step

In this capsule you learned:

• Schema-per-tenant gives physical isolation (each tenant in its own PostgreSQL schema) with simpler application code (no tenant_id in queries).
• When it's the right answer: enterprise contracts with a "dedicated schema" clause, regulation demanding physical isolation (HIPAA, PCI), deep per-tenant customization, or tenants with radically different volumes.
• When NOT to use it: "it's cleaner," "it's more secure," "we might have a customer who asks for it someday." 9 out of 10 B2B SaaS products don't need schema-per-tenant.
• Implementation: models with no tenant_id, a dependency with SET LOCAL search_path, an atomic provision_tenant function that creates the schema + tables + permissions.
• The real operational cost: Alembic does NOT handle N schemas out of the box (you need a wrapper), migrations of N schemas take N times longer, failures halfway leave an inconsistent state.
• Cross-tenant queries are heavy: a shared schema's trivial global query becomes a UNION ALL over N schemas in schema-per-tenant.
• The hybrid model: most tenants on a shared schema + RLS, only dedicated enterprise ones in their own schema. The best of both when there's explicit demand.

Before moving on you should be able to:

• Decide between RLS and schema-per-tenant for a real case with quantitative justification.
• Implement a provision_tenant that creates the schema + tables + permissions atomically.
• Modify Alembic's env.py to support per-schema migrations.
• Anticipate the total time of a migration to N schemas and the risks of parallelism.
• Design the hybrid model (shared with RLS + dedicated for specific customers).
• Argue for or against schema-per-tenant in a real architectural decision.

Next capsule — Cross-tenant leak: anti-patterns. You now know the three models in depth. Now you're going to learn to anticipate and diagnose the anti-patterns that end in real cross-tenant leaks: SQLAlchemy queries with no tenant_id filter (the classic bug), forgotten JOINs, cron jobs with the wrong role, seed code that leaks between tenants, response objects that expose the other tenant's data through a mapping mistake, and more. You're going to see the bug's exact code, how they end in a breach, and why RLS prevents them even when there's a bug in the code. It's the capsule that synthesizes the whole module into a practical defense checklist.

Resources

1. PostgreSQL — CREATE SCHEMA — the official reference for schemas.
2. PostgreSQL — search_path reference — schema-per-tenant's key setting.
3. Alembic — Operating with multi-schema environments — the official cookbook on multi-schema migrations.
4. Crunchy Data — Multi-tenant solutions in Postgres — a pragmatic analysis comparing schema-per-tenant with the other options.
5. Citus Data — Schema-based multi-tenancy — a perspective on when schema-per-tenant scales and when it doesn't.
6. AWS — Silo, Pool, and Bridge models — AWS's terminology for the models (silo = schema-per-tenant, pool = shared schema, bridge = hybrid).
7. Hasura — Schema-based multi-tenancy — Hasura's specific implementation, useful for understanding the model in production.
8. Brandur — Postgres-only stacks at scale — a perspective on when PostgreSQL hits its limits with many schemas.

Module 4 — SQL Patterns for Production APIs Guide

Next capsule: Cross-tenant leak: anti-patterns — the most common mistakes that end in a breach and how each one gets prevented.