PricingLoginStart free
FastAPI Backend Development Bootcamp · 课时

租户上下文解析与中间件

从子域名或令牌中解析当前租户，并将上下文传递给每个依赖项。

第 2 / 4 课13 个步骤

租户上下文解析与中间件 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

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.com maps to tenant acme. Read from the Host header. 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 from get_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 session

Background 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 returns 403.
  • A missing Host and no token returns 400.
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 == 404

Quick 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 Host header, with a clear precedence rule.
  • Store it in a contextvar set by middleware, reset in a finally block so it never leaks between requests.
  • Validate existence and active status in a dependency, returning 404 and 403 precisely.
  • 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.

免费开始

用 AI 导师学习 FastAPI Backend Development Bootcamp — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程
21
课程
84
免费开始学习查看完整课程

常见问题解答

「租户上下文解析与中间件」课时是免费的吗？

是的 — 「租户上下文解析与中间件」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

「租户上下文解析与中间件」这节课中我会学到什么？

从子域名或令牌中解析当前租户，并将上下文传递给每个依赖项。 你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 FastAPI Backend Development Bootcamp 需要有经验吗？

无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「租户上下文解析与中间件」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？

能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 租户隔离策略与权衡
  2. 租户上下文解析与中间件
  3. 行级安全与数据分区
  4. 用量计量、配额与计费钩子
← 返回 FastAPI Backend Development Bootcamp