Building Multi-Tenant SaaS with SQLAlchemy Row-Level Security

I've seen the same mistake twice: a founder builds a multi-tenant SaaS, implements tenant checks in their application code, and six months into production, a customer accidentally queries another tenant's data. The bug wasn't in the business logic—it was a forgotten .filter(User.tenant_id == current_tenant_id) in some API endpoint.

That's why I'm obsessive about enforcing tenant isolation at the database layer. Application code lies. SQL doesn't.

This is how CitizenApp prevents tenant data leaks: PostgreSQL Row-Level Security (RLS) + SQLAlchemy session management. Not belt-and-suspenders paranoia—it's the only pattern I trust for production SaaS.

The Problem with Application-Layer Isolation

Most developers stop at this:

// TypeScript API endpoint
async function getUsersByTenant(tenantId: string) {
  return await db.user.findMany({
    where: { tenantId }
  });
}

This works until:

A junior dev forgets the filter
A query gets refactored and the tenant check is lost
An admin endpoint bypasses the check "temporarily"
A background job runs as a system user

I've burned time debugging all four. The issue is trust—you're relying on developers to remember a rule that should be enforced by infrastructure.

PostgreSQL Row-Level Security Is Your Safety Net

RLS makes it impossible to accidentally leak data. The database itself denies access before your application code even runs.

Here's the pattern I use in CitizenApp:

-- Create a tenant table
CREATE TABLE tenants (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name VARCHAR NOT NULL,
  created_at TIMESTAMP DEFAULT NOW()
);

-- Create a users table with tenant_id
CREATE TABLE users (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
  email VARCHAR UNIQUE NOT NULL,
  created_at TIMESTAMP DEFAULT NOW()
);

-- Enable RLS on users table
ALTER TABLE users ENABLE ROW LEVEL SECURITY;

-- Create a policy: users can only see rows where tenant_id matches their session variable
CREATE POLICY users_isolation ON users
  FOR SELECT
  USING (tenant_id = current_setting('app.current_tenant_id')::UUID);

-- Same for INSERT, UPDATE, DELETE
CREATE POLICY users_insert ON users
  FOR INSERT
  WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::UUID);

CREATE POLICY users_update ON users
  FOR UPDATE
  USING (tenant_id = current_setting('app.current_tenant_id')::UUID);

CREATE POLICY users_delete ON users
  FOR DELETE
  USING (tenant_id = current_setting('app.current_tenant_id')::UUID);

The magic is current_setting('app.current_tenant_id'). This is a session variable we set at connection time. PostgreSQL checks every query against it.

Integrating with SQLAlchemy

Here's where the infrastructure meets the code. I use SQLAlchemy's SessionLocal with an event listener to set the tenant context:

from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import sessionmaker, Session
from typing import Optional
import uuid

# Database setup
DATABASE_URL = "postgresql://user:password@localhost/citizenapp"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

# Global context variable (use contextvars in production)
_current_tenant_id: Optional[str] = None

def set_tenant_context(tenant_id: str):
    """Set the current tenant for this request"""
    global _current_tenant_id
    _current_tenant_id = tenant_id

def get_tenant_context() -> str:
    """Retrieve the current tenant"""
    if not _current_tenant_id:
        raise ValueError("No tenant context set")
    return _current_tenant_id

# Event listener: set PostgreSQL session variable when connection is used
@event.listens_for(Engine, "connect")
def receive_connect(dbapi_conn, connection_record):
    cursor = dbapi_conn.cursor()
    tenant_id = get_tenant_context()
    cursor.execute(
        f"SET app.current_tenant_id = '{tenant_id}'"
    )
    cursor.close()

class TenantAwareSession(Session):
    """Custom session that enforces tenant context"""
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Verify tenant context exists
        _ = get_tenant_context()

In your FastAPI application:

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthCredentials
import jwt

app = FastAPI()
security = HTTPBearer()

def get_current_tenant(credentials: HTTPAuthCredentials = Depends(security)) -> str:
    """Extract tenant from JWT token"""
    try:
        payload = jwt.decode(
            credentials.credentials,
            "your-secret-key",
            algorithms=["HS256"]
        )
        tenant_id = payload.get("tenant_id")
        if not tenant_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        return tenant_id
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/api/users")
async def list_users(tenant_id: str = Depends(get_current_tenant)):
    """List users for current tenant"""
    set_tenant_context(tenant_id)
    db = SessionLocal()
    try:
        # This query is automatically filtered by RLS
        # Even if you forgot the WHERE clause, PostgreSQL enforces it
        users = db.query(User).all()
        return users
    finally:
        db.close()

The critical detail: RLS runs at the database level, not in your ORM layer. Your query doesn't need a .filter(User.tenant_id == tenant_id) because PostgreSQL rejects any rows that don't match the session variable.

The Gotcha: Connection Pooling

This burned me for three hours on a Friday: if you use connection pooling (Pgbouncer, for example), session variables don't persist across requests. Each request might get a different physical connection.

Solution: Set the tenant context before every query:

@app.get("/api/users")
async def list_users(tenant_id: str = Depends(get_current_tenant)):
    db = SessionLocal()
    try:
        # Set tenant context every time
        set_tenant_context(tenant_id)
        db.execute(text(f"SET app.current_tenant_id = '{tenant_id}'"))

        users = db.query(User).all()
        return users
    finally:
        db.close()

Or use a middleware to handle this globally (cleaner):

from starlette.middleware.base import BaseHTTPMiddleware

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Extract tenant from token
        token = request.headers.get("authorization", "").replace("Bearer ", "")
        payload = jwt.decode(token, "secret", algorithms=["HS256"])
        tenant_id = payload["tenant_id"]

        # Set context for this request
        request.state.tenant_id = tenant_id
        set_tenant_context(tenant_id)

        response = await call_next(request)
        return response

app.add_middleware(TenantMiddleware)

Why I Prefer This Over Application-Only Checks

Defense in depth: A bug in your code can't leak data
Compliance: Auditors see RLS in the database layer
Performance: PostgreSQL optimizes queries knowing they're filtered
Team safety: Junior devs can't accidentally break isolation

The application-level checks are still good to have—they're your first line of defense. But the database layer is where isolation actually happens.

CitizenApp processes multi-tenant data for 8+ AI features across thousands of requests daily. RLS has prevented exactly zero data leaks because it's impossible to leak data that the database refuses to return.

That's production peace of mind.