Bootcamp backendontwikkeling met FastAPI · Les

Tenantcontextresolutie en middleware

Bepaal de huidige tenant uit subdomeinen of tokens en geef de context door aan elke dependency.

Les 2 van 413 stappen

Tenantcontextresolutie en middleware is een gratis Bootcamp backendontwikkeling met FastAPI-les op CoddyKit. Dit is les 2 van 4. Je kunt 3 lessen uit dit leerpad gratis volledig lezen — daarna ontgrendelt CoddyKit PRO alle lessen, plus praktische oefeningen met een ingebouwde code-editor en een AI-tutor die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject Bootcamp backendontwikkeling met FastAPI. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus Bootcamp backendontwikkeling met FastAPI bevat in totaal 4 lessen.

Waarom tenantcontext belangrijk is

In een multi-tenant SaaS bedient één FastAPI-proces veel klanten. Elk verzoek moet precies aan één tenant worden gebonden; als dat misgaat, lekken gegevens tussen organisaties.

Het kernprobleem: tegen de tijd dat een query diep in een service of repository wordt uitgevoerd, moet de code weten voor welke tenant deze handelt. We lossen dit op met het bepalen van de tenantcontext:

Identificeren — bepaal de tenant aan de hand van het binnenkomende verzoek (subdomein of token).
Valideren — controleer of de tenant bestaat en actief is.
Doorgeven — geef die identiteit door zodat elke volgende dependency deze kan lezen.

In deze les bouwen we die pijplijn stap voor stap op.

Twee strategieën om de tenant te bepalen

Er zijn twee gebruikelijke manieren om de tenant van een verzoek te achterhalen:

Op basis van een subdomein: acme.app.com verwijst naar tenant acme. Lees dit uit de Host-header. Ideaal voor browsersessies en branding per tenant.
Op basis van een token: een JWT bevat een tenant_id (vaak als claim). Ideaal voor API-clients en mobiele apps zonder subdomein.

Volwassen systemen ondersteunen beide, met een duidelijke regel voor voorrang. Een gebruikelijke keuze: vertrouw eerst op de tokenclaim (die is ondertekend) en val daarna terug op het subdomein. Welke keuze je ook maakt, documenteer deze en dwing haar consequent af.

De tenant uit een subdomein parseren

Het subdomein komt uit de Host-header. We verwijderen het bekende basisdomein en nemen het meest linkse label. Gereserveerde labels zoals www en api moeten worden afgewezen, zodat ze nooit als tenants worden behandeld.

Deze helper bevat alleen logica voor tekenreeksen en is daarom eenvoudig afzonderlijk met eenheidstests te testen:

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))

De tenant uit een JWT-claim halen

Voor API-clients zit de tenantidentiteit als een ondertekende claim in het toegangstoken. Nadat je de handtekening hebt geverifieerd, lees je de claim tenant_id uit.

Omdat het token is ondertekend, is deze waarde betrouwbaar en hoort deze meestal voorrang te krijgen op het subdomein. Het onderstaande fragment laat de vorm voor decoderen en uitlezen zien (de verificatie wordt geïllustreerd; gebruik in productie je echte geheim en algoritme):

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"))

Context opslaan met contextvars

Zodra de tenant is bepaald, moet deze overal in het verzoek beschikbaar zijn zonder hem als argument door elke functie door te geven. Python's contextvars is hiervoor het juiste hulpmiddel: elk verzoek krijgt zijn eigen geïsoleerde waarde en het werkt correct bij gelijktijdige uitvoering met asyncio.

We verpakken de onbewerkte ContextVar in een kleine accessor die een fout veroorzaakt als er geen tenant is ingesteld. Zo wordt ontbrekende context direct en duidelijk gemeld:

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())

De middleware voor het bepalen van de tenant

Middleware is de juiste plek om de tenant te bepalen: deze wordt vóór elke route-handler uitgevoerd en ziet het onbewerkte verzoek. Hier combineren we het token en het subdomein met een prioriteitsregel, stellen we de contextvariabele in en weigeren we vroegtijdig anoniem verkeer waarvoor een tenant vereist is.

Let op de try/finally die het token van de contextvariabele reset, zodat context niet tussen verzoeken op dezelfde worker blijft hangen:

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)

Controleren of de tenant bestaat

Een identificatie bepalen is niet hetzelfde als erop vertrouwen. Een verzoek kan bijvoorbeeld ghost.app.com bevatten voor een tenant die is verwijderd of opgeschort. Voordat je echt werk uitvoert, moet je het volgende valideren:

De tenant bestaat in je register.
De tenant is actief (niet opgeschort en niet achterstallig als je facturering als voorwaarde gebruikt).

Bepaal de identiteit in middleware, maar voer het opzoeken in de database uit in een dependency, zodat je het per verzoek kunt cachen en hergebruiken. Retourneer 404 voor onbekende tenants en 403 voor opgeschorte tenants, zodat je niet prijsgeeft welke tenants bestaan.

De tenant beschikbaar maken via een dependency

