Multi-tenancy with Row-Level Security + mock auth

This capsule defines TaskFlow as a multi-tenant SaaS. You'll add the users and projects tables with tenant_id, configure Row-Level Security in PostgreSQL, implement a FastAPI dependency that sets app.tenant_id per request, and the first aggressive isolation test: tenant A trying to read tenant B's data even with "malicious" queries (with no WHERE tenant_id).

By the end you'll have secure multi-tenancy at the DB level. Even if the app code has a bug and omits the tenant filter, RLS protects on the backend.

---

Model: add users and projects

# app/models/user.py
from datetime import datetime, timezone
import uuid
from sqlalchemy import String, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
    )
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("tenants.id"),
        nullable=False,
    )
    email: Mapped[str] = mapped_column(String(200), unique=True)
    name: Mapped[str] = mapped_column(String(200))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

# app/models/project.py
from datetime import datetime, timezone
import uuid
from sqlalchemy import String, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Project(Base):
    __tablename__ = "projects"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
    )
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("tenants.id"),
        nullable=False,
    )
    name: Mapped[str] = mapped_column(String(200))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

Update app/models/__init__.py:

from app.models.tenant import Tenant
from app.models.user import User
from app.models.project import Project

__all__ = ["Tenant", "User", "Project"]

Generate the migration:

alembic revision --autogenerate -m "add users and projects with tenant_id"
alembic upgrade head

---

Enable RLS

RLS can't be declared directly in SQLAlchemy. You have to add it in a migration with raw SQL.

alembic revision -m "enable RLS on users, projects, tasks"

Edit the generated file:

# alembic/versions/XXX_enable_rls.py
def upgrade() -> None:
    # Enable RLS on multi-tenant tables
    op.execute("ALTER TABLE users ENABLE ROW LEVEL SECURITY")
    op.execute("ALTER TABLE projects ENABLE ROW LEVEL SECURITY")

    # Create policies
    op.execute("""
        CREATE POLICY tenant_isolation_users ON users
        USING (tenant_id::text = current_setting('app.tenant_id', TRUE))
    """)

    op.execute("""
        CREATE POLICY tenant_isolation_projects ON projects
        USING (tenant_id::text = current_setting('app.tenant_id', TRUE))
    """)


def downgrade() -> None:
    op.execute("DROP POLICY tenant_isolation_users ON users")
    op.execute("DROP POLICY tenant_isolation_projects ON projects")
    op.execute("ALTER TABLE users DISABLE ROW LEVEL SECURITY")
    op.execute("ALTER TABLE projects DISABLE ROW LEVEL SECURITY")

Apply:

alembic upgrade head

How the policies work

USING (tenant_id::text = current_setting('app.tenant_id', TRUE))

• tenant_id::text: cast from UUID to text to compare.
• current_setting('app.tenant_id', TRUE): reads the value of the app.tenant_id setting from the current session. The TRUE means "no error if it isn't set, return NULL".
• If the condition isn't met, the row isn't visible to the SELECT.

SET LOCAL app.tenant_id = 'xxx' sets the value only during the current transaction. Until the next SET or the COMMIT/ROLLBACK.

Important: superuser bypasses RLS

By default, superusers (postgres) bypass RLS. For tests, use a normal user:

CREATE ROLE app_user LOGIN PASSWORD 'app_password';
GRANT CONNECT ON DATABASE taskflow TO app_user;
GRANT USAGE ON SCHEMA public TO app_user;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app_user;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO app_user;

And connect the app with app_user, not postgres. In docker-compose.yml:

DATABASE_URL: postgresql+asyncpg://app_user:app_password@pgbouncer:5432/taskflow

---

Dependency: set app.tenant_id per request

# app/deps.py
from typing import AsyncGenerator
from fastapi import Depends, HTTPException, Header, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
import jwt

from app.database import SessionLocal
from app.config import settings


