Cheatsheet for multi-tenant SQLAlchemy. Three strategies; pick by isolation needs and cost.

Strategies

ModelIsolationCostWhen
Shared DB + tenant_idLogicalLowest< 1k tenants, B2C-ish
Schema-per-tenantStrongerMedium< 1k mid-size B2B
DB-per-tenantStrongestHighest< 100 enterprise / compliance

Shared DB + tenant_id

class TenantMixin:
    tenant_id: Mapped[int] = mapped_column(index=True)

class User(TenantMixin, Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    
    __table_args__ = (
        Index("ix_users_tenant_email", "tenant_id", "email", unique=True),
    )

Composite indexes — always include tenant_id first.

Tenant context via contextvar

import contextvars
current_tenant: contextvars.ContextVar[int | None] = contextvars.ContextVar("tenant", default=None)

Auto-filter via event

@event.listens_for(Session, "do_orm_execute")
def tenant_filter(execute_state):
    tid = current_tenant.get()
    if tid is None or execute_state.execution_options.get("bypass_tenant"):
        return
    execute_state.statement = execute_state.statement.options(
        with_loader_criteria(TenantMixin, lambda cls: cls.tenant_id == tid, include_aliases=True)
    )

Every SELECT filtered automatically.

Set tenant on insert

@event.listens_for(Session, "before_flush")
def set_tenant_on_insert(session, flush_context, instances):
    tid = current_tenant.get()
    if tid is None: return
    for obj in session.new:
        if isinstance(obj, TenantMixin) and obj.tenant_id is None:
            obj.tenant_id = tid

FastAPI middleware

@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
    tid = await resolve_tenant(request)
    token = current_tenant.set(tid)
    try:
        return await call_next(request)
    finally:
        current_tenant.reset(token)

RLS layer (Postgres) — defense in depth

ALTER TABLE users ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_iso ON users
USING (tenant_id = current_setting('app.tenant_id')::bigint)
WITH CHECK (tenant_id = current_setting('app.tenant_id')::bigint);

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def set_rls(conn, cur, stmt, params, ctx, executemany):
    tid = current_tenant.get()
    if tid is not None:
        cur.execute(f"SET LOCAL app.tenant_id = '{tid}'")

App must not connect as BYPASSRLS role. See Postgres-focused cheatsheets .

Schema-per-tenant

class Settings(BaseSettings):
    db_url: str

async def get_tenant_db(tenant: Tenant = Depends(get_tenant)) -> AsyncSession:
    async with AsyncSessionLocal() as session:
        await session.execute(text(f"SET search_path TO tenant_{tenant.slug}, public"))
        yield session

Each tenant: its own schema. Migrations run per schema (multi-tenant Alembic env.py).

DB-per-tenant

import functools

@functools.lru_cache(maxsize=1000)
def tenant_engine(dsn: str):
    return create_async_engine(dsn, pool_size=2, max_overflow=4)

async def get_tenant_db(tenant: Tenant = Depends(get_tenant)) -> AsyncSession:
    engine = tenant_engine(tenant.dsn)
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
    async with SessionLocal() as session:
        yield session

Cap engine cache; close on shutdown.

Cross-tenant admin access

async def get_admin_db():
    async with AdminSessionLocal() as session:
        await session.execute(text("SET LOCAL ROLE app_admin"))   # BYPASSRLS role
        yield session

@app.get("/admin/all-users", dependencies=[Depends(require_role("admin"))])
async def admin_list(db = Depends(get_admin_db)):
    return (await db.execute(select(User))).scalars().all()

CI test for tenant isolation

async def test_tenant_isolation(db_session):
    # Tenant 1
    current_tenant.set(1)
    db_session.add(User(email="[email protected]"))
    await db_session.commit()
    
    # Tenant 2 must not see it
    current_tenant.set(2)
    rows = (await db_session.execute(select(User).where(User.email == "[email protected]"))).scalars().all()
    assert rows == []

Common mistakes

  • Missing tenant_id filter in custom raw SQL — cross-tenant leak.
  • Unique constraints without tenant_id — collisions across tenants.
  • Connecting as BYPASSRLS role from app — defeats RLS.
  • Per-tenant engines without cap — connection explosion.
  • Schema-per-tenant with > 1000 schemas — Postgres catalog bloat.

Read this next

If you want my tenant-context + RLS + admin role pattern, it’s at rajpoot.dev .

Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .