Multi-Tenancy Patterns

Build secure, scalable SaaS applications with tenant isolation strategies, data partitioning patterns, and production-ready multi-tenancy implementations

The SaaS Multi-Tenancy Challenge

Multi-tenancy is the architectural pattern where a single instance of a software application serves multiple customers (tenants). Each tenant's data is isolated and invisible to other tenants, despite sharing the same infrastructure, application code, and often the same database. This is the foundation of modern SaaS economics, instead of deploying separate infrastructure for each customer, you serve thousands of tenants from a shared platform, dramatically reducing operational costs while maintaining security boundaries. Every major SaaS platform (Slack, Salesforce, AWS, Shopify) uses multi-tenancy, but the implementation details vary based on isolation requirements and scale.

Cross-Reference: For API-level analytics and cost attribution patterns for multi-tenant systems, see Web APIs Lesson 18: API Analytics & Usage Tracking, which covers metering and billing for SaaS applications.

Quick Navigation

Tenant Isolation Models

Context Propagation

Data Leakage Prevention

Rate Limiting & Quotas

Data Partitioning

Tenant Configuration

Scaling Strategies

Security & Compliance

Real-World Examples

Best Practices

Practice Exercise

The Core Tension: Resource Sharing vs Isolation

Multi-tenancy creates a fundamental trade-off between resource efficiency and isolation guarantees:

Resource Sharing Benefits

✓ Lower costs per tenant✓ Simplified ops✓ Efficient utilization

Serve thousands of tenants from shared infrastructure
Single deployment and update process
Better resource utilization across tenants

Isolation Requirements

! Data leakage risk! Noisy neighbors! Compliance

Prevent cross-tenant data access
Isolate performance (one tenant can't impact others)
Meet regulatory requirements (GDPR, HIPAA)

The Balance

The right multi-tenancy strategy depends on your business model. B2C SaaS with thousands of small customers benefits from aggressive resource sharing. B2B enterprise SaaS may require dedicated infrastructure per customer. Most successful SaaS platforms use graduated isolation, start with shared resources, migrate high-value customers to dedicated infrastructure.

Tenant Isolation Models

There are four primary isolation models, each with different trade-offs between cost, isolation, and complexity:

1. Separate Database per Tenant (Silo Model)

Each tenant gets a dedicated database. Maximum isolation, but highest operational overhead.

# Database per tenant with connection routing
from contextvars import ContextVar
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

current_tenant = ContextVar("current_tenant", default=None)

# Dedicated database per tenant
TENANT_ENGINES = {
    "acme": create_engine("postgresql://host/acme_db"),
    "globex": create_engine("postgresql://host/globex_db"),
    "initech": create_engine("postgresql://host/initech_db"),
}

def get_tenant_db():
    tenant_id = current_tenant.get()
    if not tenant_id:
        raise ValueError("No tenant context")

    engine = TENANT_ENGINES.get(tenant_id)
    if not engine:
        raise ValueError(f"Unknown tenant: {tenant_id}")

    SessionLocal = sessionmaker(bind=engine)
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Usage in FastAPI
@app.get("/api/documents")
async def list_documents(db: Session = Depends(get_tenant_db)):
    # Automatically queries the correct tenant database
    return db.query(Document).all()

✓ Strongest isolation✓ Easy compliance✗ High cost✗ Complex ops

2. Shared Database, Separate Schema per Tenant

All tenants share one database, but each has a dedicated schema (namespace). Good balance of isolation and efficiency.

# PostgreSQL schemas per tenant
-- Database: saas_db
-- Schemas: acme, globex, initech

CREATE SCHEMA acme;
CREATE TABLE acme.documents (
    id SERIAL PRIMARY KEY,
    title VARCHAR(255),
    content TEXT
);

CREATE SCHEMA globex;
CREATE TABLE globex.documents (
    id SERIAL PRIMARY KEY,
    title VARCHAR(255),
    content TEXT
);

# SQLAlchemy with schema routing
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("postgresql://host/saas_db")

def get_tenant_session(tenant_id: str):
    # Set search_path to tenant schema
    connection = engine.connect()
    connection.execute(text(f"SET search_path TO {identifier}"),
                       {"identifier": tenant_id})

    session = Session(bind=connection)
    return session

# Usage
session = get_tenant_session("acme")
# Queries acme.documents
docs = session.query(Document).all()

session = get_tenant_session("globex")
# Queries globex.documents
docs = session.query(Document).all()

✓ Good isolation✓ Shared resources~ Medium complexity~ Medium cost

3. Shared Database, Shared Schema (Pool Model with tenant_id)

All tenants share the same database and tables. Every table has a tenant_id column. Most cost-effective but requires careful filtering.

# Shared schema with tenant_id discriminator
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    tenant_id VARCHAR(50) NOT NULL,
    title VARCHAR(255),
    content TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);

-- CRITICAL: Index tenant_id as first column for performance
CREATE INDEX idx_documents_tenant ON documents(tenant_id, created_at);

# Tenant-aware session with automatic filtering
from sqlalchemy import event
from sqlalchemy.orm import Session

class TenantSession(Session):
    def __init__(self, bind, tenant_id: str):
        super().__init__(bind=bind)
        self.tenant_id = tenant_id

    def execute(self, *args, **kwargs):
        # Auto-add tenant_id filter to all queries
        return super().execute(*args, **kwargs)

# Usage
session = TenantSession(engine, tenant_id="acme")
docs = session.query(Document).all()  # Automatically filtered

# PostgreSQL Row-Level Security for defense-in-depth
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation_policy ON documents
    USING (tenant_id = current_setting('app.current_tenant')::VARCHAR);

-- Set tenant context in session
SET app.current_tenant = 'acme';
SELECT * FROM documents;  -- Only returns acme's documents

✓ Lowest cost✓ Simplest ops✗ Data leakage risk✗ Requires RLS

Critical Security Warning

With shared schema, every single query must include the tenant_id filter. A single missing filter can leak data across tenants. Use defense-in-depth: application filtering + database Row-Level Security (RLS) + regular audits.

4. Hybrid Approaches (Graduated Isolation)

Many production systems use graduated isolation: small tenants share databases, large enterprise customers get dedicated infrastructure.

# Hybrid tenant routing based on tier
from enum import Enum

