Multi-Tenant FastAPI: Tenant Isolation, RLS, and Per-Tenant LLM Routing for SaaS AI in 2026
TL;DR — Multi-tenant FastAPI: RLS, per-tenant routing, quotas. fastapi-tenancy: "Four isolation strategies: schema, database, RLS, hybrid. Resolution: header, subdomain, path, JWT. contextvars propagation. Raw ASGI middleware." SolidNumber: "Shared database, shared schema, shared backend. One PostgreSQL, one FastAPI. Isolation in application layer and database policies. 298 tables with RLS. Fernet-encrypted per-tenant LLM keys. SmartRouter validates company match." fastapi-rls-multi-tenant: "RLS at database layer — isolation guaranteed even if app has bugs. FORCE RLS. Two DB roles. Transaction-local GUCs — auto-reset on commit." Dev.to: "Atomic UPSERT counters. Redis for ephemeral, PostgreSQL for durable. Every query filters by tenant_id. Super admin does NOT see other tenants." Learn more with FastAPI tutorial, authentication, rate limiting, and monitoring.
SolidNumber frames the architecture choice: "We chose the hardest multi-tenancy model for the best economics: shared database, shared schema, shared backend. One PostgreSQL database serves every company. One FastAPI server handles every request. The isolation is entirely in the application layer and database policies."
fastapi-tenancy provides the library approach: "One library, four isolation strategies, zero per-route boilerplate. Schema-per-tenant, database-per-tenant, PostgreSQL RLS, hybrid. Resolution: header, subdomain, URL path, JWT claim."
Multi-Tenant Architecture
tenant_id in token
extracted in middleware
cannot be forged"] Header["Header
X-Tenant-ID
service-to-service"] Subdomain["Subdomain
tenant.example.com
DNS-based"] Path["URL Path
/t/{tenant_id}/
explicit routing"] end subgraph Middleware["FastAPI Middleware"] Extract["Tenant Extraction
raw ASGI middleware
zero buffering
contextvars propagation
binds tenant to request"] Context["Tenant Context
contextvars-based
propagates through
background tasks
streaming responses"] end subgraph Isolation["Isolation Strategies"] Schema["Schema-per-Tenant
one PG schema per tenant
namespace isolation
Alembic per-tenant migrations"] Database["Database-per-Tenant
separate DB per tenant
strongest isolation
highest cost"] RLS["PostgreSQL RLS
shared database
Row-Level Security
FORCE RLS
DB-level enforcement"] Hybrid["Hybrid
mix by tier
free=shared RLS
enterprise=database
per-tenant config"] end subgraph DataLayer["Data Isolation"] GUC["Transaction-Local GUC
set_config('app.tenant_id', id, true)
auto-reset on commit
no stale context in pool"] Query["Query Scoping
WHERE tenant_id = X
every query filtered
no exceptions
dependency injection"] Roles["DB Roles
app_owner: DDL only
app_user: DML only
least-privilege runtime
never superuser"] end subgraph AI["Per-Tenant AI Isolation"] LLMKeys["LLM Credentials
Fernet-encrypted AES-128
per-tenant API keys
llm_providers table
company_id binding"] Router["SmartRouter
validate company match
before API call
per-tenant model config
tier-based routing"] Memory["Memory Isolation
Redis key namespacing
tenant:{id}:memory:{key}
no wildcard queries
per-tenant conversations"] Vector["Vector Isolation
pre-filtering in search
WHERE tenant_id=X
exclude other tenants
from similarity calculation"] end subgraph Quota["Per-Tenant Quotas"] Counters["Atomic Counters
UPSERT ON CONFLICT
tenant_monthly_usage
tokens, requests, cost
durable in PostgreSQL"] RateLimit["Rate Limits
Redis sliding window
per-tenant key
tiered: free/pro/enterprise
ephemeral in Redis"] Plans["Tiered Plans
free: 10K tokens/day
pro: 1M tokens/day
enterprise: unlimited
fail-closed enforcement"] end JWT --> Extract Header --> Extract Subdomain --> Extract Path --> Extract Extract --> Context Context --> Schema Context --> Database Context --> RLS Context --> Hybrid RLS --> GUC GUC --> Query Query --> Roles Context --> LLMKeys LLMKeys --> Router Context --> Memory Context --> Vector Context --> Counters Context --> RateLimit Counters --> Plans RateLimit --> Plans style Resolution fill:#4169E1,color:#fff style Isolation fill:#39FF14,color:#000 style AI fill:#2D1B69,color:#fff style Quota fill:#FF6B6B,color:#fff
Isolation Strategy Comparison
| Strategy | Isolation | Cost | Complexity | Best For |
|---|---|---|---|---|
| Schema-per-tenant | Namespace | Medium | Medium | Mid-tier SaaS |
| Database-per-tenant | Strongest | High | High | Enterprise, regulated |
| RLS (shared) | DB-level | Low | Medium | Most SaaS AI |
| Hybrid | Tiered | Variable | High | Multi-tier platform |
| App-layer only | App-level | Lowest | Low | Simple apps |
Implementation
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class IsolationStrategy(Enum):
SCHEMA = "schema"
DATABASE = "database"
RLS = "rls"
HYBRID = "hybrid"
@dataclass
class MultiTenantFastAPIGuide:
"""Multi-tenant FastAPI implementation guide."""
def get_tenant_middleware(self) -> str:
"""Tenant resolution middleware."""
return (
"# === TENANT MIDDLEWARE ===\n"
"from fastapi import FastAPI,\n"
" Request, HTTPException\n"
"from starlette.types import (\n"
" ASGIApp, Receive, Scope,\n"
" Send)\n"
"import contextvars\n"
"import jwt\n"
"\n"
"# Context variable for tenant\n"
"tenant_ctx = contextvars.ContextVar(\n"
" 'tenant_id', default=None)\n"
"\n"
"class TenantMiddleware:\n"
" '''Raw ASGI middleware for\n"
" tenant resolution.''' \n"
" \n"
" def __init__(self, app):\n"
" self.app = app\n"
" \n"
" async def __call__(\n"
" self, scope, receive, send):\n"
" if scope['type'] != 'http':\n"
" await self.app(\n"
" scope, receive, send)\n"
" return\n"
" \n"
" # Extract tenant from JWT\n"
" headers = dict(\n"
" scope.get('headers', []))\n"
" auth = headers.get(\n"
" b'authorization', b'')\n"
" \n"
" if auth.startswith(\n"
" b'Bearer '):\n"
" token = (\n"
" auth[7:].decode())\n"
" try:\n"
" payload = (\n"
" jwt.decode(\n"
" token, SECRET_KEY,\n"
" algorithms=['HS256']))\n"
" tenant_id = (\n"
" payload.get(\n"
" 'tenant_id'))\n"
" if not tenant_id:\n"
" raise HTTPException(\n"
" 403,\n"
" 'No tenant')\n"
" # Set context var\n"
" token_ctx = (\n"
" tenant_ctx.set(\n"
" tenant_id))\n"
" scope['tenant_id'] = (\n"
" tenant_id)\n"
" except jwt.JWTError:\n"
" pass # Let auth dep handle\n"
" \n"
" # Also check X-Tenant-ID\n"
" # header\n"
" if not scope.get(\n"
" 'tenant_id'):\n"
" tid = headers.get(\n"
" b'x-tenant-id')\n"
" if tid:\n"
" scope['tenant_id'] = (\n"
" tid.decode())\n"
" tenant_ctx.set(\n"
" tid.decode())\n"
" \n"
" await self.app(\n"
" scope, receive, send)\n"
"\n"
"app = FastAPI()\n"
"app.add_middleware(\n"
" TenantMiddleware)"
)
def get_rls_setup(self) -> str:
"""PostgreSQL RLS setup."""
return (
"# === POSTGRESQL RLS ===\n"
"\n"
"# SQL: Enable RLS on tables\n"
"-- ALTER TABLE conversations\n"
"-- ENABLE ROW LEVEL SECURITY;\n"
"-- ALTER TABLE conversations\n"
"-- FORCE ROW LEVEL SECURITY;\n"
"-- \n"
"-- CREATE POLICY tenant_isolation\n"
"-- ON conversations\n"
"-- USING (\n"
"-- tenant_id =\n"
"-- current_setting(\n"
"-- 'app.tenant_id')::uuid)\n"
"-- WITH CHECK (\n"
"-- tenant_id =\n"
"-- current_setting(\n"
"-- 'app.tenant_id')::uuid);\n"
"\n"
"# Python: Set GUC per request\n"
"from sqlalchemy.ext.asyncio\n"
" import AsyncSession\n"
"from fastapi import Depends\n"
"\n"
"async def get_tenant_session(\n"
" request: Request,\n"
" session: AsyncSession = (\n"
" Depends(get_session))):\n"
" '''Set tenant GUC for RLS.''' \n"
" tenant_id = (\n"
" request.scope.get('tenant_id'))\n"
" if tenant_id:\n"
" # Transaction-local GUC\n"
" # auto-resets on commit\n"
" await session.execute(\n"
" text(\n"
" \"SELECT set_config(\"\n"
" \"'app.tenant_id', \"\n"
" \":tid, true)\"),\n"
" {'tid': str(tenant_id)})\n"
" yield session\n"
"\n"
"# Two DB roles\n"
"# app_owner: DDL (migrations)\n"
"# app_user: DML (runtime)\n"
"# Never use superuser\n"
"# for app connections"
)
def get_tenant_dependency(self) -> str:
"""Tenant dependency injection."""
return (
"# === TENANT DEPENDENCY ===\n"
"\n"
"from fastapi import Depends\n"
"from pydantic import BaseModel\n"
"\n"
"class TenantContext(BaseModel):\n"
" tenant_id: str\n"
" tier: str = 'free'\n"
" rate_limit: int = 10\n"
" token_quota: int = 100_000\n"
"\n"
"async def get_tenant(\n"
" request: Request\n"
") -> TenantContext:\n"
" '''Get tenant context.''' \n"
" tenant_id = (\n"
" request.scope.get('tenant_id'))\n"
" if not tenant_id:\n"
" raise HTTPException(\n"
" status_code=403,\n"
" detail='No tenant context')\n"
" \n"
" # Load tenant config from DB\n"
" tenant = await (\n"
" load_tenant_config(tenant_id))\n"
" return TenantContext(\n"
" tenant_id=tenant_id,\n"
" tier=tenant.tier,\n"
" rate_limit=tenant.rate_limit,\n"
" token_quota=(\n"
" tenant.token_quota))\n"
"\n"
"# Usage in endpoints\n"
"@app.post('/v1/chat')\n"
"async def chat(\n"
" prompt: str,\n"
" tenant: TenantContext = (\n"
" Depends(get_tenant)),\n"
" db: AsyncSession = Depends(\n"
" get_tenant_session)):\n"
" '''Chat with tenant\n"
" isolation.''' \n"
" # RLS auto-filters by tenant\n"
" # No manual WHERE needed\n"
" result = await call_llm(\n"
" prompt, tenant)\n"
" \n"
" # Track usage\n"
" await track_usage(\n"
" db, tenant.tenant_id,\n"
" result['tokens'])\n"
" \n"
" return result"
)
def get_per_tenant_llm(self) -> str:
"""Per-tenant LLM routing."""
return (
"# === PER-TENANT LLM ROUTING ===\n"
"from cryptography.fernet (\n"
" import Fernet)\n"
"\n"
"fernet = Fernet(ENCRYPTION_KEY)\n"
"\n"
"async def get_tenant_llm_config(\n"
" tenant_id: str) -> dict:\n"
" '''Get per-tenant LLM config.''' \n"
" # Load from llm_providers table\n"
" row = await db.fetch_one(\n"
" 'SELECT * FROM '\n"
" 'llm_providers WHERE '\n"
" 'tenant_id = ?',\n"
" tenant_id)\n"
" \n"
" if not row:\n"
" # Use platform default\n"
" return {\n"
" 'provider': 'ollama',\n"
" 'model': 'llama3.2',\n"
" 'api_key': None,\n"
" 'base_url': (\n"
" 'http://'\n"
" 'localhost:11434')\n"
" }\n"
" \n"
" # Decrypt API key\n"
" api_key = fernet.decrypt(\n"
" row['api_key_enc']).decode()\n"
" \n"
" return {\n"
" 'provider': row['provider'],\n"
" 'model': row['model'],\n"
" 'api_key': api_key,\n"
" 'base_url': row['base_url'],\n"
" }\n"
"\n"
"async def call_tenant_llm(\n"
" prompt: str,\n"
" tenant: TenantContext):\n"
" '''Call LLM with per-tenant\n"
" routing.''' \n"
" config = await (\n"
" get_tenant_llm_config(\n"
" tenant.tenant_id))\n"
" \n"
" # SmartRouter: validate\n"
" # company match\n"
" # before API call\n"
" \n"
" async with httpx.AsyncClient(\n"
" timeout=300.0\n"
" ) as client:\n"
" if config['provider'] == 'ollama':\n"
" resp = await client.post(\n"
" f\"{config['base_url']}\"\n"
" '/api/chat',\n"
" json={\n"
" 'model': config[\n"
" 'model'],\n"
" 'messages': [{\n"
" 'role':'user',\n"
" 'content':prompt}]})\n"
" else:\n"
" resp = await client.post(\n"
" f\"{config['base_url']}\"\n"
" '/v1/chat/completions',\n"
" headers={\n"
" 'Authorization':\n"
" f\"Bearer \"\n"
" f\"{config['api_key']}\"},\n"
" json={\n"
" 'model': config[\n"
" 'model'],\n"
" 'messages': [{\n"
" 'role':'user',\n"
" 'content':prompt}]})\n"
" return resp.json()\n"
"\n"
"# Redis key namespacing\n"
"# tenant:{id}:memory:{key}\n"
"# tenant:{id}:conversation:{id}\n"
"# No wildcard queries"
)
def get_quota_enforcement(self) -> str:
"""Per-tenant quota enforcement."""
return (
"# === QUOTA ENFORCEMENT ===\n"
"\n"
"async def check_tenant_quota(\n"
" tenant: TenantContext,\n"
" estimated_tokens: int) -> bool:\n"
" '''Check tenant token quota.''' \n"
" # Atomic UPSERT counter\n"
" await db.execute(\n"
" '''INSERT INTO \n"
" tenant_monthly_usage\n"
" (tenant_id, \n"
" billing_period, tokens)\n"
" VALUES (?, ?, ?)\n"
" ON CONFLICT \n"
" (tenant_id, \n"
" billing_period)\n"
" DO UPDATE SET \n"
" tokens = \n"
" tenant_monthly_usage\n"
" .tokens + ?\n"
" ''',\n"
" tenant.tenant_id,\n"
" current_period(),\n"
" estimated_tokens,\n"
" estimated_tokens)\n"
" \n"
" # Check limit\n"
" row = await db.fetch_one(\n"
" 'SELECT tokens FROM '\n"
" 'tenant_monthly_usage '\n"
" 'WHERE tenant_id = ? '\n"
" 'AND billing_period = ?',\n"
" tenant.tenant_id,\n"
" current_period())\n"
" \n"
" return (\n"
" row['tokens'] <= \n"
" tenant.token_quota)\n"
"\n"
"# Redis rate limit per tenant\n"
"async def tenant_rate_limit(\n"
" tenant_id: str,\n"
" limit: int,\n"
" window: int) -> bool:\n"
" '''Per-tenant rate limit.''' \n"
" key = (\n"
" f'rate:tenant:{tenant_id}')\n"
" current = await redis.incr(key)\n"
" if current == 1:\n"
" await redis.expire(\n"
" key, window)\n"
" return current <= limit"
)
def get_production_tips(self) -> dict:
"""Production tips."""
return {
"shared_best": "Shared database + RLS for best economics. One PostgreSQL, one FastAPI. Isolation in app layer and DB policies.",
"rls_force": "FORCE ROW LEVEL SECURITY — even table owner can't bypass. Two DB roles: app_owner (DDL), app_user (DML).",
"guc_transaction": "Transaction-local GUC: set_config('app.tenant_id', id, true). Auto-resets on commit. No stale context in pool.",
"jwt_tenant": "JWT carries tenant_id — derived from authenticated session, cannot be forged. No extra DB lookup.",
"contextvars": "contextvars-based tenant context propagates through background tasks and streaming responses.",
"fernet_keys": "Per-tenant LLM API keys: Fernet-encrypted (AES-128). SmartRouter validates company match before API call.",
"redis_namespace": "Redis key namespacing: tenant:{id}:memory:{key}. No wildcard queries. Per-tenant conversations.",
"vector_filter": "Vector store pre-filtering: WHERE tenant_id=X in similarity search. Exclude other tenants entirely.",
"atomic_upsert": "Atomic UPSERT counters for quota: ON CONFLICT DO UPDATE. Durable in PostgreSQL. Redis for ephemeral rate limits.",
"super_admin": "Super admin does NOT see other tenants' data — deliberate privacy decision. Sees aggregate stats only.",
}
Multi-Tenant FastAPI Checklist
- [ ] Tenant isolation strategies: schema-per-tenant, database-per-tenant, PostgreSQL RLS, hybrid (mix by tier)
- [ ] Shared database + RLS for best economics: one PostgreSQL, one FastAPI, isolation in app layer and DB policies
- [ ] Database-per-tenant: strongest isolation, highest cost — for enterprise or regulated industries
- [ ] Schema-per-tenant: namespace isolation, per-tenant Alembic migrations — for mid-tier SaaS
- [ ] Hybrid: mix strategies by tier (free=shared RLS, enterprise=database) — for multi-tier platform
- [ ] Tenant resolution: JWT claim (tenant_id in token), header (X-Tenant-ID), subdomain, URL path
- [ ] JWT carries tenant_id — derived from authenticated session, cannot be forged
- [ ] Middleware extracts tenant before any controller logic — controllers receive from middleware, not request params
- [ ] No code path where query runs without tenant filter
- [ ] contextvars-based tenant context — propagates through background tasks and streaming
- [ ] Raw ASGI middleware — zero buffering, correct ContextVar propagation (not BaseHTTPMiddleware)
- [ ] PostgreSQL RLS: CREATE POLICY tenant_isolation USING (tenant_id = current_setting('app.tenant_id')::uuid)
- [ ] FORCE ROW LEVEL SECURITY — even table owner cannot bypass RLS
- [ ] Two DB roles: app_owner for DDL (migrations), app_user for DML (runtime) — least-privilege
- [ ] Never use superuser for application connections — superusers bypass RLS
- [ ] Transaction-local GUC: SELECT set_config('app.tenant_id', tenant_id, true) — auto-resets on commit
- [ ] GUC auto-reset means connection returns to pool clean — no stale tenant context
- [ ] Even if application-layer filtering has a bug, database rejects cross-tenant queries
- [ ] Every table containing user data has tenant_id column — no exceptions
- [ ] Every query filters by tenant_id — no exceptions, enforced by dependency injection
- [ ] Per-tenant LLM API keys: Fernet-encrypted (AES-128) storage in llm_providers table
- [ ] SmartRouter validates agent company matches provider company before any API call
- [ ] Routing bug cannot accidentally use another company's credentials
- [ ] Per-tenant model config: different models per tenant tier
- [ ] Redis key namespacing: f'tenant:{tenant_id}:memory:{key}' — no wildcard queries
- [ ] Vector store pre-filtering: WHERE tenant_id=X in similarity search — exclude other tenants entirely
- [ ] Per-tenant agent instances with company-specific settings
- [ ] Conversations: company_id + agent_id pair, unique conversation IDs prevent cross-tenant access
- [ ] KB Scope: agents restricted to specific knowledge base categories per company
- [ ] Per-tenant quotas: atomic UPSERT counters (ON CONFLICT DO UPDATE) for token usage
- [ ] tenant_monthly_usage table with UNIQUE constraint on (tenant_id, billing_period)
- [ ] Redis for ephemeral (rate limits, concurrency), PostgreSQL for durable (billing, usage)
- [ ] Tiered plans: free, pro, enterprise with different token quotas and rate limits
- [ ] Fail-closed for quota enforcement — reject if over limit
- [ ] Super admin does NOT see other tenants' data — deliberate privacy decision, sees aggregate stats only
- [ ] RBAC with dependency injection: require_role, require_permission factories — authorization impossible to skip
- [ ] Per-tenant Alembic migrations with concurrency control
- [ ] In-process LRU + TTL L1 cache per tenant, Redis L2
- [ ] JWT algorithm-confusion prevention, SQL-injection-safe DDL, anti-enumeration errors
- [ ] Three-layer logical sandbox: edge (JWT middleware), database (WHERE tenant_id), vector (pre-filtering)
- [ ] Registration and tenant creation before tenant context exists — RLS disabled on those tables
- [ ] Read FastAPI tutorial for API setup
- [ ] Read authentication for JWT and RBAC
- [ ] Read rate limiting for per-tenant limits
- [ ] Read monitoring for per-tenant metrics
- [ ] Test: tenant A cannot access tenant B's data (RLS enforcement)
- [ ] Test: JWT without tenant_id is rejected
- [ ] Test: GUC resets after transaction — no stale tenant in pool
- [ ] Test: per-tenant LLM keys are encrypted and company-validated
- [ ] Test: quota enforcement rejects over-limit requests
- [ ] Test: vector search excludes other tenants
- [ ] Test: super admin cannot read other tenants' conversations
- [ ] Test: rate limits apply per-tenant, not globally
- [ ] Document isolation strategy, resolution mechanism, RLS policies, quota schema, routing logic
FAQ
What are the main tenant isolation strategies for FastAPI AI applications?
Schema-per-tenant, database-per-tenant, PostgreSQL RLS, and hybrid. fastapi-tenancy: "Four isolation strategies: schema-per-tenant, database-per-tenant, PostgreSQL RLS, hybrid (mix by tier). Resolution: header, subdomain, URL path, JWT claim. SQLAlchemy async, Redis, in-memory. contextvars-based propagation through background tasks and streaming. Raw ASGI middleware — zero buffering." SolidNumber: "Chose hardest multi-tenancy model for best economics: shared database, shared schema, shared backend. One PostgreSQL serves every company. One FastAPI server handles every request. Isolation entirely in application layer and database policies." Strategies: (1) Schema-per-tenant: one PostgreSQL schema per tenant, namespace isolation. (2) Database-per-tenant: separate database per tenant, strongest isolation. (3) RLS: shared database, PostgreSQL Row-Level Security enforces tenant_id filtering at DB level. (4) Hybrid: mix strategies by tier (free=shared RLS, enterprise=database). (5) Shared with app-layer: WHERE company_id=X on every query.
How do you implement PostgreSQL Row-Level Security for tenant isolation in FastAPI?
Set transaction-local GUC variables per request and let PostgreSQL enforce filtering. fastapi-rls-multi-tenant: "RLS at database layer — tenant isolation guaranteed even if application logic has bugs. FORCE ROW LEVEL SECURITY applies RLS even to table owner. Two DB roles: app_owner for DDL, app_user for DML. JWT carries tenant_id + user_id. get_rls_session sets GUC vars as transaction-local — automatically reset when transaction commits, connection returns to pool clean." SolidNumber: "298 tables have PostgreSQL RLS policies enforcing company_id filtering. Even if application-layer filtering has a bug, database rejects cross-tenant queries. Session variable app.current_company_id set at request time from JWT." RLS: (1) CREATE POLICY tenant_isolation ON table USING (tenant_id = current_setting('app.tenant_id')::uuid). (2) FORCE ROW LEVEL SECURITY — even owner can't bypass. (3) Two DB roles: app_owner (DDL), app_user (DML). (4) Set GUC: SELECT set_config('app.tenant_id', tenant_id, true) — transaction-local. (5) GUC auto-resets on commit — no stale tenant context in pool. (6) JWT carries tenant_id, no extra DB lookup.
How do you resolve and inject tenant context in FastAPI middleware?
Extract tenant_id from JWT, header, subdomain, or path, and inject via contextvars. fastapi-tenancy: "Resolution: header (X-Tenant-ID), subdomain, URL path, JWT claim — or bring your own. contextvars-based — propagates through background tasks and streaming. Raw ASGI middleware — zero buffering, correct ContextVar propagation." SolidNumber: "Every authenticated request carries JWT with company_id. Middleware extracts before any controller logic. company_id derived from authenticated session — cannot forge JWT. Controllers receive company from middleware, not from request parameters. No code path where query runs without company filter." Bravado: "Layer 1: Edge — FastAPI Middleware extracts tenant_id from JWT, binds to request lifecycle. Ensures application identity is immutable and validated before reaching business logic." Resolution: (1) JWT claim: tenant_id in token, extracted in middleware. (2) Header: X-Tenant-ID for service-to-service. (3) Subdomain: tenant.example.com. (4) URL path: /t/{tenant_id}/. (5) contextvars for propagation through async tasks. (6) Raw ASGI middleware — no buffering.
How do you implement per-tenant LLM routing and credential isolation?
Store encrypted per-tenant API keys, validate company match before routing, and namespace all AI resources. SolidNumber: "Companies that bring their own LLM API keys get Fernet-encrypted storage (AES-128). Each company keys stored in llm_providers table with company_id. SmartRouter validates agent company matches provider company before any API call. Routing bug cannot accidentally use another company credentials. Agents: company-specific settings. Conversations: company_id + agent_id pair. Memory: keyed by company_id in Redis and PostgreSQL. No wildcard queries. KB Scope: agents restricted to specific knowledge base categories per company." Routing: (1) Per-tenant LLM keys: Fernet-encrypted (AES-128) in DB. (2) SmartRouter: validate company match before API call. (3) Per-tenant model config: different models per tenant tier. (4) Redis key namespacing: f'tenant:{tenant_id}:memory:{key}'. (5) Vector store pre-filtering: WHERE tenant_id=X in similarity search. (6) Per-tenant agent instances with company-specific settings.
How do you enforce per-tenant quotas and rate limits in FastAPI?
Use atomic UPSERT counters for usage tracking and Redis sliding window for rate limits. Dev.to: "Atomic counters (UPSERT) so limits respected under concurrency. Redis for ephemeral (rate limits, concurrency) and PostgreSQL for durable (billing, usage). tenant_monthly_usage table with UNIQUE constraint on (tenant_id, billing_period). ON CONFLICT DO UPDATE. Clear roles with dependency injection — authorization impossible to skip. Isolation by default — every query filters by tenant_id, no exceptions. Super admin does NOT see other tenants data — deliberate privacy decision." Quotas: (1) Atomic UPSERT: INSERT INTO tenant_monthly_usage ... ON CONFLICT (tenant_id, billing_period) DO UPDATE SET tokens = tenant_monthly_usage.tokens + excluded.tokens. (2) Redis sliding window for per-tenant rate limits. (3) Per-tenant token quota: check before LLM call. (4) Tiered plans: free, pro, enterprise with different limits. (5) Fail-closed for quota enforcement. (6) Billing: track tokens, requests, cost per tenant.
Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →