Skip to main content

Implementing Multi-Tenancy Architecture

Multi-tenancy is a deployment pattern where one application instance serves multiple customers, with data partitioned by tenant. A poorly implemented multi-tenant system leaks one customer's data to another's API—a catastrophic breach. This guide implements a production-grade multi-tenant architecture using shared schema with row-level security, tenant resolution middleware, and query enforcement that makes data leaks impossible even with developer mistakes.

Three Approaches to Multi-Tenancy​

SaaS systems use one of three patterns:

PatternSchemaIsolationComplexityScaling
Shared Schema (Row-Level Security)One schema, all tenantsTenant ID foreign key + RLS policiesMediumHigh; one database scales 1000+ tenants
Schema Per TenantSeparate schema per tenantDatabase layer enforces schema isolationHighMedium; need tenant router middleware
Separate DatabaseOne DB per tenantComplete database isolationVery HighLow; expensive to operate but simplest to code

Shared schema with row-level security is the industry standard for early-stage SaaS (Stripe, Notion, Figma all started here): it's cost-efficient, allows hot-running queries across all tenants, and PostgreSQL's RLS policies provide defense-in-depth.

Tenant Resolution: Extracting tenant_id from Request​

Every request must resolve to a single tenant_id. Methods (in order of preference):

from fastapi import Request, Depends, HTTPException
from typing import Annotated

# 1. Extract from JWT token (most secure, already authenticated)
async def get_tenant_from_jwt(
    request: Request
) -> int:
    """
    Extract tenant_id from JWT claims.
    The token is already validated by get_current_user().
    """
    # The current_user dependency (from auth article) added this to request.state
    if hasattr(request.state, "user"):
        return request.state.user.tenant_id
    raise HTTPException(status_code=401, detail="Tenant not found in token")

# 2. Extract from subdomain (tenant.app.com vs app.com/tenant/...)
async def get_tenant_from_subdomain(request: Request) -> int:
    """
    Parse hostname: 'acme.saasapp.com' -> tenant_id = acme's account ID.
    Requires DNS wildcard or custom domain mapping.
    """
    hostname = request.headers.get("host", "")
    subdomain = hostname.split(".")[0]
    
    if subdomain in ("api", "www", "app"):
        raise HTTPException(
            status_code=400,
            detail="Request must be to a tenant subdomain (e.g., acme.saasapp.com)"
        )
    
    # Look up tenant by subdomain in database
    tenant = db.query(Tenant).filter(Tenant.slug == subdomain).first()
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")
    return tenant.id

# 3. Extract from path (e.g., /api/v1/acme/documents)
async def get_tenant_from_path(request: Request) -> int:
    """
    Parse path: '/api/v1/acme/documents' -> tenant slug 'acme'.
    Simplest for multi-tenant SaaS without subdomains.
    """
    path_parts = request.url.path.split("/")
    if len(path_parts) < 3:
        raise HTTPException(status_code=400, detail="Tenant slug required in path")
    
    tenant_slug = path_parts[2]
    tenant = db.query(Tenant).filter(Tenant.slug == tenant_slug).first()
    if not tenant:
        raise HTTPException(status_code=404, detail="Tenant not found")
    return tenant.id

# Usage in FastAPI:
@app.get("/api/v1/{tenant_slug}/documents")
async def list_documents(
    tenant_slug: str,
    tenant_id: Annotated[int, Depends(get_tenant_from_path)]
):
    """Fetch all documents for the tenant."""
    return {"tenant_id": tenant_id, "documents": []}

Middleware to Enforce Tenant Context​

Create middleware that extracts the tenant once per request and validates it in every query:

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import contextvars

tenant_context: contextvars.ContextVar[int | None] = contextvars.ContextVar(
    "tenant_id", default=None
)

class TenantContextMiddleware(BaseHTTPMiddleware):
    """
    Middleware that extracts tenant_id from request and stores it in context.
    All subsequent code can access tenant_id via contextvars.
    """
    async def dispatch(self, request: Request, call_next):
        # Extract tenant_id from the request
        try:
            tenant_id = await get_tenant_from_path(request)
        except HTTPException as e:
            return JSONResponse(status_code=e.status_code, content={"detail": e.detail})
        
        # Store in context for downstream use
        tenant_context.set(tenant_id)
        
        # Add to request.state for inspection in handlers
        request.state.tenant_id = tenant_id
        
        # Proceed with request
        response = await call_next(request)
        return response

# Register middleware
app.add_middleware(TenantContextMiddleware)

# Access tenant_id anywhere in handlers:
def get_current_tenant() -> int:
    """Retrieve the current tenant_id from context."""
    tenant_id = tenant_context.get()
    if tenant_id is None:
        raise RuntimeError("Tenant context not set")
    return tenant_id

@app.get("/documents")
async def list_documents():
    tenant_id = get_current_tenant()
    return {"tenant_id": tenant_id}

Query Filtering: Automatic Tenant Isolation​

Create a helper function to apply tenant filters automatically:

from sqlalchemy import and_
from sqlalchemy.orm import Query

def apply_tenant_filter(query: Query, model_class, tenant_id: int) -> Query:
    """
    Apply tenant_id filter to any query on a multi-tenant model.
    This is the defense-in-depth against accidental data leaks.
    """
    if not hasattr(model_class, "tenant_id"):
        raise ValueError(
            f"{model_class.__name__} is not a multi-tenant model (no tenant_id column)"
        )
    
    return query.filter(model_class.tenant_id == tenant_id)

# Usage in endpoints:
@app.get("/documents")
async def list_documents(session: Annotated[Session, Depends(get_db)]):
    """List all documents for the current tenant (guaranteed isolation)."""
    tenant_id = get_current_tenant()
    
    query = session.query(Document)
    query = apply_tenant_filter(query, Document, tenant_id)
    documents = query.all()
    
    return documents