class TenantTier(Enum):
    FREE = "free"           # Shared database, shared schema
    PRO = "pro"             # Shared database, dedicated schema
    ENTERPRISE = "enterprise"  # Dedicated database

TENANT_CONFIG = {
    "acme": {"tier": TenantTier.ENTERPRISE, "db": "acme_db"},
    "startup1": {"tier": TenantTier.PRO, "schema": "startup1"},
    "freemium_user": {"tier": TenantTier.FREE},
}

def get_tenant_session(tenant_id: str):
    config = TENANT_CONFIG[tenant_id]

    if config["tier"] == TenantTier.ENTERPRISE:
        # Dedicated database
        engine = create_engine(f"postgresql://host/{config['db']}")
        return Session(bind=engine)

    elif config["tier"] == TenantTier.PRO:
        # Dedicated schema
        connection = shared_engine.connect()
        connection.execute(text(f"SET search_path TO {config['schema']}"))
        return Session(bind=connection)

    else:  # FREE tier
        # Shared schema with tenant_id filtering
        return TenantSession(shared_engine, tenant_id=tenant_id)

✓ Optimized per tier✓ Flexible✗ Most complex

Isolation Model Comparison

Model	Isolation	Cost Efficiency	Operational Complexity	Best Use Case
Separate Database	Excellent	Low	High	Enterprise, compliance-heavy
Separate Schema	Very Good	Medium	Medium	Mid-market B2B SaaS
Shared Schema	Good (with RLS)	Excellent	Low	High-volume B2C, freemium
Hybrid	Varies	Optimized	High	Multi-tier (free to enterprise)

Tenant Context Propagation

Once you've chosen an isolation model, you need a reliable way to extract and propagate the tenant identifier throughout your application stack.

Extracting Tenant ID from Requests

# Three methods to identify tenants
from fastapi import FastAPI, Request, Header, HTTPException
import jwt

app = FastAPI()

# Method 1: Subdomain-based
# acme.myapp.com -> tenant_id = "acme"
def extract_tenant_from_subdomain(request: Request) -> str:
    host = request.headers.get("host", "")
    parts = host.split(".")
    if len(parts) >= 3:
        return parts[0]  # "acme" from "acme.myapp.com"
    raise HTTPException(400, "Invalid tenant subdomain")

# Method 2: JWT-based
# Extract tenant_id from authenticated JWT token
def extract_tenant_from_jwt(authorization: str = Header(...)) -> str:
    try:
        token = authorization.replace("Bearer ", "")
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload["tenant_id"]
    except Exception:
        raise HTTPException(401, "Invalid token")

# Method 3: Custom header
# X-Tenant-ID: acme
def extract_tenant_from_header(x_tenant_id: str = Header(...)) -> str:
    if not x_tenant_id:
        raise HTTPException(400, "Missing X-Tenant-ID header")
    return x_tenant_id

FastAPI Middleware for Tenant Context

# Middleware to extract tenant once per request
from contextvars import ContextVar
from starlette.middleware.base import BaseHTTPMiddleware

# Context variable for tenant ID (async-safe)
current_tenant: ContextVar[str] = ContextVar("current_tenant", default=None)

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract tenant from subdomain
        host = request.headers.get("host", "")
        parts = host.split(".")

        if len(parts) >= 3:
            tenant_id = parts[0]
        elif "x-tenant-id" in request.headers:
            # Fallback to header (for local dev)
            tenant_id = request.headers["x-tenant-id"]
        else:
            raise HTTPException(400, "Cannot determine tenant")

        # Validate tenant exists
        if tenant_id not in VALID_TENANTS:
            raise HTTPException(404, "Tenant not found")

        # Set context for this request
        token = current_tenant.set(tenant_id)

        try:
            response = await call_next(request)
            return response
        finally:
            # Clean up context
            current_tenant.reset(token)

app.add_middleware(TenantMiddleware)

# Now all endpoints can access tenant_id
@app.get("/api/documents")
async def list_documents():
    tenant_id = current_tenant.get()
    # Use tenant_id for queries
    return {"tenant": tenant_id}

Important: Don't Use threading.local with Async

In async frameworks like FastAPI, never use threading.local for storing tenant context. Async tasks can switch contexts within a single thread, causing tenant context to leak. Always use contextvars.ContextVar which is async-safe.

Data Leakage Prevention

Data leakage is the catastrophic failure mode in multi-tenancy. A single missing WHERE tenant_id = ? can expose one customer's data to another. Defense-in-depth is critical.

PostgreSQL Row-Level Security (RLS)

-- PostgreSQL RLS: Database-level tenant isolation
-- Even if application forgets tenant_id, database enforces it

-- Enable RLS on all tenant tables
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE users ENABLE ROW LEVEL SECURITY;

-- Create policy: Only return rows for current tenant
CREATE POLICY tenant_isolation_policy ON documents
    USING (tenant_id = current_setting('app.current_tenant')::VARCHAR);

CREATE POLICY tenant_isolation_policy ON orders
    USING (tenant_id = current_setting('app.current_tenant')::VARCHAR);

-- Set tenant context at session start
-- Application must set this before any queries
SET app.current_tenant = 'acme';

-- Now all queries are automatically filtered
SELECT * FROM documents;  -- Only returns acme's documents
SELECT * FROM orders;      -- Only returns acme's orders

-- Even malicious query without tenant_id is safe
SELECT * FROM documents WHERE id = 123;
-- RLS policy still applies, returns empty result if not acme's doc

# Python: Set RLS context
def get_tenant_db(tenant_id: str):
    connection = engine.raw_connection()
    cursor = connection.cursor()

    # Set tenant context for this session
    cursor.execute(
        "SET app.current_tenant = %s",
        (tenant_id,)
    )

    return connection

RLS Benefits

Row-Level Security is a last line of defense. Even if your application has a bug and forgets to add tenant_id to a query, RLS ensures the database only returns that tenant's data. This is defense-in-depth at the database layer.

Preventing IDOR (Insecure Direct Object Reference)

# IDOR vulnerability: User can access other tenant's data by guessing IDs
# BAD: Only checks if document exists, not if it belongs to current tenant
@app.get("/api/documents/{doc_id}")
async def get_document_bad(doc_id: int):
    doc = db.query(Document).filter(Document.id == doc_id).first()
    return doc  # ❌ Returns ANY document, even from other tenants!

