Multi-Tenancy: Architecture and Usage Guide
This guide covers Framework M's multi-tenancy architecture, including the two tenancy modes (Single and Multi-Tenant), tenant isolation patterns, and integration patterns for both backend and frontend.
Table of Contents
- Overview
- Mode A: Single Tenant (Indie)
- Mode B: Multi-Tenant (Enterprise)
- Tenant Isolation Patterns
- Using TenantContext in Code
- Frontend Multi-Tenancy
- Switching Adapters
Overview
Framework M provides a flexible multi-tenancy system with two operational modes:
- Mode A (Single Tenant / Indie): Uses
ImplicitTenantAdapterwith a fixed "default" tenant. No database lookups, perfect for standalone deployments. - Mode B (Multi-Tenant / Enterprise): Uses
HeaderTenantAdapterto extract tenant information from gateway headers. Requires an API gateway (e.g., Keycloak, Auth0, Citadel IAM) for tenant resolution.
Both modes implement the TenantProtocol interface, allowing seamless switching between deployment configurations.
TenantProtocol Interface
from typing import Protocol, Any
class TenantProtocol(Protocol):
"""Protocol for tenant resolution and attribute retrieval."""
def get_current_tenant(self) -> str:
"""Get the current tenant ID from request context."""
...
def get_tenant_attributes(self, tenant_id: str) -> dict[str, Any]:
"""Get attributes for a specific tenant."""
...
TenantContext
All tenant adapters return a TenantContext object:
from framework_m.core.interfaces.tenant import TenantContext
ctx = TenantContext(
tenant_id="acme-corp",
attributes={
"plan": "enterprise",
"features": ["advanced_reports", "api_access"],
"max_users": 100
},
is_default=False
)
Mode A: Single Tenant (Indie)
Overview
ImplicitTenantAdapter is designed for standalone deployments where you have a single tenant. It provides:
- No database lookups: Tenant is hardcoded or configured via
framework_config.toml - Unlimited features: Default attributes give full access to all features
- Zero overhead: No runtime tenant resolution cost
- Simple deployment: Perfect for indie apps, MVPs, and single-customer deployments
Configuration
Configure single-tenant mode in framework_config.toml:
[tenancy]
mode = "single"
default_tenant_id = "default" # Or your organization name
Usage
Basic Usage
from framework_m.adapters.tenant import ImplicitTenantAdapter
# Use default configuration
adapter = ImplicitTenantAdapter()
tenant_id = adapter.get_current_tenant() # "default"
attrs = adapter.get_tenant_attributes("default")
# {"plan": "unlimited", "features": "*"}
Custom Tenant ID
# Override tenant ID (useful for branding)
adapter = ImplicitTenantAdapter(tenant_id="acme-corp")
tenant_id = adapter.get_current_tenant() # "acme-corp"
Custom Attributes
# Override default attributes
adapter = ImplicitTenantAdapter(
tenant_id="my-app",
attributes={
"plan": "pro",
"features": ["reports", "api", "webhooks"],
"max_storage_gb": 100
}
)
Getting Full Context
adapter = ImplicitTenantAdapter()
ctx = adapter.get_context()
print(ctx.tenant_id) # "default"
print(ctx.attributes) # {"plan": "unlimited", "features": "*"}
print(ctx.is_default) # True
When to Use Mode A
✅ Use ImplicitTenantAdapter when:
- Building an indie app or MVP
- Single customer deployment
- No need for tenant isolation
- Want to minimize complexity
- Running on-premise for one organization
- Prototype or demo environments
❌ Don't use when:
- You need to serve multiple customers
- Tenant isolation is required
- You plan to scale to multi-tenancy later (start with Mode B)
Implementation Details
# frameworks-m/src/framework_m/adapters/tenant.py
class ImplicitTenantAdapter:
"""Single-tenant mode adapter."""
def __init__(
self,
tenant_id: str | None = None,
attributes: dict[str, Any] | None = None,
) -> None:
self._tenant_id = tenant_id or get_default_tenant_id()
self._attributes = attributes or {
"plan": "unlimited",
"features": "*",
}
def get_current_tenant(self) -> str:
return self._tenant_id
def get_tenant_attributes(self, tenant_id: str) -> dict[str, Any]:
return self._attributes.copy()
def get_context(self) -> TenantContext:
return TenantContext(
tenant_id=self._tenant_id,
attributes=self._attributes.copy(),
is_default=True,
)
Mode B: Multi-Tenant (Enterprise)
Overview
HeaderTenantAdapter is designed for multi-tenant SaaS deployments where an API gateway handles authentication and populates tenant information in HTTP headers.
Key Features:
- Gateway integration: Works with Keycloak, Auth0, AWS ALB, Citadel IAM
- Zero trust: Framework never authenticates users, only reads headers
- Feature flags: Per-tenant attributes control feature availability
- Dynamic attributes: Plan, features, limits passed in headers
- Security: Gateway ensures headers can't be spoofed
Architecture
┌─────────┐ ┌──────────────┐ ┌────────────────┐
│ Browser │─────▶│ API Gateway │─────▶│ Framework M │
│ │ │ (Keycloak) │ │ (Backend) │
└─────────┘ └──────────────┘ └────────────────┘
│
├─ Authenticate user
├─ Resolve tenant from JWT
├─ Fetch tenant attributes
└─ Add headers:
- X-Tenant-ID: acme-corp
- X-Tenant-Attributes: {...}
Configuration
Configure multi-tenant mode in framework_config.toml:
[tenancy]
mode = "multi"
default_tenant_id = "default" # Fallback if header missing
Gateway Setup
Expected Headers
The gateway must populate these headers:
X-Tenant-ID: acme-corp
X-Tenant-Attributes: {"plan": "enterprise", "features": ["reports", "api"], "max_users": 100}
Example: Keycloak Configuration
- Add Tenant Claim to JWT:
// Keycloak Mapper: Add "tenant_id" claim
{
"tenant_id": "acme-corp",
"tenant_attributes": {
"plan": "enterprise",
"features": ["advanced_reports", "api_access"],
"max_users": 100
}
}
- Configure Gateway to Extract Headers:
# Nginx config (or use Keycloak Gatekeeper/Louketo)
location /api/ {
proxy_set_header X-Tenant-ID $jwt_claim_tenant_id;
proxy_set_header X-Tenant-Attributes $jwt_claim_tenant_attributes;
proxy_pass http://framework-m-backend:8000;
}
Example: Citadel IAM Integration
Citadel IAM can be configured to automatically populate tenant headers:
# citadel-config.yaml
tenant_resolution:
source: jwt_claim
claim_name: tenant_id
attributes_claim: tenant_attributes
headers:
tenant_id: X-Tenant-ID
attributes: X-Tenant-Attributes
Usage
Basic Usage
from framework_m.adapters.tenant import HeaderTenantAdapter
# In a Litestar dependency provider
headers = {
"x-tenant-id": "acme-corp",
"x-tenant-attributes": '{"plan": "enterprise", "features": ["reports"]}',
}
adapter = HeaderTenantAdapter(headers=headers)
tenant_id = adapter.get_current_tenant() # "acme-corp"
Custom Header Names
# Use custom header names (e.g., for legacy systems)
adapter = HeaderTenantAdapter(
headers=headers,
tenant_header="x-organization-id",
attributes_header="x-organization-data"
)
Fallback Behavior
# If X-Tenant-ID header is missing, falls back to default
headers = {} # No tenant header
adapter = HeaderTenantAdapter(headers=headers)
tenant_id = adapter.get_current_tenant() # "default"
Getting Full Context
headers = {
"x-tenant-id": "acme-corp",
"x-tenant-attributes": '{"plan": "enterprise", "max_users": 100}',
}
adapter = HeaderTenantAdapter(headers=headers)
ctx = adapter.get_context()
print(ctx.tenant_id) # "acme-corp"
print(ctx.attributes) # {"plan": "enterprise", "max_users": 100}
print(ctx.is_default) # False
When to Use Mode B
✅ Use HeaderTenantAdapter when:
- Building a multi-tenant SaaS product
- Serving multiple customers from one deployment
- Need tenant data isolation
- Using an API gateway (Keycloak, Auth0, etc.)
- Require feature flags per tenant
- Different plans/tiers for customers
❌ Don't use when:
- Single customer deployment
- No API gateway infrastructure
- Can't guarantee header security
Implementation Details
# framework-m/src/framework_m/adapters/tenant.py
class HeaderTenantAdapter:
"""Multi-tenant mode adapter."""
def __init__(
self,
headers: Mapping[str, str],
tenant_header: str = "x-tenant-id",
attributes_header: str = "x-tenant-attributes",
default_tenant_id: str = "default",
) -> None:
self._headers = headers
self._tenant_header = tenant_header
self._attributes_header = attributes_header
self._default_tenant_id = default_tenant_id
def get_current_tenant(self) -> str:
return self._headers.get(self._tenant_header, self._default_tenant_id)
def get_tenant_attributes(self, tenant_id: str) -> dict[str, Any]:
attrs_json = self._headers.get(self._attributes_header, "")
if not attrs_json:
return {}
try:
result = json.loads(attrs_json)
return result if isinstance(result, dict) else {}
except json.JSONDecodeError:
return {}
def get_context(self) -> TenantContext:
tenant_id = self.get_current_tenant()
return TenantContext(
tenant_id=tenant_id,
attributes=self.get_tenant_attributes(tenant_id),
is_default=(tenant_id == self._default_tenant_id),
)
Tenant Isolation Patterns
Framework M supports three tenant isolation strategies. Choose based on your security, compliance, and scalability requirements.
Pattern 1: Logical Isolation (Recommended)
Description: All tenants share the same database and tables. Row-Level Security (RLS) filters ensure tenants only see their own data.
Implementation:
from framework_m.core.domain.base_doctype import BaseDocType
from pydantic import Field
class Invoice(BaseDocType):
"""Invoice DocType with tenant isolation."""
tenant_id: str = Field(max_length=100)
customer_name: str = Field(max_length=200)
amount: float
class Meta:
apply_rls: bool = True
rls_field: str = "tenant_id" # Filter by this field
How RLS Works:
GenericRepositoryautomatically applies RLS filters:
# User queries for invoices
invoices = await repo.list_entities(session, filters=user_filters)
# Framework automatically adds: WHERE tenant_id = 'acme-corp'
# User only sees their tenant's invoices
- Admins can bypass RLS:
# Admin user with is_system_user=True bypasses RLS
admin_invoices = await repo.list_entities(session, filters=[], user=admin_user)
# Admin sees all invoices across all tenants
Pros:
- ✅ Simple to implement and manage
- ✅ Cost-effective (single database)
- ✅ Easy backups and maintenance
- ✅ Dynamic tenant creation (no schema changes)
- ✅ Framework handles filtering automatically
Cons:
- ❌ "Noisy neighbor" problem (one tenant can impact others)
- ❌ Cannot customize schema per tenant
- ❌ Compliance concerns for highly sensitive data
Best For:
- Most SaaS applications
- B2B products with standard features
- Startups and growing companies
- When tenant data has similar structure
Pattern 2: Database-per-Tenant
Description: Each tenant gets a separate database. Complete physical isolation.
Implementation:
# framework_config.toml
[database]
url = "postgresql://user:pass@localhost/framework_default"
# Tenant-specific database bindings
[database.tenant_binds]
"acme-corp" = "postgresql://user:pass@localhost/framework_acme"
"globex" = "postgresql://user:pass@localhost/framework_globex"
# In repository or session factory
def get_database_url(tenant_id: str) -> str:
"""Get database URL for tenant."""
from framework_m.cli.config import load_config
config = load_config()
tenant_binds = config.get("database", {}).get("tenant_binds", {})
# Check for tenant-specific database
if tenant_id in tenant_binds:
return tenant_binds[tenant_id]
# Fallback to default database
return config["database"]["url"]
Pros:
- ✅ Complete data isolation
- ✅ Can customize schema per tenant
- ✅ Easy to backup individual tenants
- ✅ No RLS overhead
- ✅ Can move tenant data to different servers
Cons:
- ❌ High operational complexity
- ❌ More expensive (multiple databases)
- ❌ Schema migrations across all databases
- ❌ Connection pool management
- ❌ Cross-tenant queries impossible
Best For:
- Enterprise customers demanding isolation
- Regulated industries (healthcare, finance)
- Customers with custom schema needs
- When billing per database is acceptable
Pattern 3: Schema-per-Tenant (PostgreSQL)
Description: All tenants in one database, but each gets a separate PostgreSQL schema.
Implementation:
# Set PostgreSQL search_path per tenant
from sqlalchemy import text
async def set_tenant_schema(session, tenant_id: str):
"""Set PostgreSQL schema for tenant."""
schema_name = f"tenant_{tenant_id.replace('-', '_')}"
await session.execute(text(f"SET search_path TO {schema_name}, public"))
# In session provider
from litestar import Request
async def provide_session(request: Request):
"""Provide database session with tenant schema."""
tenant_ctx = request.state.tenant
async with get_session() as session:
# Set schema based on tenant
await set_tenant_schema(session, tenant_ctx.tenant_id)
yield session
Pros:
- ✅ Good isolation (separate schemas)
- ✅ Single database (easier management)
- ✅ Can customize schema per tenant
- ✅ Better than logical isolation for compliance
Cons:
- ❌ PostgreSQL-specific
- ❌ Schema migrations across all schemas
- ❌ Still shares database resources
- ❌ Requires careful connection management
Best For:
- PostgreSQL-only deployments
- Mid-tier isolation needs
- When database-per-tenant is too expensive
- Regulated data with moderate isolation needs
Comparison Table
| Feature | Logical Isolation | Database-per-Tenant | Schema-per-Tenant |
|---|---|---|---|
| Setup Complexity | Low | High | Medium |
| Operational Cost | Low | High | Medium |
| Data Isolation | Application-level | Physical | Schema-level |
| Custom Schema | ❌ No | ✅ Yes | ✅ Yes |
| Scalability | ✅ Excellent | ⚠️ Limited | ⚠️ Good |
| Compliance | ⚠️ Moderate | ✅ Excellent | ✅ Good |
| Framework Support | ✅ Built-in | ⚠️ Manual | ⚠️ Manual |
Using TenantContext in Code
Dependency Injection Pattern
Framework M provides TenantContext via dependency injection:
from litestar import get, post, Request
from framework_m.core.interfaces.tenant import TenantContext
@get("/api/dashboard")
async def get_dashboard(request: Request) -> dict:
"""Get tenant-specific dashboard data."""
tenant_ctx: TenantContext = request.state.tenant
return {
"tenant_id": tenant_ctx.tenant_id,
"plan": tenant_ctx.attributes.get("plan", "free"),
"features": tenant_ctx.attributes.get("features", []),
}
Feature Flags from Tenant Attributes
@post("/api/reports/advanced")
async def generate_advanced_report(request: Request) -> dict:
"""Generate advanced report (enterprise feature)."""
tenant_ctx: TenantContext = request.state.tenant
features = tenant_ctx.attributes.get("features", [])
if "advanced_reports" not in features:
raise HTTPException(
status_code=403,
detail="Advanced reports not available in your plan"
)
# Generate report...
return {"status": "success"}
Tenant-Aware Queries
When using GenericRepository, RLS filters are applied automatically:
from framework_m.adapters.db.generic_repository import GenericRepository
from framework_m.core.doctypes.invoice import Invoice
@get("/api/invoices")
async def list_invoices(
request: Request,
user: UserContext,
session: AsyncSession
) -> list[dict]:
"""List invoices for current tenant."""
repo = GenericRepository(Invoice)
# RLS automatically filters: WHERE tenant_id = 'acme-corp'
invoices = await repo.list_entities(session, user=user)
return [inv.model_dump() for inv in invoices]
Cross-Tenant Queries (Admin Only)
@get("/api/admin/all-invoices")
async def list_all_invoices(
user: UserContext,
session: AsyncSession
) -> list[dict]:
"""List invoices across all tenants (admin only)."""
if not user.is_system_user:
raise HTTPException(status_code=403, detail="Admin access required")
repo = GenericRepository(Invoice)
# Admin bypasses RLS, sees all tenants
all_invoices = await repo.list_entities(session, user=user)
return [inv.model_dump() for inv in all_invoices]
Manual Tenant Filtering
If you need manual control:
from framework_m.core.interfaces.repository import FilterSpec, FilterOperator
@get("/api/custom-query")
async def custom_query(
request: Request,
session: AsyncSession
) -> list[dict]:
"""Custom query with manual tenant filtering."""
tenant_ctx: TenantContext = request.state.tenant
repo = GenericRepository(Invoice)
# Manual tenant filter
filters = [
FilterSpec(
field="tenant_id",
operator=FilterOperator.EQ,
value=tenant_ctx.tenant_id
)
]
invoices = await repo.list_entities(session, filters=filters)
return [inv.model_dump() for inv in invoices]
Frontend Multi-Tenancy
Extracting Tenant from JWT
If your frontend receives a JWT token, extract tenant information:
// src/utils/auth.ts
interface JWTPayload {
sub: string;
tenant_id: string;
tenant_attributes: {
plan: string;
features: string[];
max_users: number;
};
}
export function decodeToken(token: string): JWTPayload {
const base64Url = token.split(".")[1];
const base64 = base64Url.replace(/-/g, "+").replace(/_/g, "/");
const jsonPayload = decodeURIComponent(
atob(base64)
.split("")
.map(c => "%" + ("00" + c.charCodeAt(0).toString(16)).slice(-2))
.join(""),
);
return JSON.parse(jsonPayload);
}
React Context for Tenant
// src/contexts/TenantContext.tsx
import { createContext, useContext, ReactNode } from 'react';
interface TenantAttributes {
plan: string;
features: string[];
max_users?: number;
}
interface TenantContextType {
tenantId: string;
attributes: TenantAttributes;
hasFeature: (feature: string) => boolean;
}
const TenantContext = createContext<TenantContextType | null>(null);
export function TenantProvider({ children }: { children: ReactNode }) {
// Extract from JWT or API response
const token = localStorage.getItem('access_token');
const payload = token ? decodeToken(token) : null;
const value: TenantContextType = {
tenantId: payload?.tenant_id || 'default',
attributes: payload?.tenant_attributes || { plan: 'free', features: [] },
hasFeature: (feature: string) =>
payload?.tenant_attributes.features.includes(feature) || false,
};
return (
<TenantContext.Provider value={value}>
{children}
</TenantContext.Provider>
);
}
export function useTenant() {
const context = useContext(TenantContext);
if (!context) {
throw new Error('useTenant must be used within TenantProvider');
}
return context;
}
Feature-Gated Components
// src/components/FeatureGate.tsx
import { useTenant } from '../contexts/TenantContext';
interface FeatureGateProps {
feature: string;
children: ReactNode;
fallback?: ReactNode;
}
export function FeatureGate({ feature, children, fallback }: FeatureGateProps) {
const { hasFeature } = useTenant();
if (!hasFeature(feature)) {
return fallback || <div>This feature is not available in your plan.</div>;
}
return <>{children}</>;
}
// Usage in components
import { FeatureGate } from './components/FeatureGate';
function Dashboard() {
return (
<div>
<h1>Dashboard</h1>
<FeatureGate feature="advanced_reports">
<AdvancedReports />
</FeatureGate>
<FeatureGate feature="api_access">
<APISettings />
</FeatureGate>
</div>
);
}
Tenant-Specific Branding
// src/hooks/useTenantBranding.ts
import { useTenant } from "../contexts/TenantContext";
interface BrandingConfig {
logo: string;
primaryColor: string;
companyName: string;
}
export function useTenantBranding(): BrandingConfig {
const { attributes } = useTenant();
return {
logo: attributes.logo_url || "/default-logo.svg",
primaryColor: attributes.primary_color || "#3b82f6",
companyName: attributes.company_name || "Framework M",
};
}
// Usage in Navbar
function Navbar() {
const branding = useTenantBranding();
return (
<nav style={{ backgroundColor: branding.primaryColor }}>
<img src={branding.logo} alt={branding.companyName} />
<span>{branding.companyName}</span>
</nav>
);
}
Tenant Selector (Admin UI)
For admin users managing multiple tenants:
// src/components/TenantSelector.tsx
import { useState, useEffect } from 'react';
export function TenantSelector() {
const [tenants, setTenants] = useState<string[]>([]);
const [currentTenant, setCurrentTenant] = useState<string>('');
useEffect(() => {
// Fetch available tenants (admin only)
fetch('/api/admin/tenants')
.then(res => res.json())
.then(data => setTenants(data.tenants));
}, []);
const switchTenant = (tenantId: string) => {
// Store selected tenant in localStorage
localStorage.setItem('admin_selected_tenant', tenantId);
setCurrentTenant(tenantId);
// Reload data for new tenant
window.location.reload();
};
return (
<select value={currentTenant} onChange={(e) => switchTenant(e.target.value)}>
{tenants.map(tenant => (
<option key={tenant} value={tenant}>{tenant}</option>
))}
</select>
);
}
Switching Adapters
Configuration-Based Selection
The simplest approach is to use create_tenant_adapter_from_headers(), which automatically selects the adapter based on framework_config.toml:
from framework_m.adapters.tenant import create_tenant_adapter_from_headers
# In Litestar dependency provider
async def provide_tenant_adapter(request: Request):
"""Provide tenant adapter based on configuration."""
adapter = create_tenant_adapter_from_headers(
headers=request.headers if request else None
)
return adapter
Factory Logic
# framework-m/src/framework_m/adapters/tenant.py
def create_tenant_adapter_from_headers(
headers: Mapping[str, str] | None = None,
) -> ImplicitTenantAdapter | HeaderTenantAdapter:
"""Create appropriate tenant adapter based on config."""
from framework_m.core.interfaces.tenant import is_multi_tenant
# If multi-tenant mode AND headers provided, use HeaderTenantAdapter
if is_multi_tenant() and headers is not None:
return HeaderTenantAdapter(headers=headers)
# Otherwise use ImplicitTenantAdapter (single-tenant mode)
return ImplicitTenantAdapter()
Environment Variable
# .env or framework_config.toml
TENANT_MODE=multi # or "single"
# framework_config.toml
[tenancy]
mode = "multi" # Uses HeaderTenantAdapter
# OR
mode = "single" # Uses ImplicitTenantAdapter
Custom Adapter via Entrypoint
For advanced use cases, register a custom tenant adapter:
# my_app/custom_tenant_adapter.py
from framework_m.core.interfaces.tenant import TenantProtocol, TenantContext
class DatabaseTenantAdapter:
"""Load tenant from database instead of headers."""
def __init__(self, request_id: str):
self.request_id = request_id
def get_current_tenant(self) -> str:
# Look up tenant from database using request_id
return lookup_tenant_from_db(self.request_id)
def get_tenant_attributes(self, tenant_id: str) -> dict:
# Fetch attributes from database
return fetch_tenant_attributes(tenant_id)
Register via entrypoint:
# setup.py or pyproject.toml
[project.entry-points."framework_m.tenant_adapters"]
database = "my_app.custom_tenant_adapter:DatabaseTenantAdapter"
Testing with MockTenantAdapter
# tests/conftest.py
import pytest
from framework_m.core.interfaces.tenant import TenantContext
class MockTenantAdapter:
"""Mock adapter for testing."""
def __init__(self, tenant_id: str = "test-tenant"):
self.tenant_id = tenant_id
def get_current_tenant(self) -> str:
return self.tenant_id
def get_tenant_attributes(self, tenant_id: str) -> dict:
return {
"plan": "enterprise",
"features": ["all"],
}
def get_context(self) -> TenantContext:
return TenantContext(
tenant_id=self.tenant_id,
attributes=self.get_tenant_attributes(self.tenant_id),
is_default=False,
)
@pytest.fixture
def mock_tenant_adapter():
"""Provide mock tenant adapter for tests."""
return MockTenantAdapter(tenant_id="test-tenant")
Best Practices
1. Always Use TenantContext from Request State
# ✅ Good: Use request.state.tenant
async def my_endpoint(request: Request):
tenant_ctx = request.state.tenant
tenant_id = tenant_ctx.tenant_id
# ❌ Bad: Don't try to resolve tenant manually
async def my_endpoint(request: Request):
tenant_id = request.headers.get("x-tenant-id") # Don't do this!
2. Rely on RLS for Data Isolation
# ✅ Good: Let framework handle tenant filtering
invoices = await repo.list_entities(session, user=user)
# ❌ Bad: Manual filtering (fragile and error-prone)
invoices = await session.execute(
select(Invoice).where(Invoice.tenant_id == tenant_id)
)
3. Test Both Tenancy Modes
# Test single-tenant mode
@pytest.mark.parametrize("tenancy_mode", ["single"])
def test_single_tenant(tenancy_mode):
adapter = ImplicitTenantAdapter()
assert adapter.get_current_tenant() == "default"
# Test multi-tenant mode
@pytest.mark.parametrize("tenancy_mode", ["multi"])
def test_multi_tenant(tenancy_mode):
headers = {"x-tenant-id": "acme"}
adapter = HeaderTenantAdapter(headers=headers)
assert adapter.get_current_tenant() == "acme"
4. Document Tenant Attributes Schema
# Define expected tenant attributes in documentation
"""
Tenant Attributes Schema:
{
"plan": str, # "free" | "pro" | "enterprise"
"features": list[str], # ["reports", "api", "webhooks"]
"max_users": int, # Maximum users allowed
"max_storage_gb": int, # Storage quota in GB
"custom_domain": str, # Optional custom domain
}
"""
5. Handle Missing Tenant Gracefully
async def get_tenant_safely(request: Request) -> TenantContext:
"""Get tenant context with fallback."""
try:
return request.state.tenant
except AttributeError:
# Fallback for public endpoints
return TenantContext(
tenant_id="public",
attributes={"plan": "free", "features": []},
is_default=True,
)
Migration Path
From Single-Tenant to Multi-Tenant
If you start with Mode A and need to migrate to Mode B:
- Add tenant_id column to all DocTypes:
# Add to all DocTypes that need isolation
class Invoice(BaseDocType):
tenant_id: str = Field(max_length=100, default="default")
class Meta:
apply_rls = True
rls_field = "tenant_id"
- Generate migration:
m migrate generate "add_tenant_id_to_all_tables"
- Update existing data:
# In migration file
def upgrade():
# Set all existing rows to "default" tenant
op.execute("UPDATE invoice SET tenant_id = 'default' WHERE tenant_id IS NULL")
- Switch configuration:
[tenancy]
mode = "multi" # Switch from "single" to "multi"
- Deploy gateway:
Set up Keycloak/Auth0 to populate X-Tenant-ID headers.
- Test thoroughly:
Ensure existing "default" tenant users still have access.
Troubleshooting
Tenant Not Resolved
Symptom: request.state.tenant is None or missing
Solution:
- Check
framework_config.tomlhas[tenancy]section - Verify
mode = "multi"if using HeaderTenantAdapter - Ensure gateway is populating
X-Tenant-IDheader - Check middleware order (tenant middleware must run early)
Users See Wrong Tenant's Data
Symptom: Users see data from other tenants
Solution:
- Verify
apply_rls = Truein DocType Meta - Check
rls_fieldpoints to correct column (usuallytenant_id) - Ensure queries use
GenericRepository(not raw SQLAlchemy) - Verify user context has correct tenant_id
Feature Flags Not Working
Symptom: Features available when they shouldn't be
Solution:
- Check
X-Tenant-Attributesheader is populated - Verify JSON is valid in attributes header
- Ensure frontend checks
tenant.attributes.featuresarray - Log
tenant_ctx.attributesto debug
Performance Issues
Symptom: Slow queries in multi-tenant mode
Solution:
- Add database index on
tenant_idcolumn:
CREATE INDEX idx_invoice_tenant_id ON invoice(tenant_id);
- Use database-per-tenant for large tenants
- Consider read replicas for heavy tenants
- Cache tenant attributes in Redis
Summary
Framework M's multi-tenancy system provides:
- Two modes: Single-tenant (ImplicitTenantAdapter) and multi-tenant (HeaderTenantAdapter)
- Three isolation patterns: Logical (RLS), database-per-tenant, schema-per-tenant
- Automatic RLS: Framework handles tenant filtering transparently
- Feature flags: Per-tenant attributes control feature availability
- Easy testing: Mock adapters for unit tests
- Migration path: Start single, scale to multi-tenant
Choose the mode and isolation pattern that fits your requirements, and let Framework M handle the complexity.