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.