# GOOD: Always verify tenant ownership
@app.get("/api/documents/{doc_id}")
async def get_document_good(doc_id: int):
    tenant_id = current_tenant.get()

    doc = db.query(Document).filter(
        Document.id == doc_id,
        Document.tenant_id == tenant_id  # ✓ Verify ownership
    ).first()

    if not doc:
        raise HTTPException(404, "Document not found")

    return doc

# BETTER: Use composite keys that include tenant_id
# Document ID: "acme:doc-123" instead of just "123"
@app.get("/api/documents/{doc_id}")
async def get_document_best(doc_id: str):
    tenant_id = current_tenant.get()

    # doc_id format: "tenant_id:uuid"
    parts = doc_id.split(":")
    if len(parts) != 2 or parts[0] != tenant_id:
        raise HTTPException(403, "Access denied")

    doc = db.query(Document).filter(Document.id == doc_id).first()
    return doc  # Safe: ID itself includes tenant verification

Defense-in-Depth Checklist

Layer 1 - Application: Always include tenant_id in WHERE clauses
Layer 2 - Database RLS: PostgreSQL policies enforce tenant isolation at DB level
Layer 3 - Composite Keys: Include tenant_id in primary keys (prevents IDOR)
Layer 4 - Auditing: Log all data access with tenant_id for forensics
Layer 5 - Penetration Testing: Regularly test for tenant leakage (automated + manual)
Layer 6 - Monitoring: Alert on queries missing tenant_id filter

Rate Limiting & Resource Quotas

In shared infrastructure, one tenant can monopolize resources (CPU, database connections, API calls) and degrade performance for others. This is the noisy neighbor problem. Rate limiting and quotas prevent resource exhaustion.

Redis-Based Rate Limiting

# Per-tenant rate limiting with Redis
import redis
from fastapi import HTTPException

redis_client = redis.Redis(host='localhost', decode_responses=True)

RATE_LIMITS = {
    "free": {"requests_per_minute": 60},
    "pro": {"requests_per_minute": 600},
    "enterprise": {"requests_per_minute": 6000},
}

def check_rate_limit(tenant_id: str, tier: str):
    """
    Token bucket rate limiting per tenant.
    Returns True if request allowed, raises 429 if rate limited.
    """
    limit = RATE_LIMITS[tier]["requests_per_minute"]
    key = f"rate_limit:{tenant_id}"

    # Increment request count
    current = redis_client.incr(key)

    # Set expiry on first request (fixed window)
    if current == 1:
        redis_client.expire(key, 60)

    # Check if over limit
    if current > limit:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded: {limit} req/min for {tier} tier"
        )

    return True

# Usage in middleware
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    tenant_id = current_tenant.get()
    tier = get_tenant_tier(tenant_id)

    check_rate_limit(tenant_id, tier)

    response = await call_next(request)
    return response

Resource Quotas (Storage, Compute)

# Enforce storage quotas per tenant
STORAGE_QUOTAS = {
    "free": 1 * 1024 * 1024 * 1024,      # 1 GB
    "pro": 100 * 1024 * 1024 * 1024,     # 100 GB
    "enterprise": 1000 * 1024 * 1024 * 1024,  # 1 TB
}

def check_storage_quota(tenant_id: str, file_size: int):
    """Check if uploading file would exceed storage quota"""
    tier = get_tenant_tier(tenant_id)
    quota = STORAGE_QUOTAS[tier]

    # Get current usage from database
    current_usage = db.query(
        func.sum(File.size)
    ).filter(
        File.tenant_id == tenant_id
    ).scalar() or 0

    if current_usage + file_size > quota:
        raise HTTPException(
            status_code=403,
            detail=f"Storage quota exceeded. Limit: {quota / (1024**3):.1f} GB"
        )

    return True

@app.post("/api/files/upload")
async def upload_file(file: UploadFile):
    tenant_id = current_tenant.get()
    content = await file.read()
    file_size = len(content)

    # Check quota before uploading
    check_storage_quota(tenant_id, file_size)

    # Upload to S3
    s3_key = f"{tenant_id}/{file.filename}"
    s3_client.put_object(Bucket="uploads", Key=s3_key, Body=content)

    # Record in database
    db.add(File(tenant_id=tenant_id, name=file.filename, size=file_size))
    db.commit()

    return {"uploaded": file.filename}

Preventing Noisy Neighbor (Database Connection Pooling)

# Connection pool limits per tenant to prevent one tenant from monopolizing DB
from sqlalchemy import create_engine, pool

# Separate connection pools per tenant tier
FREE_TIER_POOL = pool.QueuePool(
    lambda: psycopg2.connect(DB_URL),
    max_overflow=5,
    pool_size=5,  # Max 5 concurrent connections
    timeout=30
)

ENTERPRISE_TIER_POOL = pool.QueuePool(
    lambda: psycopg2.connect(DB_URL),
    max_overflow=50,
    pool_size=50,  # Max 50 concurrent connections
    timeout=30
)

def get_connection_pool(tenant_id: str):
    tier = get_tenant_tier(tenant_id)

    if tier == "free":
        return FREE_TIER_POOL
    elif tier == "enterprise":
        return ENTERPRISE_TIER_POOL
    else:
        return PRO_TIER_POOL

# Use tenant-specific pool
def get_db_connection(tenant_id: str):
    pool = get_connection_pool(tenant_id)
    return pool.connect()

Noisy Neighbor Impact

Without rate limiting, a single tenant running a script with 1000 concurrent requests can exhaust your database connection pool, causing timeouts for all other tenants. This is why per-tenant quotas are critical in shared infrastructure.

Data Partitioning Strategies

As you scale to hundreds of thousands of tenants, a single database table becomes a bottleneck. Data partitioning distributes data across multiple tables, databases, or shards based on tenant_id.

Horizontal Sharding by Tenant

# Shard tenants across multiple databases using consistent hashing
import hashlib

# Shard mapping: Hash tenant_id to determine database
SHARDS = {
    0: create_engine("postgresql://host/shard_0"),
    1: create_engine("postgresql://host/shard_1"),
    2: create_engine("postgresql://host/shard_2"),
    3: create_engine("postgresql://host/shard_3"),
}

def get_shard_for_tenant(tenant_id: str) -> int:
    """Consistent hashing to determine shard number"""
    hash_value = int(hashlib.md5(tenant_id.encode()).hexdigest(), 16)
    return hash_value % len(SHARDS)

