Routers & structure projet

TL;DR — Dans FastAPI, un APIRouter est un mini-FastAPI que tu montes dans l'app principale via include_router(). C'est l'équivalent direct des @Module / @Controller de NestJS : un découpage par domaine métier (et non par couche technique), avec un prefix, des tags, des dependencies et des responses partagés au niveau du router. La structure gagnante pour un dev senior n'est pas « un fichier par endpoint » mais une organisation en packages par feature (users/, agents/, chat/) où chaque feature expose son router, ses schémas Pydantic, son service et ses dépendances. Tu assembles le tout dans une factory create_app() testable, tu câbles les dépendances transverses (auth, DB, client LLM) au bon niveau, et tu évites le piège n°1 : importer l'app dans les routers (imports circulaires). On termine en montant un router chat qui stream les tokens d'un agent Claude via SSE.

🧠 Mental model

Pense à ton app FastAPI comme à un aéroport.

L'app (FastAPI()) est l'aéroport lui-même : un point d'entrée unique, des règles globales (douane = middleware, sécurité = auth).
Chaque APIRouter est un terminal dédié à une compagnie/destination : le terminal agents, le terminal users, le terminal billing. Chaque terminal a son propre préfixe d'enregistrement (/agents), ses portes (routes), et son personnel partagé (dépendances montées au niveau du terminal).
include_router() est le branchement physique du terminal sur l'aéroport. Tant qu'un terminal n'est pas branché, ses portes n'existent pas pour les passagers.

Le point clé venant de NestJS : NestJS t'impose l'arbre de modules par le framework (le DI container résout tout). FastAPI ne t'impose rien — la structure est une convention que tu tiens. C'est plus de liberté et plus de corde pour te pendre. Le rôle d'un senior est de réimposer la discipline que NestJS te donnait gratuitement.

                       ┌─────────────────────────────────────┐
   requête HTTP ──────▶│  FastAPI app  (create_app)           │
                       │  middlewares · exception handlers    │
                       │  dependencies globales (auth, trace) │
                       └───────┬───────────┬───────────┬──────┘
              include_router() │           │           │
                       ┌───────▼──┐  ┌──────▼───┐  ┌────▼─────┐
                       │ users/   │  │ agents/  │  │ chat/    │   ← un package par feature
                       │ router   │  │ router   │  │ router   │
                       │ schemas  │  │ schemas  │  │ schemas  │
                       │ service  │  │ service  │  │ service  │
                       │ deps     │  │ deps     │  │ deps     │
                       └──────────┘  └──────────┘  └──────────┘

Mapping mental NestJS → FastAPI :

NestJS	FastAPI	Note
`@Module()`	un package Python (`agents/`) + son `router`	la cohésion est par convention, pas par décorateur
`@Controller('agents')`	`APIRouter(prefix="/agents", tags=["agents"])`
`@Get()` / `@Post()`	`@router.get()` / `@router.post()`
`@Injectable()` service	une classe/fonction service ordinaire	pas de DI container global
Provider injecté (DI)	`Depends(...)`	résolution par appel, pas par graphe global
`AppModule` qui importe les modules	`create_app()` qui `include_router()`
Guards	`dependencies=[Depends(...)]` au niveau router

La brique de base : `APIRouter`

Un router se déclare, se remplit, puis se monte. Trois temps.

python

# agents/router.py
from fastapi import APIRouter, Depends, HTTPException, status

from agents.dependencies import get_agent_service
from agents.schemas import AgentCreate, AgentRead
from agents.service import AgentService

router = APIRouter(
    prefix="/agents",
    tags=["agents"],
    dependencies=[Depends(get_agent_service)],  # appliqué à TOUTES les routes du router
    responses={404: {"description": "Agent introuvable"}},
)


@router.post("", response_model=AgentRead, status_code=status.HTTP_201_CREATED)
async def create_agent(
    payload: AgentCreate,
    service: AgentService = Depends(get_agent_service),
) -> AgentRead:
    return await service.create(payload)


