Risoluzione del contesto del tenant e middleware
Risolva il tenant corrente da sottodomini o token e propaghi il contesto attraverso ogni dipendenza.
Risoluzione del contesto del tenant e middleware è una lezione FastAPI Backend Development Bootcamp gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento FastAPI Backend Development Bootcamp, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso FastAPI Backend Development Bootcamp include 4 lezioni in totale.
Parti di questa lezione non sono ancora state tradotte e vengono mostrate in inglese.
Why Tenant Context Matters
In a multi-tenant SaaS, a single FastAPI process serves many customers. Every request must be scoped to exactly one tenant, and getting this wrong leaks data across organizations.
The core problem: by the time a query runs deep inside a service or repository, the code needs to know which tenant it is acting for. We solve this with tenant context resolution:
- Identify the tenant from the incoming request (subdomain or token).
- Validate that the tenant exists and is active.
- Propagate that identity so every downstream dependency can read it.
This lesson builds that pipeline step by step.
Two Resolution Strategies
There are two common ways to discover the tenant for a request:
- Subdomain-based:
acme.app.commaps to tenantacme. Read from theHostheader. Great for browser sessions and per-tenant branding. - Token-based: a JWT carries a
tenant_id(often as a claim). Ideal for API clients and mobile apps where there is no subdomain.
Mature systems support both, with a clear precedence rule. A common choice: trust the token claim first (it is signed), then fall back to the subdomain. Whatever you pick, document it and enforce it consistently.
Parsing the Tenant from a Subdomain
The subdomain comes from the Host header. We strip the known base domain and take the left-most label. Reserved labels like www and api must be rejected so they are never treated as tenants.
This helper is pure string logic, so it is easy to unit-test in isolation:
BASE_DOMAIN = "app.com"
RESERVED = {"www", "api", "admin", ""}
def tenant_from_host(host: str):
# Strip port if present: 'acme.app.com:8000' -> 'acme.app.com'
host = host.split(":")[0].lower().strip()
if not host.endswith(BASE_DOMAIN):
return None
prefix = host[: -len(BASE_DOMAIN)].rstrip(".")
if not prefix:
return None
label = prefix.split(".")[0]
if label in RESERVED:
return None
return label
for h in ["acme.app.com:8000", "www.app.com", "app.com", "globex.eu.app.com"]:
print(h, "->", tenant_from_host(h))Extracting the Tenant from a JWT Claim
For API clients, the tenant identity rides inside the access token as a signed claim. After verifying the signature, you read the tenant_id claim.
Because the token is signed, this value is trustworthy and should usually win over the subdomain. The snippet below shows the decode-and-read shape (verification is illustrated; in production use your real secret and algorithm):
import base64
import json
def read_unverified_claims(token: str) -> dict:
# A JWT is header.payload.signature, each base64url-encoded.
payload_b64 = token.split(".")[1]
padding = "=" * (-len(payload_b64) % 4)
raw = base64.urlsafe_b64decode(payload_b64 + padding)
return json.loads(raw)
# Demo payload: {"sub": "user-7", "tenant_id": "acme"}
demo = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ1c2VyLTciLCJ0ZW5hbnRfaWQiOiJhY21lIn0.sig"
claims = read_unverified_claims(demo)
print("tenant_id:", claims.get("tenant_id"))Storing Context with contextvars
Once resolved, the tenant must be reachable anywhere in the request without threading it through every function argument. Python's contextvars is the right tool: each request gets its own isolated value, and it works correctly under asyncio concurrency.
We wrap the raw ContextVar in a small accessor that raises if no tenant was set, turning a missing context into a loud, early failure:
from contextvars import ContextVar
_current_tenant: ContextVar[str] = ContextVar("current_tenant")
def set_current_tenant(tenant_id: str) -> None:
_current_tenant.set(tenant_id)
def get_current_tenant() -> str:
try:
return _current_tenant.get()
except LookupError:
raise RuntimeError("No tenant in context")
set_current_tenant("acme")
print("active tenant:", get_current_tenant())The Resolution Middleware
A middleware is the right place to resolve the tenant: it runs before any route handler and sees the raw request. Here we combine token and subdomain with a precedence rule, set the contextvar, and reject anonymous-but-tenant-required traffic early.
Note the try/finally that resets the contextvar token so context never bleeds between requests on the same worker:
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
class TenantMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
tenant = resolve_from_token(request) or tenant_from_host(
request.headers.get("host", "")
)
if tenant is None:
return JSONResponse({"detail": "Tenant not identified"}, status_code=400)
token = _current_tenant.set(tenant)
try:
request.state.tenant_id = tenant
return await call_next(request)
finally:
_current_tenant.reset(token)
# app.add_middleware(TenantMiddleware)Validating the Tenant Exists
Resolving an identifier is not the same as trusting it. A request might carry ghost.app.com for a tenant that was deleted or suspended. Before doing real work you must validate:
- The tenant exists in your registry.
- It is active (not suspended, not past-due if you gate on billing).
Resolve identity in middleware, but do the database lookup in a dependency so it can be cached per-request and reused. Return 404 for unknown tenants and 403 for suspended ones, so you do not leak which tenants exist.
Exposing Tenant via a Dependency
Route handlers should not read the contextvar directly. Instead, expose a FastAPI dependency that returns the validated tenant record. This gives you one place to enforce existence and active-status, and it documents the requirement in the route signature.
The dependency reads the id set by the middleware, loads the tenant, and raises clean HTTP errors otherwise:
from fastapi import Depends, HTTPException
async def get_tenant(tenant_id: str = Depends(get_current_tenant)):
tenant = await tenant_repo.find_by_slug(tenant_id)
if tenant is None:
raise HTTPException(status_code=404, detail="Unknown tenant")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant suspended")
return tenant
@router.get("/projects")
async def list_projects(tenant=Depends(get_tenant)):
return await project_repo.list_for(tenant.id)Propagating Context to the Database
The biggest payoff of a single source of truth for the tenant is automatic data scoping. Two common patterns:
- Application-level filtering: every repository call appends
WHERE tenant_id = :tid, reading the id fromget_current_tenant(). - Postgres Row-Level Security (RLS): set a session variable per request, and policies enforce isolation in the database itself.
With RLS you run SET app.tenant_id at the start of each connection's use, so even a forgotten filter cannot cross tenants:
from contextlib import asynccontextmanager
@asynccontextmanager
async def tenant_scoped_session(session_factory):
tenant_id = get_current_tenant()
async with session_factory() as session:
# Bind tenant for the lifetime of this connection's RLS policies.
await session.execute(
text("SET app.tenant_id = :tid"), {"tid": tenant_id}
)
yield sessionBackground Tasks Lose Context
A subtle trap: contextvars are tied to the current execution context. Code that runs after the response (a Celery task, an asyncio.create_task spawned without the context, or a thread-pool job) will not see the tenant unless you pass it explicitly.
Rule of thumb: capture the tenant id while still inside the request, then hand it to the background unit of work as an explicit argument. Never rely on the contextvar surviving the request boundary.
def enqueue_report(background_tasks, tenant=Depends(get_tenant)):
# Capture the id NOW; the worker runs outside this request's context.
tenant_id = tenant.id
background_tasks.add_task(build_report, tenant_id=tenant_id)
return {"status": "queued"}
async def build_report(tenant_id: str):
# Re-establish context inside the task before touching the DB.
set_current_tenant(tenant_id)
await generate(tenant_id)Middleware Ordering and Testing
Order matters. The tenant middleware must run before authorization and logging so those layers can already see the tenant. In Starlette/FastAPI, middleware added last runs first (it wraps the outermost layer), so add tenant resolution after CORS but ensure it precedes auth in execution.
For tests, drive requests through the real middleware stack and assert isolation:
- Two requests with different subdomains must never see each other's rows.
- An unknown subdomain returns
404; a suspended tenant returns403. - A missing
Hostand no token returns400.
def test_subdomain_isolation(client):
r1 = client.get("/projects", headers={"Host": "acme.app.com"})
r2 = client.get("/projects", headers={"Host": "globex.app.com"})
assert r1.status_code == 200 and r2.status_code == 200
assert r1.json() != r2.json()
def test_unknown_tenant_404(client):
r = client.get("/projects", headers={"Host": "ghost.app.com"})
assert r.status_code == 404Quick Check
Test your understanding of context propagation across the request lifecycle.
Recap
You built a complete tenant context pipeline for multi-tenant FastAPI:
- Resolve the tenant from a JWT claim (preferred, signed) or the subdomain in the
Hostheader, with a clear precedence rule. - Store it in a
contextvarset by middleware, reset in afinallyblock so it never leaks between requests. - Validate existence and active status in a dependency, returning
404and403precisely. - Propagate to the database via per-call filters or Postgres RLS session variables for defense in depth.
- Beware background tasks and threads: pass the tenant id explicitly and re-establish context, because contextvars do not survive the request boundary.
One source of truth for the tenant, enforced at every layer, is what keeps a SaaS from leaking data.
Domande Frequenti
La lezione «Risoluzione del contesto del tenant e middleware» è gratuita?
Sì — il testo completo di «Risoluzione del contesto del tenant e middleware» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso FastAPI Backend Development Bootcamp, passa a CoddyKit PRO. Il corso FastAPI Backend Development Bootcamp include 4 lezioni in totale.
Cosa imparerò in «Risoluzione del contesto del tenant e middleware»?
Risolva il tenant corrente da sottodomini o token e propaghi il contesto attraverso ogni dipendenza. Eserciti FastAPI Backend Development Bootcamp con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare FastAPI Backend Development Bootcamp?
Non è richiesta alcuna esperienza precedente. FastAPI Backend Development Bootcamp su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.
Quanto tempo richiede la lezione «Risoluzione del contesto del tenant e middleware»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione FastAPI Backend Development Bootcamp?
Sì. Ogni lezione FastAPI Backend Development Bootcamp include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Strategie di isolamento dei tenant e relativi compromessi
- Risoluzione del contesto del tenant e middleware
- Sicurezza a livello di riga e partizionamento dei dati
- Misurazione dell’utilizzo, quote e hook di fatturazione