Route-handlers horen de contextvariabele niet rechtstreeks te lezen. Stel in plaats daarvan een FastAPI-dependency beschikbaar die het gevalideerde tenantrecord retourneert. Zo heb je één plek om het bestaan en de actieve status af te dwingen en wordt de vereiste gedocumenteerd in de routesignatuur.

De dependency leest de id die door de middleware is ingesteld, laadt de tenant en veroorzaakt anders duidelijke HTTP-fouten:

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)

Context doorgeven aan de database

De grootste winst van één bron van waarheid voor de tenant is automatische gegevensafbakening. Er zijn twee veelgebruikte patronen:

Filteren op toepassingsniveau: elke aanroep van een repository voegt WHERE tenant_id = :tid toe en leest de id uit get_current_tenant().
Postgres Row-Level Security (RLS): stel per verzoek een sessievariabele in en laat beleidsregels de isolatie in de database zelf afdwingen.

Met RLS voer je aan het begin van elk gebruik van een verbinding SET app.tenant_id uit, zodat zelfs een vergeten filter geen gegevens tussen tenants kan laten lekken:

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

Achtergrondtaken verliezen hun context

Een subtiele valkuil: contextvariabelen zijn gekoppeld aan de huidige uitvoeringscontext. Code die na het antwoord wordt uitgevoerd (een Celery-taak, een asyncio.create_task die zonder de context is gestart, of een taak in een threadpool) ziet de tenant niet, tenzij je deze expliciet doorgeeft.

Vuistregel: leg de tenant-id vast terwijl je nog binnen het verzoek werkt en geef deze vervolgens als expliciet argument door aan de achtergrondtaak. Vertrouw er nooit op dat de contextvariabele de verzoekgrens overleeft.

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)

Volgorde van middleware en tests

De volgorde is belangrijk. De tenant-middleware moet vóór autorisatie en logboekregistratie worden uitgevoerd, zodat die lagen de tenant al kunnen zien. In Starlette/FastAPI wordt middleware die als laatste is toegevoegd als eerste uitgevoerd (deze vormt de buitenste laag). Voeg tenantbepaling dus na CORS toe, maar zorg dat deze in de uitvoering vóór authenticatie komt.

Laat voor tests verzoeken door de echte middlewarestapel lopen en controleer de isolatie:

Twee verzoeken met verschillende subdomeinen mogen nooit elkaars rijen zien.
Een onbekend subdomein retourneert 404; een opgeschorte tenant retourneert 403.
Een ontbrekende Host zonder token retourneert 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

Korte controle

Test je begrip van het doorgeven van context gedurende de levenscyclus van een verzoek.

Beantwoord dit in de interactieve les

Samenvatting

Je hebt een complete tenantcontextpijplijn gebouwd voor multi-tenant FastAPI:

Bepaal de tenant aan de hand van een JWT-claim (voorkeur, ondertekend) of het subdomein in de Host-header, met een duidelijke prioriteitsregel.
Sla deze op in een contextvar die door middleware wordt ingesteld en in een finally-blok wordt gereset, zodat deze nooit tussen verzoeken lekt.
Valideer het bestaan en de actieve status in een dependency en retourneer precies 404 of 403.
Geef de tenant door aan de database via filters per aanroep of sessievariabelen van Postgres RLS voor extra bescherming.
Let op achtergrondtaken en threads: geef de tenant-id expliciet door en stel de context opnieuw in, omdat contextvariabelen de verzoekgrens niet overleven.

Eén bron van waarheid voor de tenant, die in elke laag wordt afgedwongen, voorkomt dat een SaaS gegevens laat lekken.

Gratis beginnen

Leer Bootcamp backendontwikkeling met FastAPI met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen: 21
Lessen: 84

Gratis beginnen met leren Bekijk de volledige cursus

Veelgestelde vragen

Is de les “Tenantcontextresolutie en middleware” gratis?

Ja — je kunt hier op het web alle 3 lessen van het leerpad Bootcamp backendontwikkeling met FastAPI, waaronder “Tenantcontextresolutie en middleware”, gratis volledig lezen. Daarna ontgrendelt CoddyKit PRO alle lessen, plus interactieve oefeningen met een ingebouwde code-editor en een AI-tutor die 24/7 beschikbaar is. De cursus Bootcamp backendontwikkeling met FastAPI bevat in totaal 4 lessen.

Wat leer ik in “Tenantcontextresolutie en middleware”?

Bepaal de huidige tenant uit subdomeinen of tokens en geef de context door aan elke dependency. Je oefent met Bootcamp backendontwikkeling met FastAPI door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met Bootcamp backendontwikkeling met FastAPI te beginnen?

Ervaring vooraf is niet nodig. Bootcamp backendontwikkeling met FastAPI op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 2 van 4.

Hoe lang duurt de les “Tenantcontextresolutie en middleware”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over Bootcamp backendontwikkeling met FastAPI?

Ja. Elke les over Bootcamp backendontwikkeling met FastAPI bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

Strategieën en afwegingen voor tenantisolatie
Tenantcontextresolutie en middleware
Beveiliging en partitionering op rijniveau
Gebruiksmeting, quota's en billinghooks

← Terug naar Bootcamp backendontwikkeling met FastAPI