@router.get("/{agent_id}", response_model=AgentRead)
async def get_agent(
    agent_id: str,
    service: AgentService = Depends(get_agent_service),
) -> AgentRead:
    agent = await service.get(agent_id)
    if agent is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Agent introuvable")
    return agent

Puis on l'assemble dans la factory de l'app — jamais au niveau d'un module global mutable :

python

# app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

from agents.router import router as agents_router
from users.router import router as users_router
from chat.router import router as chat_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    # démarrage : on crée les ressources coûteuses UNE fois (client LLM, pool DB)
    from anthropic import AsyncAnthropic
    app.state.anthropic = AsyncAnthropic()
    yield
    # arrêt : on ferme proprement
    await app.state.anthropic.close()


def create_app() -> FastAPI:
    app = FastAPI(title="Agent Platform", lifespan=lifespan)
    app.include_router(users_router)
    app.include_router(agents_router)
    app.include_router(chat_router)
    return app


app = create_app()

Pourquoi une factory create_app() et pas un app = FastAPI() au niveau module ? Parce que tes tests veulent une app fraîche par cas (avec des dependency_overrides différents), et qu'un singleton module-level rend ça pénible. C'est l'équivalent FastAPI du Test.createTestingModule() de NestJS.

La bonne façon vs. la mauvaise façon

❌ La mauvaise façon — découpage par couche technique (le piège de l'ex-MVC).

app/
  routers/        # tous les endpoints de toutes les features
    users.py
    agents.py
    chat.py
  models/         # tous les schémas
  services/       # tous les services

Le problème : pour toucher la feature agents, tu ouvres 3 dossiers. Le couplage entre features devient invisible. À 40 endpoints, c'est un plat de spaghettis. NestJS t'évitait ça en forçant le module ; ici personne ne te force.

✅ La bonne façon — découpage par feature (vertical slice).

agents/
  __init__.py
  router.py       # APIRouter + endpoints de la feature
  schemas.py      # modèles Pydantic d'entrée/sortie
  service.py      # logique métier (pas de FastAPI ici !)
  dependencies.py # Depends() propres à la feature
  models.py       # ORM / persistance (optionnel)

Règle d'or : service.py ne doit importer ni fastapi ni Request. La logique métier doit être appelable depuis un endpoint, un worker, un cron, un test — sans serveur HTTP. Si ton service lève des HTTPException, tu as collé ta logique au transport. Lève des exceptions métier (AgentNotFound) et traduis-les en HTTP dans un exception handler global.

Câbler les dépendances au bon niveau

FastAPI offre trois niveaux pour attacher une dépendance. Choisir le bon est une décision senior.

python

# 1) Niveau ROUTE — spécifique à un endpoint
@router.get("/{agent_id}")
async def get_agent(agent_id: str, user=Depends(current_user)): ...

# 2) Niveau ROUTER — toutes les routes de la feature (ex: auth sur /agents)
router = APIRouter(prefix="/agents", dependencies=[Depends(require_auth)])

# 3) Niveau APP — toutes les routes (ex: request-id / tracing)
app = FastAPI(dependencies=[Depends(add_request_id)])

Quand dependencies=[...] est utilisé au niveau router/app, la valeur de retour est ignorée : c'est fait pour les guards (effet de bord ou levée d'exception), exactement comme les guards NestJS. Quand tu as besoin de la valeur (le user courant, une session DB), tu déclares Depends() dans la signature de la route.

Le pattern « ressource partagée » : un seul client LLM

Erreur classique de débutant : instancier AsyncAnthropic() dans chaque endpoint. Tu paies une reconstruction de pool de connexions HTTP à chaque requête. La bonne façon : créer le client une fois au démarrage (lifespan), puis l'injecter via une dépendance.

python

# chat/dependencies.py
from anthropic import AsyncAnthropic
from fastapi import Request


def get_anthropic(request: Request) -> AsyncAnthropic:
    # le client a été créé dans lifespan et stocké sur app.state
    return request.app.state.anthropic

python