def get_tenant_engine(tenant_id: str):
    """Get database engine for this tenant's shard"""
    shard_id = get_shard_for_tenant(tenant_id)
    return SHARDS[shard_id]

# Usage
def get_documents(tenant_id: str):
    engine = get_tenant_engine(tenant_id)
    with Session(bind=engine) as session:
        return session.query(Document).filter(
            Document.tenant_id == tenant_id
        ).all()

# Benefits:
# - Distributes load across 4 databases
# - Each shard handles ~25% of tenants
# - Can scale by adding more shards

PostgreSQL Declarative Partitioning

-- PostgreSQL 10+ native partitioning by tenant_id
-- Automatically routes queries to correct partition

-- Create partitioned table
CREATE TABLE documents (
    id SERIAL,
    tenant_id VARCHAR(50) NOT NULL,
    title VARCHAR(255),
    content TEXT,
    created_at TIMESTAMP DEFAULT NOW()
) PARTITION BY LIST (tenant_id);

-- Create partition per tenant (or per shard)
CREATE TABLE documents_acme PARTITION OF documents
    FOR VALUES IN ('acme');

CREATE TABLE documents_globex PARTITION OF documents
    FOR VALUES IN ('globex');

CREATE TABLE documents_initech PARTITION OF documents
    FOR VALUES IN ('initech');

-- Queries automatically route to correct partition
SELECT * FROM documents WHERE tenant_id = 'acme';
-- Postgres only scans documents_acme partition

-- Insert also routes to correct partition
INSERT INTO documents (tenant_id, title, content)
VALUES ('acme', 'Report', 'Q4 earnings');
-- Goes to documents_acme partition

-- Benefits:
-- ✓ Faster queries (smaller tables to scan)
-- ✓ Easier to drop tenant data (DROP TABLE documents_acme)
-- ✓ Can move partitions to different tablespaces/servers

Indexing Strategies for Multi-Tenant Tables

-- CRITICAL: tenant_id must be FIRST column in all indexes
-- This allows index-only scans for single-tenant queries

-- BAD: Index without tenant_id prefix
CREATE INDEX idx_documents_created ON documents(created_at);
-- Query: SELECT * FROM documents WHERE tenant_id = 'acme' AND created_at > '2024-01-01'
-- Problem: Index scan can't filter by tenant_id, must scan entire index

-- GOOD: Composite index with tenant_id first
CREATE INDEX idx_documents_tenant_created ON documents(tenant_id, created_at);
-- Same query now uses index efficiently, only scans acme's portion

-- Pattern: Always prefix indexes with tenant_id
CREATE INDEX idx_documents_tenant_status ON documents(tenant_id, status);
CREATE INDEX idx_orders_tenant_date ON orders(tenant_id, order_date);
CREATE INDEX idx_users_tenant_email ON users(tenant_id, email);

-- PostgreSQL: Use partial indexes for large tenants
CREATE INDEX idx_documents_acme_created ON documents(created_at)
    WHERE tenant_id = 'acme';
-- Smaller index, faster queries for this specific high-volume tenant

Partitioning Performance Impact

A properly partitioned table with 1M tenants can be 100x faster than a monolithic table. Query that scans 1 billion rows becomes a scan of 1 million rows (per tenant). Index size also shrinks, improving cache hit rates.

Tenant Configuration & Customization

Different tenants need different features, branding, and settings. Store tenant-specific configuration in a flexible schema that supports per-tenant customization without code changes.

Feature Flags Per Tenant

# Feature flags: Enable/disable features per tenant
CREATE TABLE tenant_features (
    tenant_id VARCHAR(50) NOT NULL,
    feature_key VARCHAR(100) NOT NULL,
    enabled BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (tenant_id, feature_key)
);

# Python: Feature flag checks
def is_feature_enabled(tenant_id: str, feature: str) -> bool:
    """Check if feature is enabled for this tenant"""
    result = db.query(TenantFeature).filter(
        TenantFeature.tenant_id == tenant_id,
        TenantFeature.feature_key == feature
    ).first()

    return result.enabled if result else False

# Usage in endpoints
@app.post("/api/advanced-analytics")
async def advanced_analytics():
    tenant_id = current_tenant.get()

    if not is_feature_enabled(tenant_id, "advanced_analytics"):
        raise HTTPException(
            403,
            "Advanced analytics not available in your plan"
        )

    # Feature code here...
    return {"analytics": "..."}

# Feature flag examples:
# - api_v2_enabled
# - export_to_csv
# - advanced_analytics
# - white_label_branding
# - sso_integration
# - custom_domains

Tenant Branding & Customization

# Store branding configuration as JSONB in PostgreSQL
CREATE TABLE tenant_config (
    tenant_id VARCHAR(50) PRIMARY KEY,
    branding JSONB DEFAULT '{}'::jsonb,
    settings JSONB DEFAULT '{}'::jsonb,
    updated_at TIMESTAMP DEFAULT NOW()
);

# Example branding JSON
{
  "logo_url": "https://cdn.acme.com/logo.png",
  "primary_color": "#1e40af",
  "secondary_color": "#10b981",
  "company_name": "Acme Corporation",
  "custom_domain": "app.acme.com",
  "email_from_address": "noreply@acme.com",
  "support_email": "support@acme.com"
}

# Python: Load tenant branding
def get_tenant_branding(tenant_id: str) -> dict:
    """Get branding configuration for tenant"""
    config = db.query(TenantConfig).filter(
        TenantConfig.tenant_id == tenant_id
    ).first()

    return config.branding if config else DEFAULT_BRANDING

# Use in frontend
@app.get("/api/config")
async def get_config():
    tenant_id = current_tenant.get()
    branding = get_tenant_branding(tenant_id)

    return {
        "branding": branding,
        "features": get_tenant_features(tenant_id)
    }

# Frontend uses this to customize UI
# - Logo in navbar
# - Primary color for buttons
# - Company name in footer
# - Custom domain in emails

JSONB Settings with Validation

# Flexible tenant settings stored as JSONB
from pydantic import BaseModel, field_validator
from zoneinfo import available_timezones