async def get_current_tenant_id(
    authorization: str = Header(None, alias="Authorization"),
) -> str:
    """Decode the JWT and extract tenant_id."""
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Missing or invalid Authorization header")

    token = authorization.removeprefix("Bearer ")

    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
        tenant_id = payload.get("tenant_id")
        if not tenant_id:
            raise HTTPException(401, "Token missing tenant_id")
        return tenant_id
    except jwt.PyJWTError:
        raise HTTPException(401, "Invalid token")


async def get_db_with_tenant(
    tenant_id: str = Depends(get_current_tenant_id),
) -> AsyncGenerator[AsyncSession, None]:
    """Session with `app.tenant_id` set for RLS."""
    async with SessionLocal() as session:
        # Set tenant_id on the session — RLS reads it
        await session.execute(
            text("SET LOCAL app.tenant_id = :tid"),
            {"tid": tenant_id}
        )
        yield session

Endpoints use get_db_with_tenant as a dependency:

# app/main.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.database import get_db
from app.deps import get_db_with_tenant
from app.models import Project


@app.get("/projects")
async def list_projects(
    db: AsyncSession = Depends(get_db_with_tenant),
):
    # RLS automatically filters by the tenant_id that was set
    result = await db.execute(select(Project))
    return result.scalars().all()

Note: the query is SELECT * FROM projects — with no WHERE tenant_id = .... RLS adds it automatically.

---

Mock auth

To avoid implementing full OAuth, a simple mock:

# app/routers/auth.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import jwt
from datetime import datetime, timezone, timedelta

from app.config import settings


router = APIRouter()


class LoginRequest(BaseModel):
    email: str
    tenant_id: str  # Mock: the client declares its tenant


class LoginResponse(BaseModel):
    access_token: str


@router.post("/auth/login", response_model=LoginResponse)
async def login(request: LoginRequest):
    """Mock login: in real production this would verify the password against the DB.

    For the TaskFlow demo, it simply generates a JWT with the provided tenant_id.
    """
    # In real production: verify email/password in the DB
    # Here: we accept any valid login

    payload = {
        "sub": request.email,
        "tenant_id": request.tenant_id,
        "exp": datetime.now(timezone.utc) + timedelta(hours=24),
    }
    token = jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
    return LoginResponse(access_token=token)

Include the router in main:

# app/main.py
from app.routers import auth

app.include_router(auth.router)

---

Basic CRUD endpoints for projects

# app/routers/projects.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from pydantic import BaseModel
import uuid

from app.deps import get_db_with_tenant, get_current_tenant_id
from app.models import Project


router = APIRouter()


class ProjectCreate(BaseModel):
    name: str


class ProjectResponse(BaseModel):
    id: uuid.UUID
    name: str
    tenant_id: uuid.UUID


@router.get("/projects")
async def list_projects(
    db: AsyncSession = Depends(get_db_with_tenant),
):
    result = await db.execute(select(Project))
    return [ProjectResponse(id=p.id, name=p.name, tenant_id=p.tenant_id) for p in result.scalars()]


@router.post("/projects", status_code=status.HTTP_201_CREATED)
async def create_project(
    data: ProjectCreate,
    tenant_id: str = Depends(get_current_tenant_id),
    db: AsyncSession = Depends(get_db_with_tenant),
):
    project = Project(name=data.name, tenant_id=uuid.UUID(tenant_id))
    db.add(project)
    await db.commit()
    return ProjectResponse(id=project.id, name=project.name, tenant_id=project.tenant_id)

Include it in main:

from app.routers import projects
app.include_router(projects.router)

---

RLS isolation test — the critical test

This is the test of the module. If it passes, multi-tenancy is well implemented.

# tests/test_rls_isolation.py
import pytest
import uuid
from httpx import AsyncClient
from sqlalchemy import text


