Multi-Tenant Architecture Patterns

The Production Nightmare: When Tenant A Saw Tenant B's Orders

It was a Friday afternoon (of course it was). I was about to leave for the weekend when Slack lit up. A restaurant manager reported seeing orders from a completely different restaurant in their dashboard. Then another report came in. Then another.

My blood ran cold. This wasn't a UI glitch—this was a catastrophic multi-tenant data leak. Restaurant A could see Restaurant B's orders, revenue data, customer information. Everything.

The root cause? One missing line of code:

# What I wrote (WRONG):
@app.get("/orders")
async def get_orders(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    orders = db.query(Order).offset(skip).limit(limit).all()
    return orders

# What I should have written:
@app.get("/orders")
async def get_orders(
    skip: int = 0, 
    limit: int = 100,
    tenant_id: str = Depends(get_tenant_id),
    db: Session = Depends(get_db)
):
    orders = db.query(Order).filter(Order.tenant_id == tenant_id).offset(skip).limit(limit).all()
    return orders

One missing filter(Order.tenant_id == tenant_id) exposed every tenant's data to every other tenant. That Friday night, I learned that multi-tenancy isn't a feature—it's a fundamental architectural constraint that must be enforced at every level.

What is Multi-Tenancy?

Multi-tenancy means serving multiple customers (tenants) from a single application instance while keeping their data completely isolated.

In my POS system, I onboarded several types of clients:

• Tenant 1: A local restaurant client

• Tenant 2: A fast food chain client

• Tenant 3: A retail shop client

Each tenant should:

• See only their own data

• Be completely unaware other tenants exist

• Have isolated configuration and customization

• Never be able to access another tenant's data

Three Multi-Tenant Isolation Strategies

Let me show you each strategy as implemented in the POS system.

Strategy 1: Separate Database Per Tenant

Pros:

• Strongest isolation (physically separate data)

• Easy backup/restore per tenant

• Can customize schema per tenant

• Simple compliance (delete tenant = drop database)

Cons:

• Expensive (one DB instance per tenant)

• Complex connection management

• Hard to do cross-tenant analytics

• Schema migrations run N times

# config/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from typing import Dict
import os

class MultiTenantDatabaseManager:
    """Manages separate database per tenant"""
    
    _engines: Dict[str, any] = {}
    _sessions: Dict[str, any] = {}
    
    @classmethod
    def get_engine(cls, tenant_id: str):
        """Get or create database engine for tenant"""
        if tenant_id not in cls._engines:
            # Each tenant has separate database
            database_url = f"postgresql://user:password@localhost:5432/{tenant_id}_db"
            
            cls._engines[tenant_id] = create_engine(
                database_url,
                pool_size=5,
                max_overflow=10,
                pool_pre_ping=True
            )
            
            cls._sessions[tenant_id] = sessionmaker(
                autocommit=False,
                autoflush=False,
                bind=cls._engines[tenant_id]
            )
        
        return cls._engines[tenant_id]
    
    @classmethod
    def get_session(cls, tenant_id: str):
        """Get database session for tenant"""
        if tenant_id not in cls._sessions:
            cls.get_engine(tenant_id)
        
        return cls._sessions[tenant_id]()

# Usage in endpoints
from fastapi import Depends, Header

def get_tenant_db(x_tenant_id: str = Header(...)):
    """Dependency to get tenant-specific database session"""
    db = MultiTenantDatabaseManager.get_session(x_tenant_id)
    try:
        yield db
    finally:
        db.close()

@app.get("/orders")
async def get_orders(db: Session = Depends(get_tenant_db)):
    # No need for tenant_id filter - this database only has this tenant's data
    orders = db.query(Order).all()
    return orders

When to use: Enterprise SaaS where tenants pay premium, need guaranteed isolation, or have compliance requirements.

Strategy 2: Shared Database, Separate Schemas

Pros:

• Good isolation (schema-level separation)

• Cheaper than separate databases

• Easier connection pooling

• Can do cross-tenant queries if needed

Cons:

• Still complex migrations

• Need schema switching logic

• Backup/restore more complex

# config/database.py
from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import NullPool

# Single database with multiple schemas
engine = create_engine(
    "postgresql://user:password@localhost:5432/pos_db",
    poolclass=NullPool  # Don't pool to avoid schema switching issues
)

def get_tenant_session(tenant_id: str):
    """Get session with search_path set to tenant schema"""
    session = Session(bind=engine)
    
    # Set PostgreSQL search_path to tenant schema
    session.execute(text(f"SET search_path TO {tenant_id}, public"))
    
    return session

# Automatic schema creation for new tenants
from sqlalchemy import MetaData

def create_tenant_schema(tenant_id: str):
    """Create schema and tables for new tenant"""
    with engine.connect() as conn:
        # Create schema
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {tenant_id}"))
        conn.commit()
        
        # Create tables in schema
        metadata = MetaData(schema=tenant_id)
        # Define your models with this metadata
        # metadata.create_all(bind=engine)

# Usage
@app.get("/orders")
async def get_orders(x_tenant_id: str = Header(...)):
    db = get_tenant_session(x_tenant_id)
    try:
        # Queries automatically use tenant schema due to search_path
        orders = db.query(Order).all()
        return orders
    finally:
        db.close()

When to use: Mid-market SaaS with moderate tenant count, need good isolation but not separate databases.

Strategy 3: Shared Database, Shared Tables (Discriminator Column)

Pros:

• Simplest to implement

• Best performance (shared connection pool)

• Easy cross-tenant analytics

• Cheapest option

Cons:

• Higher risk of data leaks (my Friday nightmare)

• Must filter every query by tenant_id

• Indexes include tenant_id (larger indexes)

# models/base.py
from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declared_attr

class TenantMixin:
    """Mixin to add tenant_id to all models"""
    
    @declared_attr
    def tenant_id(cls):
        return Column(String(50), nullable=False, index=True)

# models/order.py
from sqlalchemy import Column, Integer, Float, DateTime, JSON
from models.base import TenantMixin, Base

class Order(TenantMixin, Base):
    __tablename__ = "orders"
    
    id = Column(Integer, primary_key=True)
    # tenant_id inherited from TenantMixin
    order_number = Column(String(50), nullable=False)
    total_amount = Column(Float, nullable=False)
    items = Column(JSON, nullable=False)
    
    # Composite index for performance
    __table_args__ = (
        Index('ix_orders_tenant_id_created_at', 'tenant_id', 'created_at'),
    )

This is what I use in the POS system. Here's how to do it safely:

The x-tenant-id Header Flow Through All Services

The tenant ID must flow through every request, to every service, for every operation.

Enforcing Tenant Isolation at Every Layer

# 1. Middleware Layer: Extract and validate tenant_id
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract tenant_id from header
        tenant_id = request.headers.get("x-tenant-id")
        
        if not tenant_id and request.url.path not in ["/health", "/docs"]:
            raise HTTPException(400, "Missing x-tenant-id header")
        
        # Validate tenant exists (prevent fake tenant_ids)
        if tenant_id and not await self.validate_tenant(tenant_id):
            raise HTTPException(404, "Tenant not found")
        
        # Store in request state
        request.state.tenant_id = tenant_id
        
        response = await call_next(request)
        return response
    
    async def validate_tenant(self, tenant_id: str) -> bool:
        # Check if tenant exists in tenants table
        # Cache this check for performance
        return True  # Simplified

# 2. JWT Token Layer: Verify token contains matching tenant_id
import jwt
from config.settings import settings

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        public_paths = ["/health", "/auth/login", "/auth/register"]
        if request.url.path in public_paths:
            return await call_next(request)
        
        # Extract JWT token
        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            raise HTTPException(401, "Missing authorization header")
        
        token = auth_header.replace("Bearer ", "")
        
        try:
            payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")
        
        # Critical: Verify token tenant matches header tenant
        token_tenant = payload.get("tenant_id")
        header_tenant = request.state.tenant_id
        
        if token_tenant != header_tenant:
            raise HTTPException(403, "Tenant ID mismatch - potential security violation")
        
        # Store user info
        request.state.user_id = payload.get("user_id")
        request.state.user_role = payload.get("role")
        
        response = await call_next(request)
        return response

# 3. Repository Layer: Always filter by tenant_id
class OrderRepository:
    def __init__(self, db: Session, tenant_id: str):
        self.db = db
        self.tenant_id = tenant_id
    
    def get_by_id(self, order_id: int):
        return self.db.query(Order).filter(
            Order.id == order_id,
            Order.tenant_id == self.tenant_id  # Never forget this!
        ).first()
    
    def list_all(self, skip: int = 0, limit: int = 100):
        return self.db.query(Order).filter(
            Order.tenant_id == self.tenant_id  # Never forget this!
        ).offset(skip).limit(limit).all()
    
    def create(self, order_data: dict):
        order = Order(
            **order_data,
            tenant_id=self.tenant_id  # Automatically set from repository
        )
        self.db.add(order)
        self.db.commit()
        self.db.refresh(order)
        return order

# 4. Service Layer: Pass tenant_id to repository
class OrderService:
    def __init__(self, db: Session, tenant_id: str):
        self.repository = OrderRepository(db, tenant_id)
    
    def get_order(self, order_id: int):
        order = self.repository.get_by_id(order_id)
        if not order:
            raise HTTPException(404, "Order not found")
        return order

# 5. Endpoint Layer: Inject tenant_id from request state
from fastapi import Depends

def get_tenant_id(request: Request) -> str:
    """Dependency to extract tenant_id from request state"""
    return request.state.tenant_id

def get_order_service(
    tenant_id: str = Depends(get_tenant_id),
    db: Session = Depends(get_db)
) -> OrderService:
    """Dependency to create OrderService with tenant isolation"""
    return OrderService(db, tenant_id)

@app.get("/orders/{order_id}")
async def get_order(
    order_id: int,
    service: OrderService = Depends(get_order_service)
):
    # Service already has tenant_id, no way to forget filtering
    return service.get_order(order_id)

PostgreSQL Row-Level Security (Defense in Depth)

Even with application-level filtering, I added database-level protection:

-- Enable row-level security on orders table
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;

-- Create policy: users can only see their tenant's data
CREATE POLICY tenant_isolation_policy ON orders
    USING (tenant_id = current_setting('app.current_tenant_id')::text);

-- Create policy for inserts
CREATE POLICY tenant_insert_policy ON orders
    FOR INSERT
    WITH CHECK (tenant_id = current_setting('app.current_tenant_id')::text);

-- Similar for other tables
ALTER TABLE products ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation_policy ON products
    USING (tenant_id = current_setting('app.current_tenant_id')::text);

ALTER TABLE payments ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation_policy ON payments
    USING (tenant_id = current_setting('app.current_tenant_id')::text);

# config/database.py
from sqlalchemy import event, text

@event.listens_for(SessionLocal, "after_begin")
def set_tenant_context(session, transaction, connection):
    """Set tenant context for PostgreSQL RLS"""
    if hasattr(session, 'tenant_id'):
        connection.execute(
            text(f"SET LOCAL app.current_tenant_id = '{session.tenant_id}'")
        )

# Usage
def get_tenant_db(tenant_id: str = Depends(get_tenant_id)):
    db = SessionLocal()
    db.tenant_id = tenant_id  # Set tenant_id on session
    try:
        yield db
    finally:
        db.close()

# Now even if you forget to filter by tenant_id, PostgreSQL won't return other tenants' data
@app.get("/orders")
async def get_orders(db: Session = Depends(get_tenant_db)):
    # Even without explicit filter, RLS ensures only current tenant's orders returned
    orders = db.query(Order).all()
    return orders

This saved me multiple times! When I forgot to add tenant filters during rapid development, RLS prevented data leaks.

Redis Cache with Tenant Prefixes

Caching adds another layer where tenant isolation is critical:

# services/cache.py
import redis
from typing import Optional, Any
import json

class TenantAwareCache:
    """Redis cache with automatic tenant prefixing"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def _make_key(self, tenant_id: str, key: str) -> str:
        """Prefix all keys with tenant_id"""
        return f"tenant:{tenant_id}:{key}"
    
    def get(self, tenant_id: str, key: str) -> Optional[Any]:
        """Get value from cache"""
        cache_key = self._make_key(tenant_id, key)
        value = self.redis.get(cache_key)
        
        if value:
            return json.loads(value)
        return None
    
    def set(self, tenant_id: str, key: str, value: Any, ttl: int = 3600):
        """Set value in cache with TTL"""
        cache_key = self._make_key(tenant_id, key)
        self.redis.setex(cache_key, ttl, json.dumps(value))
    
    def delete(self, tenant_id: str, key: str):
        """Delete value from cache"""
        cache_key = self._make_key(tenant_id, key)
        self.redis.delete(cache_key)
    
    def clear_tenant(self, tenant_id: str):
        """Clear all cache for a tenant"""
        pattern = f"tenant:{tenant_id}:*"
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys)

# Usage in services
class ProductService:
    def __init__(self, db: Session, tenant_id: str, cache: TenantAwareCache):
        self.db = db
        self.tenant_id = tenant_id
        self.cache = cache
    
    def get_product(self, product_id: int):
        # Try cache first
        cache_key = f"product:{product_id}"
        cached = self.cache.get(self.tenant_id, cache_key)
        
        if cached:
            return cached
        
        # Not in cache, query database
        product = self.db.query(Product).filter(
            Product.id == product_id,
            Product.tenant_id == self.tenant_id
        ).first()
        
        if product:
            # Cache for 1 hour
            self.cache.set(self.tenant_id, cache_key, product.dict(), ttl=3600)
        
        return product

Preventing Cross-Tenant Data Leaks: Comprehensive Checklist

Here's my checklist to prevent the nightmare I experienced:

1. Request Level

• Extract tenant_id from header

• Validate tenant exists

• Extract tenant_id from JWT

• Verify header tenant_id matches JWT tenant_id

• Store tenant_id in request context

2. Service Level

• Pass tenant_id to all service constructors

• Services accept tenant_id as first parameter

• Services pass tenant_id to repositories

3. Repository Level

• Every query includes WHERE tenant_id = ?

• Every insert sets tenant_id field

• Use repository base class to enforce filtering

4. Database Level

• Enable PostgreSQL Row-Level Security

• Set tenant context in database session

• Test RLS policies thoroughly

5. Cache Level

• Prefix all cache keys with tenant_id

• Never cache cross-tenant data

• Clear tenant cache on tenant deletion

6. Testing Level

• Test with multiple tenants

• Verify isolation in integration tests

• Audit all queries for tenant filtering

Automated Testing for Tenant Isolation

# tests/test_tenant_isolation.py
import pytest
from fastapi.testclient import TestClient

def test_order_isolation_between_tenants():
    """Ensure tenant A cannot see tenant B's orders"""
    
    # Create order for tenant A
    client_a = TestClient(app)
    response_a = client_a.post(
        "/orders",
        headers={
            "x-tenant-id": "tenant_a",
            "authorization": f"Bearer {tenant_a_token}"
        },
        json={"items": [{"product_id": 1, "quantity": 2}]}
    )
    assert response_a.status_code == 201
    order_a_id = response_a.json()["id"]
    
    # Create order for tenant B
    client_b = TestClient(app)
    response_b = client_b.post(
        "/orders",
        headers={
            "x-tenant-id": "tenant_b",
            "authorization": f"Bearer {tenant_b_token}"
        },
        json={"items": [{"product_id": 2, "quantity": 1}]}
    )
    assert response_b.status_code == 201
    
    # Tenant A tries to access their orders - should only see their order
    response_a_orders = client_a.get(
        "/orders",
        headers={
            "x-tenant-id": "tenant_a",
            "authorization": f"Bearer {tenant_a_token}"
        }
    )
    orders_a = response_a_orders.json()
    assert len(orders_a) == 1
    assert orders_a[0]["id"] == order_a_id
    
    # Tenant A tries to access tenant B's order directly - should fail
    response_forbidden = client_a.get(
        f"/orders/{order_b_id}",
        headers={
            "x-tenant-id": "tenant_a",
            "authorization": f"Bearer {tenant_a_token}"
        }
    )
    assert response_forbidden.status_code == 404  # Not found, not 403
    
    # Tenant A tries to use tenant B's header with their token - should fail
    response_mismatch = client_a.get(
        "/orders",
        headers={
            "x-tenant-id": "tenant_b",  # Wrong tenant!
            "authorization": f"Bearer {tenant_a_token}"  # Tenant A token
        }
    )
    assert response_mismatch.status_code == 403  # Tenant mismatch

def test_cache_isolation():
    """Ensure cached data is tenant-isolated"""
    cache = TenantAwareCache(redis_client)
    
    # Cache data for tenant A
    cache.set("tenant_a", "product:1", {"name": "Pizza", "price": 10.00})
    
    # Cache data for tenant B (same key!)
    cache.set("tenant_b", "product:1", {"name": "Burger", "price": 8.00})
    
    # Verify isolation
    product_a = cache.get("tenant_a", "product:1")
    product_b = cache.get("tenant_b", "product:1")
    
    assert product_a["name"] == "Pizza"
    assert product_b["name"] == "Burger"

The Real Bug: How One Tenant Saw Another's Data

Let me share the exact bug that caused the Friday incident:

# chatbot/orchestrator.py - THE BUG
class ChatbotOrchestrator:
    async def get_dashboard_summary(self, tenant_id: str):
        """Aggregate data from all services"""
        
        # Get orders (CORRECT - includes tenant_id)
        orders = await self.pos_core_service.get_orders(tenant_id)
        
        # Get inventory (BUG - forgot tenant_id!)
        inventory = await self.inventory_service.get_all_products()
        
        # Get payments (CORRECT)
        payments = await self.payment_service.get_payments(tenant_id)
        
        return {
            "orders": orders,
            "inventory": inventory,  # This returned ALL tenants' products!
            "payments": payments
        }

# inventory/service.py - THE MISSING FILTER
class InventoryService:
    def get_all_products(self):
        # BUG: No tenant filtering!
        return self.db.query(Product).all()  # Returns products from ALL tenants

# The fix
class InventoryService:
    def get_all_products(self, tenant_id: str):
        # FIXED: Filter by tenant
        return self.db.query(Product).filter(
            Product.tenant_id == tenant_id
        ).all()

The chatbot showed aggregated data, so it wasn't immediately obvious. But when a tenant expanded the inventory section, they saw products from other restaurants.

Lesson: When aggregating data from multiple services, verify each service call includes tenant_id.

Performance Considerations

Multi-tenancy adds overhead. Here's how to optimize:

1. Index Strategy

-- Composite indexes with tenant_id first
CREATE INDEX ix_orders_tenant_date ON orders(tenant_id, created_at DESC);
CREATE INDEX ix_products_tenant_category ON products(tenant_id, category);

-- Partition large tables by tenant_id (PostgreSQL 10+)
CREATE TABLE orders (
    id INTEGER,
    tenant_id VARCHAR(50),
    order_number VARCHAR(50),
    -- other columns
) PARTITION BY LIST (tenant_id);

-- Create partition per tenant
CREATE TABLE orders_tenant_a PARTITION OF orders FOR VALUES IN ('tenant_a');
CREATE TABLE orders_tenant_b PARTITION OF orders FOR VALUES IN ('tenant_b');

2. Connection Pooling

# config/database.py
from sqlalchemy.pool import QueuePool

engine = create_engine(
    settings.DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,  # Shared across all tenants
    max_overflow=40,
    pool_pre_ping=True,
    pool_recycle=3600  # Recycle connections after 1 hour
)

# Don't create separate pools per tenant (unless Strategy 1)
# Shared pool is much more efficient

3. Query Optimization

# Bad: N+1 query problem
def get_orders_with_products(tenant_id: str):
    orders = db.query(Order).filter(Order.tenant_id == tenant_id).all()
    
    for order in orders:
        # N additional queries!
        order.products = db.query(Product).filter(
            Product.id.in_(order.product_ids),
            Product.tenant_id == tenant_id
        ).all()
    
    return orders

# Good: Single query with join
from sqlalchemy.orm import joinedload

def get_orders_with_products(tenant_id: str):
    return db.query(Order).filter(
        Order.tenant_id == tenant_id
    ).options(
        joinedload(Order.products)
    ).all()

Key Learnings

1. Tenant isolation must be enforced at every layer

   • Middleware, services, repositories, database, cache

   • Defense in depth prevents catastrophic leaks

2. Always verify token tenant matches header tenant

   • Prevents malicious users from accessing other tenants

   • Simple check, massive security benefit

3. Use PostgreSQL Row-Level Security as safety net

   • Catches bugs before they become data breaches

   • Small performance cost, huge security gain

4. Prefix all cache keys with tenant_id

   • Easy to forget, easy to exploit

   • Automated testing is critical

5. Test with multiple tenants from day one

   • Don't add multi-tenancy later

   • Design for it from the start

Common Mistakes

1. Forgetting tenant filter in queries

   • Use repository pattern to enforce filtering

   • Enable PostgreSQL RLS

2. Not validating tenant_id from JWT

   • User could send arbitrary tenant_id in header

   • Always verify against JWT claims

3. Caching data without tenant prefix

   • All tenants share Redis

   • Unprefixed keys leak data

4. Using separate database without connection pooling strategy

   • Too many connections kill database

   • Need connection pool manager

When to Use Each Strategy

Separate Database:

• Enterprise customers paying premium

• Compliance requirements (HIPAA, GDPR)

• < 100 tenants (connection management gets complex)

Separate Schemas:

• Need strong isolation

• Medium tenant count (100-1000)

• Want to balance cost and isolation

Shared Tables (Discriminator Column):

• High tenant count (1000+)

• Need best performance

• Can enforce isolation at application layer

• My choice for POS system

Next Steps

Now that you understand multi-tenant architecture, the next article dives into service layer architecture—how to structure your code within each service for maintainability and testability.

We'll cover:

• Three-layer architecture (Domain, Application, Infrastructure)

• Dependency injection with FastAPI

• Repository pattern implementation

• Why layers matter for testing

Next Article: 04-service-layer-architecture.md - Learn how to structure services so they're testable, maintainable, and ready to scale.

---

Remember: Multi-tenancy isn't a feature you add later. It's a fundamental architectural constraint. Design for it from day one, test it thoroughly, and never trust yourself to remember that WHERE clause.