# chat/service.py — AUCUN import fastapi ici
from anthropic import AsyncAnthropic
from anthropic.types import MessageParam


class ChatService:
    def __init__(self, client: AsyncAnthropic) -> None:
        self._client = client

    async def answer(self, messages: list[MessageParam]) -> str:
        message = await self._client.messages.create(
            model="claude-opus-4-8",
            max_tokens=1024,
            messages=messages,
        )
        # content est une liste de blocs ; on extrait le texte
        return "".join(b.text for b in message.content if b.type == "text")

python

# chat/dependencies.py (suite)
from anthropic import AsyncAnthropic
from fastapi import Depends, Request

from chat.service import ChatService


def get_chat_service(
    client: AsyncAnthropic = Depends(get_anthropic),
) -> ChatService:
    return ChatService(client)

Le service ne sait pas qu'il vit dans une API web. Tu peux le tester sans HTTP, le réutiliser dans un worker, le mocker proprement.

Tester une structure à routers : `dependency_overrides`

C'est là que la factory + le découpage par feature paient. Tu remplaces la dépendance qui fournit le client LLM par un faux, sans toucher au code de prod.

python

# tests/test_chat.py
import pytest
from fastapi.testclient import TestClient

from app import create_app
from chat.dependencies import get_chat_service


class FakeChatService:
    async def answer(self, messages):
        return "réponse déterministe"


@pytest.fixture
def client() -> TestClient:
    app = create_app()
    app.dependency_overrides[get_chat_service] = lambda: FakeChatService()
    return TestClient(app)


def test_chat_endpoint(client: TestClient) -> None:
    resp = client.post("/chat", json={"message": "bonjour"})
    assert resp.status_code == 200
    assert resp.json()["reply"] == "réponse déterministe"

dependency_overrides est l'équivalent FastAPI de overrideProvider().useValue() de NestJS. Aucun appel réseau, aucun coût de token, déterminisme total. Note qu'on override get_chat_service (le point d'injection public), pas get_anthropic — on remplace au niveau le plus haut qui isole le test du reste.

Composition profonde : routers de routers

Un router peut en inclure un autre. Utile pour versionner (/api/v1) ou regrouper un sous-domaine.

python

# api/v1/__init__.py
from fastapi import APIRouter

from agents.router import router as agents_router
from users.router import router as users_router

v1 = APIRouter(prefix="/api/v1")
v1.include_router(users_router)   # devient /api/v1/users
v1.include_router(agents_router)  # devient /api/v1/agents

python

# app.py
app.include_router(v1)

Les prefix se concatènent, les tags et dependencies s'accumulent (ils ne s'écrasent pas). Une dépendance posée sur v1 s'applique à tous les sous-routers — pratique pour exiger l'auth sur tout /api/v1 d'un coup.

⚙️ En production

Imports circulaires (le piège n°1). Si agents/router.py importe app et que app.py importe agents_router, tu as une boucle. Règle : les routers ne connaissent jamais l'app. L'app connaît les routers. Le flux d'import est toujours unidirectionnel : app.py → feature/router.py → feature/service.py. Si une dépendance a besoin de l'app, passe par request.app.state, pas par un import.

Ne crée pas de ressources coûteuses par requête. Client HTTP (httpx/AsyncAnthropic), pool DB, client Redis → créés une fois dans lifespan, stockés sur app.state, injectés via Depends. Un AsyncAnthropic() dans une route, c'est un nouveau pool de connexions TLS à chaque appel : latence + fuite de sockets sous charge. Et ferme-les dans la partie yield du lifespan, sinon tu traînes des connexions ouvertes au shutdown.

Sync vs async — le piège silencieux. Une route déclarée def (sync) est exécutée dans un threadpool ; une route async def tourne sur l'event loop. Si tu mets un appel bloquant (time.sleep, requests.get, un driver DB sync) dans une async def, tu bloques l'event loop entier et tu effondres le débit de tout le serveur. Règle : I/O async → async def + libs async ; code bloquant inévitable → def (threadpool) ou await anyio.to_thread.run_sync(...). Pour un agent LLM qui stream, c'est forcément async def + AsyncAnthropic.