@pytest.mark.asyncio
async def test_tenant_a_cannot_read_tenant_b_data(
    client: AsyncClient,
    tenant_a_id: str,
    tenant_b_id: str,
    tenant_a_token: str,
    tenant_b_token: str,
):
    """Tenant A must not be able to read Tenant B's projects, even with a 'malicious' query."""

    # Tenant A creates a project
    response_a = await client.post(
        "/projects",
        json={"name": "Tenant A's Secret Project"},
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    assert response_a.status_code == 201
    project_a_id = response_a.json()["id"]

    # Tenant B creates a project
    response_b = await client.post(
        "/projects",
        json={"name": "Tenant B's Project"},
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    assert response_b.status_code == 201

    # Tenant A lists — sees only its own
    response = await client.get(
        "/projects",
        headers={"Authorization": f"Bearer {tenant_a_token}"}
    )
    projects = response.json()
    assert len(projects) == 1
    assert projects[0]["id"] == project_a_id
    assert projects[0]["name"] == "Tenant A's Secret Project"

    # Tenant B lists — sees only its own
    response = await client.get(
        "/projects",
        headers={"Authorization": f"Bearer {tenant_b_token}"}
    )
    projects = response.json()
    assert len(projects) == 1
    assert projects[0]["name"] == "Tenant B's Project"


@pytest.mark.asyncio
async def test_rls_blocks_direct_select_without_filter(db_session_tenant_a, db_session_tenant_b):
    """Even with raw SQL and no WHERE tenant_id, RLS filters."""

    # Tenant B inserts a project
    await db_session_tenant_b.execute(text("""
        INSERT INTO projects (id, name, tenant_id)
        VALUES (gen_random_uuid(), 'Tenant B Project', :tid)
    """), {"tid": tenant_b_id})
    await db_session_tenant_b.commit()

    # Tenant A does a SELECT with no filter — should NOT see B's project
    result = await db_session_tenant_a.execute(text("SELECT name FROM projects"))
    rows = result.fetchall()
    names = [r[0] for r in rows]
    assert "Tenant B Project" not in names


@pytest.mark.asyncio
async def test_without_tenant_id_set_no_rows_visible(db_no_tenant):
    """Without SET app.tenant_id, no row is visible (security default)."""

    # Insert some projects (with a superuser that bypasses RLS)
    # ... setup ...

    # Query WITHOUT setting app.tenant_id
    result = await db_no_tenant.execute(text("SELECT * FROM projects"))
    rows = result.fetchall()
    assert len(rows) == 0  # RLS blocks everything if tenant_id isn't set

These tests are aggressive: they attempt explicit ways to bypass. They pass only with RLS configured correctly.

Fixtures setup

# tests/conftest.py
import pytest
import asyncio
from testcontainers.postgres import PostgresContainer
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy import text
from httpx import AsyncClient
import jwt
import uuid

from app.main import app
from app.config import settings


@pytest.fixture(scope="session")
def event_loop():
    return asyncio.new_event_loop()


@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16") as pg:
        yield pg


@pytest.fixture(scope="session")
async def setup_db(postgres_container):
    """Create the DB with the full schema + RLS enabled."""
    db_url = postgres_container.get_connection_url().replace("psycopg2", "asyncpg")
    settings.database_url = db_url

    engine = create_async_engine(db_url)

    async with engine.begin() as conn:
        # Apply all migrations via Alembic or directly
        from alembic.config import Config
        from alembic import command

        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", db_url)
        command.upgrade(alembic_cfg, "head")

        # Create app_user (no RLS bypass)
        await conn.execute(text("CREATE ROLE app_user LOGIN PASSWORD 'pwd'"))
        await conn.execute(text("GRANT ALL ON ALL TABLES IN SCHEMA public TO app_user"))
        await conn.execute(text("GRANT USAGE ON ALL SEQUENCES IN SCHEMA public TO app_user"))

    yield engine
    await engine.dispose()


@pytest.fixture
async def tenant_a_id(setup_db):
    """Create a tenant for tests."""
    engine = setup_db
    async with engine.begin() as conn:
        result = await conn.execute(
            text("INSERT INTO tenants (name) VALUES ('Tenant A') RETURNING id")
        )
        return str(result.scalar())


@pytest.fixture
async def tenant_b_id(setup_db):
    engine = setup_db
    async with engine.begin() as conn:
        result = await conn.execute(
            text("INSERT INTO tenants (name) VALUES ('Tenant B') RETURNING id")
        )
        return str(result.scalar())


@pytest.fixture
def tenant_a_token(tenant_a_id):
    payload = {"sub": "a@test.com", "tenant_id": tenant_a_id}
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)


@pytest.fixture
def tenant_b_token(tenant_b_id):
    payload = {"sub": "b@test.com", "tenant_id": tenant_b_id}
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)


