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.
For further actions, you may consider blocking this person and/or reporting abuse