Sécurité — auth au bon niveau. Mets le guard d'auth sur le router (dependencies=[Depends(require_auth)]) plutôt que de le répéter sur 20 routes : impossible d'oublier une route. Pour les routes publiques (healthcheck, login), mets-les dans un router séparé sans le guard. Ne mets jamais de clé API (ANTHROPIC_API_KEY) en dur : le SDK la lit de l'environnement.

Observabilité. Un dependencies=[Depends(add_request_id)] au niveau app pose un identifiant de corrélation sur chaque requête. Les tags des routers alimentent la doc OpenAPI et le regroupement dans /docs. Pour un agent, logge le usage retourné par Claude (message.usage.input_tokens / output_tokens) par requête — c'est ton coût en clair.

Failure modes côté LLM. Le SDK Anthropic retry déjà les 429/5xx avec backoff (max_retries=2 par défaut). Mappe les exceptions typées vers des codes HTTP propres dans un exception handler global, plutôt que de laisser une stacktrace fuiter :

python

import anthropic
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


def register_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(anthropic.RateLimitError)
    async def _rate_limited(request: Request, exc: anthropic.RateLimitError):
        return JSONResponse(status_code=429, content={"error": "upstream rate limited"})

    @app.exception_handler(anthropic.APIError)
    async def _upstream(request: Request, exc: anthropic.APIError):
        return JSONResponse(status_code=502, content={"error": "LLM upstream error"})

Tradeoff senior : monolithe modulaire vs micro-routers. Garde tout dans une app à routers tant que tu peux (déploiement unique, refactor trivial). Le découpage par feature te donne déjà la frontière : le jour où chat/ doit scaler indépendamment, tu extrais le package en service séparé sans réécrire la logique, parce que service.py n'a jamais dépendu de l'app. Ne sur-découpe pas trop tôt.

Tie-in IA : un router `chat` qui stream un agent Claude (SSE + tool-use)

Voici un router complet qui stream les tokens d'un agent. On utilise messages.stream() (recommandé pour toute sortie longue : ça évite les timeouts de requête), thinking adaptatif, et l'effort. Le streaming token par token passe par StreamingResponse en SSE.

python

# chat/schemas.py
from pydantic import BaseModel, Field


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=10_000)


class ChatReply(BaseModel):
    reply: str
    input_tokens: int
    output_tokens: int

python

# chat/router.py
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse

from anthropic import AsyncAnthropic
from chat.dependencies import get_anthropic
from chat.schemas import ChatReply, ChatRequest

router = APIRouter(prefix="/chat", tags=["chat"])


@router.post("", response_model=ChatReply)
async def chat(
    payload: ChatRequest,
    client: AsyncAnthropic = Depends(get_anthropic),
) -> ChatReply:
    # réponse non streamée : on attend le message final
    async with client.messages.stream(
        model="claude-opus-4-8",
        max_tokens=1024,
        thinking={"type": "adaptive"},
        messages=[{"role": "user", "content": payload.message}],
    ) as stream:
        message = await stream.get_final_message()

    text = "".join(b.text for b in message.content if b.type == "text")
    return ChatReply(
        reply=text,
        input_tokens=message.usage.input_tokens,
        output_tokens=message.usage.output_tokens,
    )


@router.post("/stream")
async def chat_stream(
    payload: ChatRequest,
    client: AsyncAnthropic = Depends(get_anthropic),
) -> StreamingResponse:
    async def event_source():
        async with client.messages.stream(
            model="claude-opus-4-8",
            max_tokens=2048,
            thinking={"type": "adaptive", "display": "summarized"},
            messages=[{"role": "user", "content": payload.message}],
        ) as stream:
            # text_stream ne yield QUE les deltas de texte visible
            async for delta in stream.text_stream:
                yield f"data: {delta}\n\n"
            yield "event: done\ndata: [DONE]\n\n"

    return StreamingResponse(event_source(), media_type="text/event-stream")

