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

flowchart TD subgraph Resolution["Tenant Resolution"] JWT["JWT Claim
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 →