class TenantSettings(BaseModel):
    """Schema for tenant settings with validation"""
    timezone: str = "UTC"
    date_format: str = "YYYY-MM-DD"
    currency: str = "USD"
    max_users: int = 10
    session_timeout_minutes: int = 30
    two_factor_required: bool = False
    ip_whitelist: list[str] = []

    @field_validator('timezone')
    @classmethod
    def validate_timezone(cls, v):
        # Validate timezone exists (Python 3.9+ stdlib)
        if v not in available_timezones():
            raise ValueError(f"Invalid timezone: {v}")
        return v

# Update tenant settings
@app.put("/api/admin/settings")
async def update_settings(settings: TenantSettings):
    tenant_id = current_tenant.get()

    # Validate settings against schema
    settings_dict = settings.model_dump()

    # Update JSONB column
    db.query(TenantConfig).filter(
        TenantConfig.tenant_id == tenant_id
    ).update({
        "settings": settings_dict,
        "updated_at": datetime.now(timezone.utc)
    })
    db.commit()

    return {"updated": True}

# Query JSONB fields
# PostgreSQL JSONB operators for efficient queries
SELECT * FROM tenant_config
WHERE settings->>'currency' = 'EUR';

SELECT * FROM tenant_config
WHERE (settings->>'max_users')::int > 100;

JSONB Benefits

JSONB in PostgreSQL gives you schema flexibility without sacrificing performance. You can index JSONB fields, query with operators, and add new settings without database migrations. Perfect for tenant-specific configuration.

Scaling Strategies

As your SaaS grows, you need strategies to scale from hundreds to millions of tenants while maintaining performance and cost efficiency.

Graduated Isolation (Tier-Based Architecture)

# Graduated isolation: Different infrastructure per tier
# Small tenants share resources, large tenants get dedicated

class TenantTier(Enum):
    FREE = "free"           # Pool: Shared DB, shared schema, aggressive rate limits
    STARTER = "starter"     # Pool: Shared DB, shared schema, moderate limits
    BUSINESS = "business"   # Bridge: Shared DB, dedicated schema
    ENTERPRISE = "enterprise"  # Silo: Dedicated database, dedicated compute

# Routing logic based on tier
def get_infrastructure_for_tenant(tenant_id: str):
    tier = db.query(Tenant).filter(Tenant.id == tenant_id).first().tier

    if tier == TenantTier.ENTERPRISE:
        # Dedicated infrastructure
        return {
            "database": f"postgresql://dedicated/{tenant_id}_db",
            "cache": f"redis://dedicated/{tenant_id}:6379",
            "storage": f"s3://{tenant_id}-bucket",
            "rate_limit": None,  # No limits for enterprise
        }

    elif tier == TenantTier.BUSINESS:
        # Shared database, dedicated schema
        return {
            "database": "postgresql://shared/multi_tenant_db",
            "schema": tenant_id,
            "cache": "redis://shared:6379",
            "storage": "s3://shared-bucket",
            "rate_limit": 10000,  # 10k req/min
        }

    else:  # FREE or STARTER
        # Fully shared, pool model
        return {
            "database": "postgresql://shared/multi_tenant_db",
            "schema": "public",
            "cache": "redis://shared:6379",
            "storage": "s3://shared-bucket",
            "rate_limit": 100 if tier == TenantTier.FREE else 1000,
        }

Tenant Migration (Moving Up Tiers)

# Migrate tenant from shared to dedicated infrastructure
# Zero-downtime migration process

class TenantMigrationService:
    def migrate_to_dedicated(self, tenant_id: str):
        """
        Migrate tenant from shared schema to dedicated database.
        Steps:
        1. Provision new dedicated database
        2. Copy data (pg_dump + restore)
        3. Switch routing (atomic update)
        4. Cleanup old data
        """
        print(f"Starting migration for {tenant_id}...")

        # Step 1: Provision dedicated database
        new_db_url = self.provision_database(tenant_id)

        # Step 2: Copy data from shared to dedicated
        self.copy_tenant_data(
            source_db="postgresql://shared/multi_tenant_db",
            dest_db=new_db_url,
            tenant_id=tenant_id
        )

        # Step 3: Switch routing (atomic update in tenant registry)
        db.query(Tenant).filter(Tenant.id == tenant_id).update({
            "tier": TenantTier.ENTERPRISE,
            "database_url": new_db_url,
            "migrated_at": datetime.now(timezone.utc)
        })
        db.commit()

        # Step 4: Cleanup old data (optional, can keep for rollback)
        # self.delete_tenant_data_from_shared(tenant_id)

        print(f"Migration complete for {tenant_id}")

    def copy_tenant_data(self, source_db, dest_db, tenant_id):
        """Copy all tenant data from source to destination"""
        # Export tenant data using COPY with filtering
        tables = ["documents", "orders", "users"]
        for table in tables:
            query = f"COPY (SELECT * FROM {table} WHERE tenant_id = '{tenant_id}') TO STDOUT"
            export_cmd = f'psql {source_db} -c "\{query}"'
            import_cmd = f'psql {dest_db} -c "COPY {table} FROM STDIN"'
            subprocess.run(
                f"{export_cmd} | {import_cmd}",
                shell=True, check=True
            )

Cost Allocation & Metering

# Track resource usage per tenant for cost allocation and billing
CREATE TABLE tenant_usage (
    tenant_id VARCHAR(50) NOT NULL,
    metric VARCHAR(50) NOT NULL,  -- 'api_requests', 'storage_gb', 'compute_hours'
    value BIGINT NOT NULL,
    timestamp TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (tenant_id, metric, timestamp)
);

# Python: Record usage metrics
def record_usage(tenant_id: str, metric: str, value: int):
    """Record usage metric for billing"""
    db.add(TenantUsage(
        tenant_id=tenant_id,
        metric=metric,
        value=value,
        timestamp=datetime.now(timezone.utc)
    ))
    db.commit()

# Track API requests (in middleware)
@app.middleware("http")
async def usage_tracking_middleware(request: Request, call_next):
    tenant_id = current_tenant.get()

    response = await call_next(request)

    # Record API request
    record_usage(tenant_id, "api_requests", 1)

    return response

# Track storage usage (on file upload)
@app.post("/api/files/upload")
async def upload_file(file: UploadFile):
    tenant_id = current_tenant.get()
    file_size = len(await file.read())

    # Record storage
    record_usage(tenant_id, "storage_bytes", file_size)

    # ... upload logic ...