# For safety, create a custom session class:
class TenantAwareSession:
    """
    A session wrapper that auto-applies tenant filters to all queries.
    Prevents mistakes where developers forget the tenant_id filter.
    """
    def __init__(self, session: Session, tenant_id: int):
        self.session = session
        self.tenant_id = tenant_id
    
    def query(self, model_class):
        """Query a model; automatically filter by tenant."""
        query = self.session.query(model_class)
        if hasattr(model_class, "tenant_id"):
            query = query.filter(model_class.tenant_id == self.tenant_id)
        return query
    
    def __getattr__(self, name):
        """Delegate other methods to the underlying session."""
        return getattr(self.session, name)

# Use in dependency:
async def get_tenant_aware_db() -> TenantAwareSession:
    """Returns a session that auto-filters queries by tenant."""
    session = Session(engine)
    tenant_id = get_current_tenant()
    return TenantAwareSession(session, tenant_id)

@app.get("/documents")
async def list_documents(
    session: Annotated[TenantAwareSession, Depends(get_tenant_aware_db)]
):
    """
    Automatic tenant filtering: queries are limited to current tenant.
    Developers can't accidentally leak data even if they forget the filter.
    """
    documents = session.query(Document).all()
    return documents

PostgreSQL Row-Level Security (RLS)​

PostgreSQL RLS policies enforce tenant isolation at the database layer (defense-in-depth):

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

-- Create a policy: users can only see their own tenant's data
CREATE POLICY tenant_isolation ON users
  USING (tenant_id = current_setting('app.tenant_id')::int);

-- Create a policy: users can only insert rows for their tenant
CREATE POLICY insert_own_tenant ON users
  FOR INSERT WITH CHECK (tenant_id = current_setting('app.tenant_id')::int);

-- Disable for superuser (admin/migration purposes)
ALTER TABLE users FORCE ROW LEVEL SECURITY;  -- Only superusers can disable this

Set the tenant context at the database level:

def execute_with_tenant_context(session: Session, tenant_id: int):
    """
    Execute any query with tenant context set at the database level.
    PostgreSQL RLS policies will enforce isolation.
    """
    session.execute(
        sa.text(f"SET app.tenant_id TO {tenant_id}")
    )

Cross-Tenant Data Validation​

When a user updates their account, verify the resource belongs to their tenant:

@app.put("/documents/{document_id}")
async def update_document(
    document_id: int,
    update: DocumentUpdate,
    session: Annotated[Session, Depends(get_db)]
):
    """
    Update a document. Verify it belongs to the current tenant.
    """
    tenant_id = get_current_tenant()
    
    document = session.query(Document).filter(
        Document.id == document_id,
        Document.tenant_id == tenant_id  # Critical: verify tenant ownership
    ).first()
    
    if not document:
        # Return 404 whether document doesn't exist or is owned by another tenant
        # (don't reveal which via different status code)
        raise HTTPException(status_code=404, detail="Document not found")
    
    document.title = update.title
    document.content = update.content
    session.commit()
    return document

Tenant Usage Metrics and Rate Limiting​

Track per-tenant API usage:

from redis import Redis

redis_client = Redis.from_url(os.getenv("REDIS_URL"))

def check_rate_limit(tenant_id: int, limit: int = 1000) -> bool:
    """
    Check if tenant has exceeded their request limit.
    Uses Redis for fast, distributed rate limiting.
    """
    key = f"rate_limit:{tenant_id}:{datetime.utcnow().strftime('%Y%m%d')}"
    current = redis_client.incr(key)
    
    if current == 1:
        # First request of the day; set expiry
        redis_client.expire(key, 86400)  # 24 hours
    
    return current <= limit

class RateLimitMiddleware(BaseHTTPMiddleware):
    """Enforce per-tenant rate limits."""
    async def dispatch(self, request: Request, call_next):
        tenant_id = getattr(request.state, "tenant_id", None)
        if tenant_id and not check_rate_limit(tenant_id):
            return JSONResponse(
                status_code=429,
                content={"detail": "Rate limit exceeded"}
            )
        return await call_next(request)

app.add_middleware(RateLimitMiddleware)

Key Takeaways​

  • Resolve tenant_id from request (JWT, subdomain, or path) at the middleware layer.
  • Apply tenant filters to every query automatically using a custom session wrapper.
  • Use PostgreSQL RLS as defense-in-depth; enforce isolation at both application and database layers.
  • Verify tenant ownership on update/delete operations; return 404 for unauthorized access (don't leak existence).
  • Track per-tenant usage for billing and rate limiting.

Frequently Asked Questions​

What if I need to run a query across all tenants (e.g., analytics)?​

Create a separate analytics database or a read-only replica with all tenants. Query it separately without the tenant filter. Alternatively, loop through tenants in code: for tenant_id in get_all_tenant_ids(): ... This prevents accidental exposure.

Can I query two tenants' data in a single request?​

No. The tenant context is single-valued per request. If you need to aggregate, fetch each tenant's data separately and combine in application code. This maintains isolation and makes auditing clearer.

How do I test multi-tenant isolation in unit tests?​

Create fixtures for each tenant. Mock get_current_tenant() to return different tenant IDs in different tests. Verify that queries return data only for the specified tenant.

What if I migrate between multi-tenancy models (e.g., shared schema to database-per-tenant)?​

This is a major architectural change. Use a temporary "dual-write" period where you write to both old and new systems, validate they match, then switch reads. Plan for 2–4 weeks.

How do I handle tenant-specific features (A/B tests, beta flags)?​

Store a features JSON column on the Tenant model. Query it when rendering: if 'new_dashboard' in tenant.features: ... This avoids code branches and allows per-tenant control.

Further Reading​