Pour une boucle tool-use (l'agent appelle tes fonctions), le squelette : tu boucles tant que stop_reason == "tool_use", tu exécutes chaque bloc tool_use, et tu renvoies les tool_result dans un message user. Garde la boucle dans le service, jamais dans le router.

python

# agents/service.py — boucle tool-use minimale, sans dépendance FastAPI
from anthropic import AsyncAnthropic
from anthropic.types import MessageParam, ToolParam

TOOLS: list[ToolParam] = [
    {
        "name": "get_weather",
        "description": "Donne la météo actuelle. Appelle-la dès que l'utilisateur "
                       "demande le temps qu'il fait dans une ville.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    }
]


async def run_agent(client: AsyncAnthropic, user_msg: str) -> str:
    messages: list[MessageParam] = [{"role": "user", "content": user_msg}]
    for _ in range(10):  # garde-fou : jamais de boucle infinie
        resp = await client.messages.create(
            model="claude-opus-4-8",
            max_tokens=1024,
            thinking={"type": "adaptive"},
            tools=TOOLS,
            messages=messages,
        )
        if resp.stop_reason != "tool_use":
            return "".join(b.text for b in resp.content if b.type == "text")

        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for block in resp.content:
            if block.type == "tool_use":
                output = await execute_tool(block.name, block.input)  # ton code
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})
    raise RuntimeError("boucle tool-use non convergente")

Pour de la sortie structurée (extraction, classification), préfère client.messages.parse() avec un schéma plutôt qu'un prefill d'assistant (les prefills renvoient un 400 sur Opus 4.8). Le usage te donne le coût ; à 5 $/Mtok en entrée et 25 $/Mtok en sortie pour claude-opus-4-8, logge-le par requête.

🏋️ Exercices

Exercice 1 — Refactor vertical slice (implémenter)

Objectif. On te donne un main.py monolithique de 200 lignes avec users + agents + chat mélangés. Découpe-le en packages par feature (users/, agents/, chat/), chacun avec router.py, schemas.py, service.py, dependencies.py. Assemble via create_app(). Aucun endpoint ne doit changer de chemin. Indice/Solution. Commence par les schemas (zéro dépendance), puis service (importe schemas), puis dependencies (importe service), puis router (importe tout). Le flux d'import doit être un DAG, jamais une boucle. Vérifie avec python -c "import app" qu'il n'y a pas d'import circulaire.

Exercice 2 — Auth au niveau router (implémenter)

Objectif. Crée un router public (/health, /login) sans auth, et protège tout /api/v1 avec un guard require_token posé une seule fois. Une route oubliée doit être impossible. Indice/Solution. v1 = APIRouter(prefix="/api/v1", dependencies=[Depends(require_token)]). Le guard ne retourne rien : il lève HTTPException(401) si le header Authorization est absent/invalide. Teste qu'une route nouvellement ajoutée à v1 est protégée sans modification.

Exercice 3 — Client LLM partagé via lifespan (production-grade)

Objectif. Pars d'un code qui fait AsyncAnthropic() dans chaque route. Refactore pour créer le client une fois dans lifespan, l'injecter via Depends(get_anthropic), et le fermer au shutdown. Mesure la différence de latence p50 sur 100 requêtes. Indice/Solution. Stocke sur app.state.anthropic. get_anthropic lit request.app.state.anthropic. La fermeture (await client.close()) va après le yield. Tu devrais voir disparaître le coût de handshake TLS répété.

Exercice 4 — Test sans réseau (production-grade)

Objectif. Écris un test du endpoint /chat qui ne fait aucun appel à l'API Anthropic et ne consomme aucun token, via dependency_overrides. Indice/Solution. Override get_chat_service (ou get_anthropic) par un fake déterministe. Si tu n'arrives pas à override proprement, c'est le signe que tu instancies le client trop bas dans la stack — remonte l'injection.

Exercice 5 — Stream SSE token par token (break-then-fix)

