If you're building a SaaS product, you're building a multi-tenant system. Whether you're serving 10 customers or 10,000, your API needs to keep their data separate, secure, and performant. Get it wrong, and you'll face data leaks, performance bottlenecks, and angry customers.
Multi-tenancy isn't just about adding a tenant_id column to your database. It's about architectural decisions that affect routing, data isolation, security, performance, and scalability. This guide walks through practical strategies for building multi-tenant APIs, using examples from a hypothetical multi-tenant version of the PetStore API.
Multi-tenancy means serving multiple customers (tenants) from a single application instance. Each tenant gets their own isolated environment, but they share the same codebase and infrastructure.
Think of it like an apartment building: multiple families (tenants) live in the same building (application), each with their own private space (data), but sharing common infrastructure (servers, database, code).
Cost efficiency: One application serves many customers, reducing infrastructure and maintenance costs.
Faster updates: Deploy once, update everyone. No managing separate instances per customer.
Resource optimization: Share resources across tenants, improving utilization.
Easier scaling: Add new tenants without deploying new infrastructure.
The tradeoff? Complexity. You need robust isolation, security, and performance management.
The first decision: how do you identify which tenant is making a request?
Each tenant gets their own subdomain:
https://acme-pets.petstore.com/api/pets
https://fluffy-friends.petstore.com/api/pets
https://paws-and-claws.petstore.com/api/pets
Pros:- Clear tenant separation - Easy to implement SSL certificates per tenant - Feels more "branded" to customers - Simple to route at load balancer level
Cons:- Requires wildcard DNS setup - More complex SSL certificate management - Can't easily share sessions across tenants
Implementation:
from flask import Flask, request, g
import re
app = Flask(__name__)
@app.before_request
def identify_tenant():
host = request.headers.get('Host', '')
# Extract tenant from subdomain
match = re.match(r'^([a-z0-9-]+)\.petstore\.com', host)
if match:
tenant_slug = match.group(1)
tenant = Tenant.query.filter_by(slug=tenant_slug).first()
if not tenant:
return jsonify({'error': 'Tenant not found'}), 404
g.tenant = tenant
else:
return jsonify({'error': 'Invalid host'}), 400
@app.route('/api/pets')
def list_pets():
# g.tenant is automatically available
pets = Pet.query.filter_by(tenant_id=g.tenant.id).all()
return jsonify([pet.to_dict() for pet in pets])
Tenant identifier in the URL path:
https://api.petstore.com/acme-pets/api/pets
https://api.petstore.com/fluffy-friends/api/pets
https://api.petstore.com/paws-and-claws/api/pets
Pros:- Simpler DNS setup - Single SSL certificate - Easier local development - Can share authentication across tenants
Cons:- Less "branded" feeling - Tenant slug visible in every URL - Slightly more complex routing logic
Implementation:
from flask import Flask, g
app = Flask(__name__)
@app.url_value_preprocessor
def extract_tenant(endpoint, values):
if values:
tenant_slug = values.pop('tenant_slug', None)
if tenant_slug:
tenant = Tenant.query.filter_by(slug=tenant_slug).first()
if not tenant:
abort(404, 'Tenant not found')
g.tenant = tenant
@app.route('/<tenant_slug>/api/pets')
def list_pets(tenant_slug):
pets = Pet.query.filter_by(tenant_id=g.tenant.id).all()
return jsonify([pet.to_dict() for pet in pets])
Pass tenant identifier in a custom header:
GET /api/pets
X-Tenant-ID: acme-pets
Pros:- Clean URLs - Easy to test - Flexible for different client types
Cons:- Not RESTful (URLs don't identify resources completely) - Harder to debug (can't just click a link) - Requires custom client configuration
Implementation:
@app.before_request
def identify_tenant():
tenant_id = request.headers.get('X-Tenant-ID')
if not tenant_id:
return jsonify({'error': 'X-Tenant-ID header required'}), 400
tenant = Tenant.query.filter_by(slug=tenant_id).first()
if not tenant:
return jsonify({'error': 'Tenant not found'}), 404
g.tenant = tenant
Recommendation: Use subdomain routing for customer-facing APIs (better branding), path-based for internal APIs (simpler), and header-based for mobile apps (cleaner).
Keeping tenant data separate is critical. Three main approaches:
All tenants share the same database and tables. Each row has a tenant_id column.
CREATE TABLE pets (
id SERIAL PRIMARY KEY,
tenant_id INTEGER NOT NULL REFERENCES tenants(id),
name VARCHAR(100),
species VARCHAR(50),
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_pets_tenant ON pets(tenant_id);
Pros:- Simple to implement - Easy to manage - Cost-effective - Easy to query across tenants (for analytics)
Cons:- Risk of data leakage if you forget WHERE tenant_id = ?- Performance can degrade with many tenants - Hard to customize per tenant - Compliance issues (some customers require physical separation)
Implementation with SQLAlchemy:
from flask_sqlalchemy import SQLAlchemy
from flask import g
db = SQLAlchemy()
class TenantMixin:
"""Automatically filter by tenant"""
tenant_id = db.Column(db.Integer, db.ForeignKey('tenants.id'), nullable=False)
@classmethod
def query_for_tenant(cls):
return cls.query.filter_by(tenant_id=g.tenant.id)
class Pet(db.Model, TenantMixin):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100))
species = db.Column(db.String(50))
# Usage
@app.route('/api/pets')
def list_pets():
pets = Pet.query_for_tenant().all() # Automatically filtered
return jsonify([pet.to_dict() for pet in pets])
Add a safety check to prevent accidental cross-tenant queries:
from sqlalchemy import event
from sqlalchemy.orm import Session
@event.listens_for(Session, 'before_flush')
def check_tenant_isolation(session, flush_context, instances):
"""Ensure all objects belong to current tenant"""
if not hasattr(g, 'tenant'):
return
for obj in session.new | session.dirty:
if hasattr(obj, 'tenant_id'):
if obj.tenant_id != g.tenant.id:
raise ValueError(f'Attempted to modify object from different tenant')
Each tenant gets their own PostgreSQL schema:
-- Tenant 1
CREATE SCHEMA acme_pets;
CREATE TABLE acme_pets.pets (...);
-- Tenant 2
CREATE SCHEMA fluffy_friends;
CREATE TABLE fluffy_friends.pets (...);
Pros:- Better isolation than shared schema - Can customize schema per tenant - Easier to backup/restore individual tenants - Better performance (smaller tables)
Cons:- More complex migrations - Harder to query across tenants - Schema limit per database (PostgreSQL: ~1000 practical limit)
Implementation:
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
class TenantDatabaseManager:
def __init__(self, base_url):
self.base_url = base_url
self.sessions = {}
def get_session(self, tenant_slug):
if tenant_slug not in self.sessions:
engine = create_engine(self.base_url)
# Set search_path to tenant schema
@event.listens_for(engine, 'connect')
def set_search_path(dbapi_conn, connection_record):
cursor = dbapi_conn.cursor()
cursor.execute(f'SET search_path TO {tenant_slug}, public')
cursor.close()
session = scoped_session(sessionmaker(bind=engine))
self.sessions[tenant_slug] = session
return self.sessions[tenant_slug]
db_manager = TenantDatabaseManager('postgresql://localhost/petstore')
@app.before_request
def setup_tenant_db():
g.db = db_manager.get_session(g.tenant.slug)
@app.route('/api/pets')
def list_pets():
pets = g.db.query(Pet).all() # Queries tenant-specific schema
return jsonify([pet.to_dict() for pet in pets])
Each tenant gets their own database:
acme_pets_db
fluffy_friends_db
paws_and_claws_db
Pros:- Complete isolation - Easy to scale (different servers) - Easy to backup/restore - Meets strict compliance requirements - Can customize everything per tenant
Cons:- Most complex to manage - Expensive (more database instances) - Migrations are harder - Cross-tenant queries nearly impossible
Implementation:
class TenantDatabaseManager:
def __init__(self):
self.engines = {}
def get_engine(self, tenant):
if tenant.id not in self.engines:
# Each tenant has their own connection string
engine = create_engine(tenant.database_url)
self.engines[tenant.id] = engine
return self.engines[tenant.id]
db_manager = TenantDatabaseManager()
@app.before_request
def setup_tenant_db():
engine = db_manager.get_engine(g.tenant)
g.db = scoped_session(sessionmaker(bind=engine))
@app.teardown_request
def cleanup_db(exception=None):
if hasattr(g, 'db'):
g.db.remove()
Recommendation: Start with shared database/shared schema for simplicity. Move to separate schemas or databases only when you need it (large tenants, compliance requirements, performance issues).
Different tenants have different plans and limits. Implement per-tenant rate limiting:
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
app,
key_func=lambda: f"{g.tenant.id}:{get_remote_address()}",
default_limits=[]
)
def get_tenant_rate_limit():
"""Return rate limit based on tenant's plan"""
plan_limits = {
'free': '100 per hour',
'basic': '1000 per hour',
'premium': '10000 per hour',
'enterprise': '100000 per hour'
}
return plan_limits.get(g.tenant.plan, '100 per hour')
@app.route('/api/pets')
@limiter.limit(get_tenant_rate_limit)
def list_pets():
pets = Pet.query_for_tenant().all()
return jsonify([pet.to_dict() for pet in pets])
Add rate limit info to response headers:
@app.after_request
def add_rate_limit_headers(response):
if hasattr(g, 'tenant'):
# Get current usage from Redis
key = f"rate_limit:{g.tenant.id}"
current = redis_client.get(key) or 0
limit = get_tenant_limit_number(g.tenant.plan)
response.headers['X-RateLimit-Limit'] = str(limit)
response.headers['X-RateLimit-Remaining'] = str(limit - int(current))
return response
Allow tenants to customize behavior:
class TenantConfig(db.Model):
id = db.Column(db.Integer, primary_key=True)
tenant_id = db.Column(db.Integer, db.ForeignKey('tenants.id'))
key = db.Column(db.String(100))
value = db.Column(db.Text)
@classmethod
def get(cls, key, default=None):
config = cls.query.filter_by(
tenant_id=g.tenant.id,
key=key
).first()
return config.value if config else default
# Usage
@app.route('/api/pets')
def list_pets():
# Check tenant-specific setting
include_deleted = TenantConfig.get('include_deleted_pets', 'false') == 'true'
query = Pet.query_for_tenant()
if not include_deleted:
query = query.filter_by(deleted=False)
pets = query.all()
return jsonify([pet.to_dict() for pet in pets])
Automate tenant provisioning:
from flask import request, jsonify
@app.route('/api/admin/tenants', methods=['POST'])
def create_tenant():
data = request.json
# Validate tenant data
if not data.get('slug') or not data.get('name'):
return jsonify({'error': 'slug and name required'}), 400
# Check if slug is available
if Tenant.query.filter_by(slug=data['slug']).first():
return jsonify({'error': 'Slug already taken'}), 409
# Create tenant
tenant = Tenant(
slug=data['slug'],
name=data['name'],
plan=data.get('plan', 'free'),
status='active'
)
db.session.add(tenant)
db.session.commit()
# Provision tenant resources
provision_tenant(tenant)
return jsonify(tenant.to_dict()), 201
def provision_tenant(tenant):
"""Set up everything a new tenant needs"""
# 1. Create database schema (if using separate schemas)
if USE_SEPARATE_SCHEMAS:
db.session.execute(f'CREATE SCHEMA {tenant.slug}')
db.session.commit()
# Run migrations for tenant schema
run_migrations_for_schema(tenant.slug)
# 2. Create default data
create_default_data(tenant)
# 3. Set up default configuration
create_default_config(tenant)
# 4. Send welcome email
send_welcome_email(tenant)
def create_default_data(tenant):
"""Create starter data for new tenant"""
# Create admin user
admin = User(
tenant_id=tenant.id,
email=f'admin@{tenant.slug}.com',
role='admin'
)
db.session.add(admin)
# Create sample pet
sample_pet = Pet(
tenant_id=tenant.id,
name='Sample Pet',
species='dog',
breed='Golden Retriever'
)
db.session.add(sample_pet)
db.session.commit()
Handle schema changes across all tenants:
from alembic import op
import sqlalchemy as sa
def upgrade():
"""Add new column to pets table for all tenants"""
if USE_SHARED_SCHEMA:
# Simple: just add column
op.add_column('pets', sa.Column('microchip_id', sa.String(50)))
elif USE_SEPARATE_SCHEMAS:
# Add column to each tenant schema
tenants = Tenant.query.all()
for tenant in tenants:
op.execute(f'SET search_path TO {tenant.slug}')
op.add_column('pets', sa.Column('microchip_id', sa.String(50)))
elif USE_SEPARATE_DATABASES:
# Run migration on each database
tenants = Tenant.query.all()
for tenant in tenants:
engine = create_engine(tenant.database_url)
with engine.connect() as conn:
conn.execute('ALTER TABLE pets ADD COLUMN microchip_id VARCHAR(50)')
Track per-tenant metrics:
from prometheus_client import Counter, Histogram
request_count = Counter(
'api_requests_total',
'Total API requests',
['tenant', 'endpoint', 'method', 'status']
)
request_duration = Histogram(
'api_request_duration_seconds',
'API request duration',
['tenant', 'endpoint']
)
@app.before_request
def start_timer():
g.start_time = time.time()
@app.after_request
def record_metrics(response):
if hasattr(g, 'tenant') and hasattr(g, 'start_time'):
duration = time.time() - g.start_time
request_count.labels(
tenant=g.tenant.slug,
endpoint=request.endpoint,
method=request.method,
status=response.status_code
).inc()
request_duration.labels(
tenant=g.tenant.slug,
endpoint=request.endpoint
).observe(duration)
return response
Never trust client-provided tenant IDs:
@app.route('/api/pets/<int:pet_id>')
def get_pet(pet_id):
pet = Pet.query.get_or_404(pet_id)
# CRITICAL: Verify pet belongs to current tenant
if pet.tenant_id != g.tenant.id:
abort(404) # Don't reveal it exists
return jsonify(pet.to_dict())
Don't leak information about other tenants:
@app.route('/<tenant_slug>/api/pets')
def list_pets(tenant_slug):
tenant = Tenant.query.filter_by(slug=tenant_slug).first()
if not tenant:
# Don't say "tenant not found" - that confirms it doesn't exist
abort(404)
# Verify user has access to this tenant
if not current_user.has_access_to_tenant(tenant):
abort(404) # Not 403, to avoid leaking existence
g.tenant = tenant
pets = Pet.query_for_tenant().all()
return jsonify([pet.to_dict() for pet in pets])
Log any attempts to access other tenants' data:
@app.before_request
def audit_tenant_access():
if hasattr(g, 'tenant') and current_user.is_authenticated:
if current_user.tenant_id != g.tenant.id:
log_security_event(
event='cross_tenant_access_attempt',
user_id=current_user.id,
user_tenant=current_user.tenant_id,
requested_tenant=g.tenant.id,
endpoint=request.endpoint
)
abort(403)
Multi-tenant API design is about balancing isolation, performance, and complexity. Start simple with shared database/shared schema, and evolve your architecture as needs grow.
Key takeaways:
Build these patterns in from the start. Retrofitting multi-tenancy into an existing API is painful. But get it right early, and you'll have a solid foundation for scaling to thousands of tenants.