# Monthly billing report
def generate_monthly_invoice(tenant_id: str, month: str):
    """Calculate costs based on usage"""
    usage = db.query(
        TenantUsage.metric,
        func.sum(TenantUsage.value).label('total')
    ).filter(
        TenantUsage.tenant_id == tenant_id,
        func.date_trunc('month', TenantUsage.timestamp) == month
    ).group_by(TenantUsage.metric).all()

    # Pricing
    costs = {
        "api_requests": 0.0001,  # $0.0001 per request
        "storage_gb": 0.10,       # $0.10 per GB
        "compute_hours": 2.00,    # $2 per hour
    }

    total = 0
    for metric, value in usage:
        if metric == "storage_bytes":
            value = value / (1024**3)  # Convert to GB
        total += value * costs.get(metric, 0)

    return {"tenant_id": tenant_id, "month": month, "total": total}

Usage-Based Pricing

Modern SaaS platforms use usage-based pricing (Stripe, AWS, Snowflake). Track every API call, storage byte, and compute minute per tenant. This data drives billing, cost optimization, and tier upgrades. Implement metering from day one.

Security & Compliance

Multi-tenant SaaS applications must meet stringent security and compliance requirements, especially for enterprise customers. Encryption, audit logging, and regulatory compliance are non-negotiable.

Encryption at Rest (AWS KMS per Tenant)

# Encrypt sensitive data with per-tenant KMS keys
import boto3
from base64 import b64encode, b64decode

kms_client = boto3.client('kms')

# Each tenant gets a dedicated KMS key
TENANT_KMS_KEYS = {
    "acme": "arn:aws:kms:us-east-1:123456789012:key/acme-key-id",
    "globex": "arn:aws:kms:us-east-1:123456789012:key/globex-key-id",
}

def encrypt_sensitive_data(tenant_id: str, plaintext: str) -> str:
    """Encrypt data with tenant-specific KMS key"""
    kms_key_id = TENANT_KMS_KEYS[tenant_id]

    response = kms_client.encrypt(
        KeyId=kms_key_id,
        Plaintext=plaintext.encode('utf-8')
    )

    # Store encrypted ciphertext in database
    ciphertext = b64encode(response['CiphertextBlob']).decode('utf-8')
    return ciphertext

def decrypt_sensitive_data(tenant_id: str, ciphertext: str) -> str:
    """Decrypt data with tenant-specific KMS key"""
    ciphertext_blob = b64decode(ciphertext.encode('utf-8'))

    response = kms_client.decrypt(
        CiphertextBlob=ciphertext_blob
    )

    return response['Plaintext'].decode('utf-8')

# Usage: Encrypt credit card, SSN, etc.
@app.post("/api/payment-methods")
async def add_payment_method(card_number: str):
    tenant_id = current_tenant.get()

    # Encrypt credit card with tenant's KMS key
    encrypted_card = encrypt_sensitive_data(tenant_id, card_number)

    db.add(PaymentMethod(
        tenant_id=tenant_id,
        encrypted_card_number=encrypted_card
    ))
    db.commit()

    return {"added": True}

Audit Logging (SOC2 / ISO 27001)

# Comprehensive audit logging for compliance
CREATE TABLE audit_log (
    id SERIAL PRIMARY KEY,
    tenant_id VARCHAR(50) NOT NULL,
    user_id VARCHAR(50),
    action VARCHAR(100) NOT NULL,  -- 'document.create', 'user.login', 'settings.update'
    resource_type VARCHAR(50),
    resource_id VARCHAR(100),
    ip_address INET,
    user_agent TEXT,
    request_id VARCHAR(100),
    metadata JSONB,
    timestamp TIMESTAMP DEFAULT NOW()
);

-- Index for compliance queries
CREATE INDEX idx_audit_tenant_time ON audit_log(tenant_id, timestamp);
CREATE INDEX idx_audit_action ON audit_log(action);

# Python: Audit logging decorator
from functools import wraps