Objectif. Monte le endpoint /chat/stream. Puis introduis le bug : remplace async def event_source par une version qui utilise requests.get (bloquant) au milieu. Observe que tout le serveur se fige sous 2 requêtes concurrentes. Corrige. Indice/Solution. Le code bloquant dans une coroutine gèle l'event loop. Fix : tout en async (AsyncAnthropic, stream.text_stream). Si tu dois appeler du bloquant, await anyio.to_thread.run_sync(...). Vérifie avec deux curl en parallèle sur /chat/stream.

Exercice 6 — Boucle tool-use bornée (break-then-fix)

Objectif. Implémente run_agent avec tool-use. Puis casse-le : retire le garde-fou for _ in range(10) et fournis un outil qui renvoie toujours une erreur, de sorte que l'agent ré-appelle l'outil sans fin. Constate la boucle infinie / l'explosion de coût. Corrige. Indice/Solution. Borne le nombre d'itérations et propage l'erreur outil avec is_error: true dans le tool_result pour que le modèle puisse changer de stratégie. Logge usage à chaque tour pour voir le coût grimper avant le fix.

🎤 En entretien

Q : Pourquoi include_router() plutôt que tout déclarer sur l'app ? R : Pour le découpage par domaine — prefix/tags/dependencies partagés, doc OpenAPI groupée, et surtout une frontière nette qui permet d'extraire une feature en service séparé sans réécrire la logique.

Q : Découpage par couche technique ou par feature, et pourquoi ? R : Par feature (vertical slice). Le par-couche disperse une feature sur 3 dossiers et masque le couplage ; le par-feature colocalise router/schema/service et garde les imports en DAG.

Q : Comment éviter de recréer le client LLM / le pool DB à chaque requête ? R : On le crée une fois dans le lifespan, on le stocke sur app.state, on l'injecte via Depends, et on le ferme au shutdown — sinon on paie un handshake TLS par requête et on fuit des sockets sous charge.

Q : Quand mettre une dépendance au niveau route, router ou app ? R : Route = spécifique et valeur utilisée dans la signature ; router = guard transverse à une feature (auth sur /agents) ; app = transverse global sans valeur de retour (request-id, tracing). Les dependencies=[...] au niveau router/app ignorent la valeur retournée : c'est pour les guards.

Routers & structure projet ​

🧠 Mental model ​

La brique de base : APIRouter ​

La bonne façon vs. la mauvaise façon ​

Câbler les dépendances au bon niveau ​

Le pattern « ressource partagée » : un seul client LLM ​

Tester une structure à routers : dependency_overrides ​

Composition profonde : routers de routers ​

⚙️ En production ​

Tie-in IA : un router chat qui stream un agent Claude (SSE + tool-use) ​

🏋️ Exercices ​

Exercice 1 — Refactor vertical slice (implémenter) ​

Exercice 2 — Auth au niveau router (implémenter) ​

Exercice 3 — Client LLM partagé via lifespan (production-grade) ​

Exercice 4 — Test sans réseau (production-grade) ​

Exercice 5 — Stream SSE token par token (break-then-fix) ​

Exercice 6 — Boucle tool-use bornée (break-then-fix) ​

🎤 En entretien ​

Routers & structure projet

🧠 Mental model

La brique de base : `APIRouter`

La bonne façon vs. la mauvaise façon

Câbler les dépendances au bon niveau

Le pattern « ressource partagée » : un seul client LLM

Tester une structure à routers : `dependency_overrides`

Composition profonde : routers de routers

⚙️ En production

Tie-in IA : un router `chat` qui stream un agent Claude (SSE + tool-use)

🏋️ Exercices

Exercice 1 — Refactor vertical slice (implémenter)

Exercice 2 — Auth au niveau router (implémenter)

Exercice 3 — Client LLM partagé via lifespan (production-grade)

Exercice 4 — Test sans réseau (production-grade)

Exercice 5 — Stream SSE token par token (break-then-fix)

Exercice 6 — Boucle tool-use bornée (break-then-fix)

🎤 En entretien