fastapi-tenancy
Production-ready multi-tenancy for FastAPI.
Four isolation strategies. Five resolution strategies. Full async support. 95%+ test coverage.
Four isolation strategies
Schema, Database, Row-Level Security, and Hybrid — all with automatic DDL and per-connection setup.
Five resolution strategies
Header, Subdomain, Path, JWT, and fully custom resolvers via a simple protocol.
Async-first
Built on SQLAlchemy 2.0 async, asyncio-safe LRU engine cache, and non-blocking Redis throughout.
SQL-injection safe
Every schema name, database name, and identifier is validated against a strict allowlist before any DDL runs.
Protocol-driven
TenantStore, BaseTenantResolver, and BaseIsolationProvider are runtime-checkable protocols — swap any component freely.
Two-level cache
In-process LRU cache with optional Redis write-through layer. Hot tenant lookups cost microseconds, not milliseconds.
Rate limiting
Per-tenant sliding-window rate limiting backed by Redis atomic Lua scripts. Zero additional dependencies beyond Redis.
Migrations
Bounded-concurrent Alembic migrations across all tenants. Migrate 1 000 tenants in minutes, not hours.
Fully tested
95%+ branch coverage. Unit, integration, and end-to-end test suites targeting Python 3.11–3.13.
At a glance¶
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_tenancy import TenancyConfig, TenancyManager, TenancyMiddleware
from fastapi_tenancy.dependencies import get_current_tenant, make_tenant_db_dependency
from fastapi_tenancy.storage.database import SQLAlchemyTenantStore
# 1. Configure
config = TenancyConfig(
database_url="postgresql+asyncpg://user:pass@localhost/myapp",
resolution_strategy="header", # read X-Tenant-ID header
isolation_strategy="schema", # schema-per-tenant
)
# 2. Wire
store = SQLAlchemyTenantStore(config.database_url)
manager = TenancyManager(config, store)
# 3. App
@asynccontextmanager
async def lifespan(app: FastAPI):
await manager.initialize()
yield
await manager.close()
app = FastAPI(lifespan=lifespan)
app.add_middleware(TenancyMiddleware, manager=manager, excluded_paths=["/health"])
# 4. Tenant-scoped session per request
get_db = make_tenant_db_dependency(manager)
@app.get("/orders")
async def list_orders(session: Annotated[AsyncSession, Depends(get_db)]):
# session is already pointing at the current tenant's schema
result = await session.execute(select(Order))
return result.scalars().all()
Isolation strategies¶
| Strategy | Mechanism | Databases | Best for |
|---|---|---|---|
schema |
Dedicated schema per tenant, search_path set per-connection |
PostgreSQL, MSSQL, SQLite* | Most SaaS apps |
database |
Separate database per tenant, LRU engine cache | PostgreSQL, MySQL | Strict regulatory isolation |
rls |
Shared tables, PostgreSQL RLS policies, SET LOCAL GUC |
PostgreSQL only | Large tenant counts |
hybrid |
Route by tenant tier — premium → schema, standard → rls | PostgreSQL | Tiered SaaS products |
*SQLite falls back to table-name prefixing.