def audit_log(action: str):
    """Decorator to automatically log actions"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            tenant_id = current_tenant.get()
            user_id = get_current_user_id()

            # Execute function
            result = await func(*args, **kwargs)

            # Log action
            db.add(AuditLog(
                tenant_id=tenant_id,
                user_id=user_id,
                action=action,
                resource_type=kwargs.get('resource_type'),
                resource_id=kwargs.get('resource_id'),
                ip_address=request.client.host,
                user_agent=request.headers.get('user-agent'),
                metadata={"args": str(args), "kwargs": str(kwargs)}
            ))
            db.commit()

            return result
        return wrapper
    return decorator

# Usage
@app.delete("/api/documents/{doc_id}")
@audit_log("document.delete")
async def delete_document(doc_id: int):
    tenant_id = current_tenant.get()

    doc = db.query(Document).filter(
        Document.id == doc_id,
        Document.tenant_id == tenant_id
    ).first()

    db.delete(doc)
    db.commit()

    return {"deleted": doc_id}

# Compliance query: Show all deletions in last 90 days
SELECT * FROM audit_log
WHERE tenant_id = 'acme'
  AND action LIKE '%.delete'
  AND timestamp > NOW() - INTERVAL '90 days'
ORDER BY timestamp DESC;

GDPR Compliance (Data Portability & Right to Deletion)

# GDPR: Right to data portability
@app.get("/api/admin/export-data")
async def export_tenant_data():
    """Export all tenant data in portable format (GDPR Article 20)"""
    tenant_id = current_tenant.get()

    # Export all tenant data
    data = {
        "documents": [
            doc.to_dict() for doc in
            db.query(Document).filter(Document.tenant_id == tenant_id).all()
        ],
        "users": [
            user.to_dict() for user in
            db.query(User).filter(User.tenant_id == tenant_id).all()
        ],
        "orders": [
            order.to_dict() for order in
            db.query(Order).filter(Order.tenant_id == tenant_id).all()
        ],
    }

    # Return as JSON file
    return Response(
        content=json.dumps(data, indent=2),
        media_type="application/json",
        headers={
            "Content-Disposition": f"attachment; filename={tenant_id}_data.json"
        }
    )

# GDPR: Right to deletion (Right to be forgotten)
@app.delete("/api/admin/delete-tenant")
async def delete_tenant_data():
    """Permanently delete all tenant data (GDPR Article 17)"""
    tenant_id = current_tenant.get()

    # Confirm deletion request (require admin approval)
    if not is_tenant_admin():
        raise HTTPException(403, "Only tenant admins can delete data")

    # Delete from all tables
    db.query(Document).filter(Document.tenant_id == tenant_id).delete()
    db.query(Order).filter(Order.tenant_id == tenant_id).delete()
    db.query(User).filter(User.tenant_id == tenant_id).delete()
    db.query(AuditLog).filter(AuditLog.tenant_id == tenant_id).delete()
    db.query(Tenant).filter(Tenant.id == tenant_id).delete()

    # Delete all tenant files from S3
    s3 = boto3.resource('s3')
    bucket = s3.Bucket("uploads")
    bucket.objects.filter(Prefix=f"{tenant_id}/").delete()

    db.commit()

    # Log deletion (in separate compliance log)
    compliance_log.info(f"Tenant {tenant_id} data deleted per GDPR request")

    return {"deleted": True, "tenant_id": tenant_id}

Compliance Certifications

Enterprise SaaS customers require compliance certifications:

SOC 2 Type II: Third-party audit of security controls (annual)
ISO 27001: Information security management certification
GDPR: EU data protection regulation (data portability, right to deletion)
HIPAA: Healthcare data protection (for health tech SaaS)
PCI DSS: Payment card industry compliance (if storing payment data)

Real-World Multi-Tenancy Examples

How major SaaS platforms implement multi-tenancy at massive scale.

Slack

Model: Hybrid (Graduated Isolation)

Small teams: Pool model (shared schema, MySQL sharding)
Enterprise: Dedicated shards for large customers (10k+ users)
Sharding: 100+ MySQL shards, routed by workspace_id
Scale: 20M+ workspaces, billions of messages/day
Key pattern: Message storage in S3, metadata in MySQL

Salesforce

Model: Multi-Tenant Database (Pool Model)

ALL customers: Share same Oracle database tables
Metadata-driven: Dynamic schema per tenant (stored as metadata)
Isolation: OrgId (tenant_id) on every row, enforced by DB
Scale: 150k+ customers, trillions of records
Key innovation: Custom objects defined per tenant at runtime

AWS (Multi-Tenant Services)

Model: Hybrid (Service-Dependent)

Lambda: Pool model (shared compute, isolated execution contexts)
RDS: Silo model (dedicated databases per customer)
S3: Pool model (shared storage, bucket-level isolation)
DynamoDB: Pool model (shared tables, partition key isolation)
Key pattern: IAM policies enforce cross-account isolation

Shopify

Model: Sharded Multi-Tenant (Pool)

Sharding: 1000+ MySQL shards, hash-based routing by shop_id
All merchants: Share same table schemas (shop_id discriminator)
Resiliency: Shard isolation (one shard failure doesn't affect others)
Scale: 2M+ merchants, peak 80k+ req/sec
Key pattern: Background jobs partitioned by shard for parallelism

Common Pattern

All major SaaS platforms use graduated isolation: start with pool model for cost efficiency, migrate high-value customers to dedicated infrastructure. Sharding is critical at scale (1M+ tenants). Metadata-driven schemas provide flexibility without schema migrations.

Multi-Tenancy Best Practices

Golden Rules

DO: Defense in Depth

Application-level filtering + Database RLS + Audit logging
Never trust a single layer of isolation
Regular penetration testing for tenant leakage
Monitor for queries missing tenant_id

DON'T: Trust Application Layer Alone

Never rely solely on application code for isolation
A single bug can leak all customer data
Must have database-level enforcement (RLS)
Don't skip audit logging "to save costs"

DO: Start Simple, Scale Gradually

Begin with pool model (shared schema) for first 1000 tenants
Add sharding when you hit performance bottlenecks
Offer dedicated infrastructure as paid upgrade
Measure before premature optimization

DON'T: Over-Engineer Day One

Don't build sharding for 10 customers
Don't offer 5 isolation tiers with no customers
Avoid separate databases per tenant initially
Complexity has a cost, earn it with scale

Common Pitfalls & Solutions

✗ Missing tenant_id in JOIN

Forgot to add tenant_id filter in JOIN, leaking data across tenants.

-- BAD: Missing tenant_id in JOIN
SELECT d.*, u.name
FROM documents d
JOIN users u ON d.user_id = u.id
WHERE d.tenant_id = 'acme';
-- Joins users from ALL tenants!

✓ Solution:

-- GOOD: tenant_id in all JOIN conditions
SELECT d.*, u.name
FROM documents d
JOIN users u ON d.user_id = u.id
  AND d.tenant_id = u.tenant_id
WHERE d.tenant_id = 'acme';

✗ Global Background Jobs

Background job processes all tenants in one query, slow and blocks on large tenants.

# BAD: One job for all tenants
def send_daily_reports():
    for tenant in get_all_tenants():
        generate_report(tenant)
# Blocks on slow tenants

✓ Solution:

# GOOD: Separate job per tenant
def send_daily_reports():
    for tenant in get_all_tenants():
        celery.send_task(
            'generate_report',
            args=[tenant.id]
        )
# Parallel, isolated execution

✗ Shared Redis Keys

Cache keys without tenant_id prefix, causing cache collision across tenants.

# BAD: Key collision
redis.set("user:123", data)
# Overwrites across tenants!

✓ Solution:

# GOOD: Prefix with tenant_id
redis.set(f"{tenant_id}:user:123", data)
# Isolated per tenant

✗ No Monitoring for Tenant Leakage

Production bug leaks data for weeks before discovery. No alerts configured.

✓ Solution:

# Monitor slow queries without tenant_id
-- PostgreSQL: Log queries
log_statement = 'all'

# Alert on queries missing tenant_id
SELECT query FROM pg_stat_statements
WHERE query NOT LIKE '%tenant_id%'
  AND query LIKE '%SELECT%FROM%';

Practice Exercise: Build a Multi-Tenant SaaS API

Project: Task Management SaaS

Build a multi-tenant task management API where each company gets isolated data, rate limits, and custom branding. Implement the complete multi-tenancy stack.

Requirements

Implement shared schema with tenant_id column in all tables
Create PostgreSQL tables: tenants, users, tasks, projects
Enable Row-Level Security (RLS) on all tables
Create RLS policies to enforce tenant isolation at database level
Write middleware to extract tenant_id from subdomain (e.g., acme.taskapp.com)
Store tenant context in contextvars.ContextVar for async safety

Implement Redis-based rate limiting per tenant (free: 100 req/min, pro: 1000 req/min)
Add storage quota enforcement (free: 1GB, pro: 100GB)
Track storage usage when tasks/files are created
Return 429 status code when rate limit exceeded
Return 403 when storage quota exceeded

Create endpoints: GET /api/tasks, POST /api/tasks, PUT /api/tasks/:id, DELETE /api/tasks/:id
ALL queries must include tenant_id filter
Prevent IDOR by verifying tenant ownership before updates/deletes
Use composite keys: tenant_id:task_id in responses
Test: Verify user from tenant A cannot access tenant B's data

Add tenant_config table with JSONB columns for branding and settings
Implement GET /api/config to return tenant branding (logo, colors, company name)
Implement PUT /api/admin/settings to update timezone, date format, etc.
Use Pydantic models to validate settings before saving
Create feature flag system: is_feature_enabled(tenant_id, "advanced_reports")

Create audit_log table with tenant_id, user_id, action, timestamp
Log all create/update/delete operations automatically (use decorator)
Implement GET /api/admin/audit-log to query logs per tenant
Add IP address and user-agent to logs for forensics
Write test: Verify audit log contains record after task deletion

Write integration test: Create tasks for two tenants, verify isolation
Test IDOR vulnerability: Try to access another tenant's task by ID
Test rate limiting: Send 101 requests and verify 429 on 101st
Test RLS: Disable application filter, verify DB still blocks cross-tenant queries
Test data export: Verify GDPR export includes all tenant data

Bonus Challenges

Sharding: Implement horizontal sharding across 4 databases using tenant_id hash
Graduated Isolation: Migrate a tenant from shared to dedicated database (zero downtime)
Usage Metering: Track API requests, storage, compute per tenant for billing
Tenant Migration: Export tenant data, import to new instance, verify integrity

Key Takeaways

Multi-tenancy is the foundation of SaaS economics, serve thousands of customers from shared infrastructure while maintaining isolation
Four isolation models: Separate database (silo), separate schema (bridge), shared schema (pool), and hybrid (graduated isolation)
Pool model (shared schema) is most cost-efficient but requires defense-in-depth: application filtering + PostgreSQL RLS + audit logging
Context propagation: Extract tenant_id from subdomain/JWT/header, store in ContextVar, inject into all database queries
Data leakage prevention: Always include tenant_id in WHERE clauses, use Row-Level Security (RLS), implement IDOR protection with composite keys
Rate limiting & quotas: Prevent noisy neighbor problem with Redis rate limiting, storage quotas, and per-tenant connection pools
Data partitioning: Scale to millions of tenants with horizontal sharding, PostgreSQL declarative partitioning, and tenant_id-prefixed indexes
Tenant configuration: Use JSONB for flexible per-tenant settings, feature flags, and white-label branding without schema migrations
Scaling strategies: Start with pool model, graduate high-value customers to dedicated infrastructure, implement usage-based billing
Security & compliance: Encrypt sensitive data with AWS KMS per tenant, implement comprehensive audit logging for SOC2/ISO 27001, support GDPR data portability and right to deletion
Real-world patterns: Slack uses graduated isolation, Salesforce uses pool model with metadata-driven schemas, Shopify shards across 1000+ databases
Golden rules: Defense-in-depth (never trust one layer), start simple and scale gradually, always prefix cache keys with tenant_id, monitor for queries missing tenant_id filters

Back to "Lambda & Kappa Architecture"

Architecture EssentialsLesson 8

Next: "Onion Architecture"

Multi-Tenancy Patterns

The SaaS Multi-Tenancy Challenge

Quick Navigation

The Core Tension: Resource Sharing vs Isolation

Resource Sharing Benefits

Isolation Requirements

The Balance

Tenant Isolation Models

1. Separate Database per Tenant (Silo Model)

2. Shared Database, Separate Schema per Tenant

3. Shared Database, Shared Schema (Pool Model with tenant_id)

Critical Security Warning

4. Hybrid Approaches (Graduated Isolation)

Isolation Model Comparison

Tenant Context Propagation

Extracting Tenant ID from Requests

FastAPI Middleware for Tenant Context

Important: Don't Use threading.local with Async

Data Leakage Prevention

PostgreSQL Row-Level Security (RLS)

RLS Benefits

Preventing IDOR (Insecure Direct Object Reference)

Defense-in-Depth Checklist

Rate Limiting & Resource Quotas

Redis-Based Rate Limiting

Resource Quotas (Storage, Compute)

Preventing Noisy Neighbor (Database Connection Pooling)

Noisy Neighbor Impact

Data Partitioning Strategies

Horizontal Sharding by Tenant

PostgreSQL Declarative Partitioning

Indexing Strategies for Multi-Tenant Tables

Partitioning Performance Impact

Tenant Configuration & Customization

Feature Flags Per Tenant

Tenant Branding & Customization

JSONB Settings with Validation

JSONB Benefits

Scaling Strategies

Graduated Isolation (Tier-Based Architecture)

Tenant Migration (Moving Up Tiers)

Cost Allocation & Metering

Usage-Based Pricing

Security & Compliance

Encryption at Rest (AWS KMS per Tenant)

Audit Logging (SOC2 / ISO 27001)

GDPR Compliance (Data Portability & Right to Deletion)

Compliance Certifications

Real-World Multi-Tenancy Examples

Slack

Salesforce

AWS (Multi-Tenant Services)

Shopify

Common Pattern

Multi-Tenancy Best Practices

Golden Rules

DO: Defense in Depth

DON'T: Trust Application Layer Alone

DO: Start Simple, Scale Gradually

DON'T: Over-Engineer Day One

Common Pitfalls & Solutions

✗ Missing tenant_id in JOIN

✗ Global Background Jobs

✗ Shared Redis Keys

✗ No Monitoring for Tenant Leakage

Practice Exercise: Build a Multi-Tenant SaaS API

Project: Task Management SaaS

Requirements

Tenant Isolation (Pool Model)

Rate Limiting & Quotas

CRUD APIs with Tenant Filtering

Tenant Configuration

Audit Logging & Security

Testing & Validation

Bonus Challenges

Key Takeaways