Chapter 8: Row-Level Security in Postgres via SQLAlchemy. Per-row tenant isolation enforced at the DB level — even if your app code forgets the WHERE clause.

Why RLS

Shared-DB multi-tenancy means every query needs WHERE tenant_id = $1. Forget once; data leaks across tenants. RLS fixes this at the DB.

ALTER TABLE users ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON users
USING (tenant_id = current_setting('app.tenant_id')::bigint);

Now SELECT/UPDATE/DELETE only see rows where tenant_id matches the session-set value. WHERE clauses become redundant for safety.

Schema

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

tenant_id always present. Indexed for performance.

Migration to enable RLS

def upgrade():
    op.execute("ALTER TABLE users ENABLE ROW LEVEL SECURITY")
    op.execute("""
        CREATE POLICY tenant_isolation ON users
        USING (tenant_id = current_setting('app.tenant_id')::bigint)
        WITH CHECK (tenant_id = current_setting('app.tenant_id')::bigint)
    """)

USING for SELECT/UPDATE/DELETE. WITH CHECK for INSERT/UPDATE.

Setting tenant per request

import contextvars

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

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def set_tenant(conn, cursor, statement, parameters, context, executemany):
    tid = current_tenant.get()
    if tid is not None:
        cursor.execute(f"SET LOCAL app.tenant_id = '{tid}'")

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

Per-request: set contextvar; events apply at query time. SET LOCAL resets at transaction end.

Alternative: Depends-based

async def get_db(request: Request) -> AsyncSession:
    async with AsyncSessionLocal() as session:
        if tenant_id := getattr(request.state, "tenant_id", None):
            await session.execute(text("SET LOCAL app.tenant_id = :tid"), {"tid": tenant_id})
        yield session

Set at session start. Cleaner if you don’t need cross-cutting events.

Bypass for admin / system

For ops that need cross-tenant access:

class SystemRole(str, enum.Enum):
    ADMIN = "admin"

async def get_admin_db() -> AsyncSession:
    async with AsyncAdminSessionLocal() as session:  # uses BYPASSRLS role
        yield session

Create a Postgres role with BYPASSRLS:

CREATE ROLE app_admin WITH LOGIN BYPASSRLS PASSWORD '...';

Used only by admin paths; audit every use.

RLS for INSERT

CREATE POLICY tenant_insert ON users FOR INSERT
WITH CHECK (tenant_id = current_setting('app.tenant_id')::bigint);

Inserts that specify a different tenant_id are rejected.

Application doesn’t need WHERE

Once RLS is set:

users = await session.execute(select(User))  # already tenant-filtered

Postgres adds the predicate. No code change needed.

You can still add WHERE tenant_id = ... for clarity or planner hints; it’ll match what RLS enforces.

Performance

RLS predicates must be in indexes:

__table_args__ = (
    Index("ix_users_tenant_email", "tenant_id", "email"),
)

For queries by email, the planner uses tenant_id + email because RLS adds the tenant_id predicate.

EXPLAIN to verify: index used; partition pruned (if combined with partitioning).

Test bypass attempts

CI test:

async def test_rls_isolation(db_session):
    # Tenant 1 creates a user
    await db_session.execute(text("SET LOCAL app.tenant_id = '1'"))
    db_session.add(User(tenant_id=1, email="[email protected]"))
    await db_session.commit()
    
    # Tenant 2 shouldn't see it
    await db_session.execute(text("SET LOCAL app.tenant_id = '2'"))
    rows = (await db_session.execute(select(User).where(User.email == "[email protected]"))).scalars().all()
    assert rows == []

Verify RLS is enforced. Catches misconfigured policies.

Common mistakes

1. App role with BYPASSRLS

Defeats the purpose. App role should NOT have BYPASSRLS.

2. Forgetting WITH CHECK

Without it, INSERTs can specify any tenant_id. Add WITH CHECK or trigger validation.

3. RLS without indexes

Tenant_id added to every query; no index → seq scans. Always index tenant_id (composite with other predicates).

4. Per-connection (not LOCAL) settings

SET app.tenant_id (without LOCAL) persists across the connection’s life in pool. Use SET LOCAL.

5. Mixing RLS and explicit WHERE

Both is fine; they AND together. Just be aware.

Schema-per-tenant alternative

If RLS adds unwanted complexity:

async def get_db_for_tenant(tenant_id: int) -> AsyncSession:
    schema = f"tenant_{tenant_id}"
    async with AsyncSessionLocal() as session:
        await session.execute(text(f"SET search_path TO {schema}, public"))
        yield session

Each tenant has its own schema. Migrations run per schema; isolation is physical.

Tradeoff: more schemas; more ops. Limited to ~hundreds of tenants.

DB-per-tenant

For strictest isolation (compliance, large enterprise):

async def get_db(tenant_id: int) -> AsyncSession:
    creds = await secrets.get(f"tenant_db/{tenant_id}")
    engine = await get_or_create_engine(creds.dsn)
    async with AsyncSession(engine) as session:
        yield session

Each tenant: own DB. Most expensive; most isolated.

See Multi-Tenancy Patterns .

What’s next

Chapter 9: Postgres-specific upsert and bulk ops.

Read this next

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 .