@pytest.fixture
async def client(setup_db):
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

Run the tests:

pytest tests/test_rls_isolation.py -v

If the 3 tests pass, multi-tenancy is well implemented.

---

Pitfalls and common mistakes

1. Forgetting to connect the app with a non-superuser.

If the app uses postgres (superuser), it bypasses RLS. The tests would "pass" locally but the protection doesn't exist. Verify the connection with app_user.

2. RLS without BYPASSRLS on superusers for migrations.

Migrations run with a superuser and modify tables. If RLS blocks, the migrations fail. By default, the superuser bypasses — that's what you want.

3. current_setting('app.tenant_id', TRUE) with FALSE instead of TRUE.

-- Without TRUE, error if the setting doesn't exist
USING (tenant_id::text = current_setting('app.tenant_id', FALSE))

-- With TRUE, returns NULL if it doesn't exist (safer)
USING (tenant_id::text = current_setting('app.tenant_id', TRUE))

TRUE means "missing_ok = TRUE". Recommended to avoid errors in initial setups.

4. SET app.tenant_id (without LOCAL) that persists across transactions.

SET app.tenant_id = '...';        -- Persists until the connection closes
SET LOCAL app.tenant_id = '...';  -- Only during this transaction

With PgBouncer transaction mode, connections are reused. SET (not SET LOCAL) can leak the tenant between requests. Always SET LOCAL.

5. Forgetting policies on INSERT/UPDATE.

USING (...) applies to SELECT/UPDATE/DELETE. For INSERT, use WITH CHECK:

CREATE POLICY tenant_isolation_users ON users
USING (tenant_id::text = current_setting('app.tenant_id', TRUE))
WITH CHECK (tenant_id::text = current_setting('app.tenant_id', TRUE));

WITH CHECK prevents inserting with a tenant_id different from the one that was set.

6. Hardcoded JWT secret.

JWT_SECRET = "dev-secret" in the code is a demo. Real production: a rotated secret, stored securely.

7. FORCE ROW LEVEL SECURITY.

ALTER TABLE users FORCE ROW LEVEL SECURITY;

Makes the table's owner also subject to RLS. Useful if you want even the postgres user (in migration queries) not to bypass. By default, it doesn't.

---

Summary and next step

What you have now:

• The users and projects tables with tenant_id.
• RLS enabled with policies that filter by current_setting('app.tenant_id').
• app_user, a non-superuser that respects RLS.
• A mock JWT for auth.
• A FastAPI dependency get_db_with_tenant that sets SET LOCAL app.tenant_id.
• The /projects endpoints working with multi-tenancy.
• An isolation test that proves tenant A doesn't see tenant B's data.

Commit:

git add .
git commit -m "feat: multi-tenancy with RLS + JWT mock + projects CRUD"

In the next capsule we add TaskFlow's central entity: tasks with basic CRUD, cursor pagination, and soft delete. You'll implement GET /tasks with an opaque cursor based on (created_at DESC, id DESC), DELETE /tasks/{id} that sets deleted_at, and a partial index on deleted_at IS NULL for fast queries.

---

Resources

1. PostgreSQL Docs — Row Security Policies — official reference.
2. PostgreSQL Docs — current_setting — runtime parameters.
3. Supabase — Row Level Security — real case with RLS.
4. Crunchy Data — Multi-tenancy with RLS — deep dive.
5. PostgreSQL Wiki — RLS — history and patterns.
6. GitLab — Multi-tenancy decisions — real architectural decisions.
7. JWT.io — Decoder — for debugging tokens.

---

Capsule 03 of 08 — Module 8 — SQL Patterns for Production APIs Guide