Skip to content

Gestion des erreurs

TL;DR — Dans une API FastAPI, une erreur n'est pas un try/except posé au hasard : c'est une frontière contractuelle entre ton domaine et le client HTTP. Tu lèves des exceptions métier typées au plus près du problème, tu les traduis en réponses HTTP cohérentes via des exception handlers centralisés, et tu n'écris jamais except Exception: return {"error": ...} à la main dans chaque endpoint. La règle d'or : fail fast, fail typed, fail observable. En production, ça veut dire un format d'erreur stable (un schéma Pydantic, pas un dict improvisé), des request_id corrélables dans les logs, des secrets jamais fuités dans le detail, et pour un agent LLM — distinguer les erreurs retryables (429, 529, timeout réseau) des erreurs fatales (400, 401, refus de safety) parce que la stratégie de retry n'est pas la même.

Tu viens de PHP/TS. Tu connais le réflexe « j'attrape tout, je renvoie un JSON ». FastAPI te pousse vers quelque chose de plus discipliné, et c'est tant mieux : la gestion d'erreurs est l'endroit où une API junior et une API senior se distinguent le plus vite.


🧠 Mental model

Pense à ton API comme à un bâtiment avec un comptoir d'accueil.

À l'intérieur (ton code métier, tes services, ton appel à l'agent Claude), les gens parlent leur langue technique : « la connexion DB a sauté », « le modèle a refusé », « cet utilisateur n'existe pas ». Ce sont des exceptions métier — riches, typées, pleines de contexte.

Le comptoir d'accueil, c'est ta couche de traduction (exception_handlers). Son seul boulot : prendre une exception interne et la transformer en quelque chose que le visiteur HTTP comprend — un code de statut, un corps JSON propre, un message poli. Le visiteur n'a pas à savoir que c'est PostgreSQL qui a planté ; il reçoit un 503 avec « service temporairement indisponible ».

Le piège junior, c'est de mettre un comptoir d'accueil dans chaque bureau : un try/except dans chaque endpoint, chacun inventant son propre format. Le résultat : 12 formats d'erreur différents, des secrets qui fuient, et un client qui ne peut rien parser de manière fiable.

        Couche HTTP (le client, le navigateur, l'agent appelant)

                          │  réponse JSON normalisée + status code

   ┌──────────────────────┴───────────────────────┐
   │   EXCEPTION HANDLERS  (le comptoir d'accueil)  │   ← traduction centralisée
   │   exception métier  ──►  JSONResponse propre   │
   └──────────────────────▲───────────────────────┘
                          │  raise DomainError(...)

   ┌──────────────────────┴───────────────────────┐
   │   Couche métier / services / appel LLM         │   ← lève des exceptions typées
   │   (ne sait RIEN du HTTP, ne renvoie pas de 404)│
   └────────────────────────────────────────────────┘

Cette séparation est la chose à intérioriser : le code métier lève, la frontière traduit. Tout le reste découle de ça.


La mécanique de base

Le contrat HTTP : quel code pour quelle erreur

Avant tout code, fixons le vocabulaire — c'est ce qu'on attend d'un senior en entretien.

CodeSensRetryable ?Côté responsable
400Requête malformée / validation métierNonClient
401Non authentifiéNonClient
403Authentifié mais interditNonClient
404Ressource introuvableNonClient
409Conflit d'état (doublon, version périmée)NonClient
422Validation du schéma (FastAPI le fait pour toi)NonClient
429Rate limitOui (avec backoff)Client
500Bug non géré chez toiNon*Serveur
503Dépendance indisponible (DB, LLM down)OuiServeur

* Un 500 n'est pas « retryable » au sens où retenter ne corrigera pas le bug — mais le client peut légitimement réessayer plus tard. La nuance compte pour un agent.

HTTPException : le cas simple, et sa limite

FastAPI fournit HTTPException. C'est l'outil correct pour les cas triviaux où l'endpoint est la frontière.

python
from fastapi import FastAPI, HTTPException

app = FastAPI()

# Faux store en mémoire pour l'exemple
_users: dict[int, dict[str, str]] = {1: {"name": "Alice"}}


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict[str, str]:
    user = _users.get(user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

C'est très bien pour une route isolée. Mais voici la mauvaise habitude qu'il faut tuer tout de suite :

python
# ❌ ANTI-PATTERN : try/except généraliste dans l'endpoint
@app.get("/users/{user_id}")
async def get_user_wrong(user_id: int) -> dict:
    try:
        user = fetch_user_from_db(user_id)   # logique métier
        return user
    except Exception as exc:                 # on attrape TOUT
        return {"error": str(exc)}           # status 200 !!! + fuite du message brut

Trois péchés mortels ici :

  1. Status 200 alors qu'il y a eu une erreur — le client croit que tout va bien.
  2. str(exc) fuite potentiellement une stack trace, un nom de table SQL, un chemin de fichier — cadeau pour un attaquant.
  3. Format ad hoc ({"error": ...}) différent du {"detail": ...} de FastAPI : ton frontend doit gérer deux schémas.

La version senior : le service lève une exception métier, et un handler centralisé la traduit. On y vient.

Exceptions métier + handlers centralisés (l'idiome senior)

On définit des exceptions de domaine qui ne savent rien du HTTP. Puis on enregistre des handlers qui les traduisent.

python
from __future__ import annotations

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel


# --- 1. Le schéma d'erreur stable (contrat avec le client) ---
class ErrorBody(BaseModel):
    code: str            # ex: "user_not_found" — stable, parsable par le client
    message: str         # message humain, jamais une stack trace
    request_id: str | None = None


# --- 2. Les exceptions métier (zéro dépendance à FastAPI) ---
class DomainError(Exception):
    """Base de toutes nos erreurs métier."""
    status_code: int = 400
    code: str = "domain_error"

    def __init__(self, message: str) -> None:
        super().__init__(message)
        self.message = message


class UserNotFound(DomainError):
    status_code = 404
    code = "user_not_found"


class QuotaExceeded(DomainError):
    status_code = 429
    code = "quota_exceeded"


# --- 3. La frontière : un seul handler pour toute la famille ---
app = FastAPI()


@app.exception_handler(DomainError)
async def handle_domain_error(request: Request, exc: DomainError) -> JSONResponse:
    request_id = request.headers.get("x-request-id")
    body = ErrorBody(code=exc.code, message=exc.message, request_id=request_id)
    return JSONResponse(status_code=exc.status_code, content=body.model_dump())

Maintenant le service métier devient limpide — il lève, il ne traduit pas :

python
async def load_user(user_id: int) -> dict[str, str]:
    user = _users.get(user_id)
    if user is None:
        raise UserNotFound(f"no user with id={user_id}")
    return user


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict[str, str]:
    return await load_user(user_id)   # aucun try/except ici. La frontière s'en charge.

Le gain est énorme : tu ajoutes une nouvelle erreur métier (PaymentDeclined(DomainError)) et elle est traduite automatiquement, sans toucher un seul endpoint. Une seule source de vérité pour le format d'erreur.

Surcharger les erreurs de validation (422)

Par défaut, FastAPI renvoie un 422 avec un format {"detail": [...]} détaillé via RequestValidationError. C'est verbeux et incohérent avec ton ErrorBody. On l'aligne :

python
from fastapi.exceptions import RequestValidationError


@app.exception_handler(RequestValidationError)
async def handle_validation_error(
    request: Request, exc: RequestValidationError
) -> JSONResponse:
    body = ErrorBody(
        code="validation_error",
        message="Request payload failed validation.",
        request_id=request.headers.get("x-request-id"),
    )
    # On expose les détails de validation dans un champ dédié, pas dans message
    content = body.model_dump() | {"errors": exc.errors()}
    return JSONResponse(status_code=422, content=content)

Le filet de sécurité : le 500 non géré

Tout ce que tu n'as pas anticipé doit finir en 500 propre — surtout pas en stack trace exposée. On enregistre un handler attrape-tout, mais on logge l'exception complète côté serveur (pour toi) tout en renvoyant un message neutre (pour le client).

python
import logging
import uuid

logger = logging.getLogger("api")


@app.exception_handler(Exception)
async def handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
    # Un id de corrélation pour retrouver la trace dans les logs
    incident_id = str(uuid.uuid4())
    logger.exception(
        "unhandled_exception",
        extra={"incident_id": incident_id, "path": request.url.path},
    )
    body = ErrorBody(
        code="internal_error",
        message=f"Unexpected error. Reference: {incident_id}",
    )
    return JSONResponse(status_code=500, content=body.model_dump())

⚠️ Subtilité FastAPI/Starlette : le handler Exception attrape-tout n'intercepte pas les exceptions levées dans un BackgroundTask ni, par défaut, certaines erreurs survenant après le début du streaming. Et si tu utilises un middleware custom, l'ordre d'exécution change le comportement. On y revient en production.


L'angle agent IA : gérer les erreurs d'un appel LLM

C'est ici que ton stack (FastAPI qui sert un agent Claude) rencontre la réalité. Un appel à un LLM est un appel réseau vers un service tiers facturé au token : il va échouer, parfois, et la façon dont tu gères ces échecs détermine si ton agent est fiable ou pas.

Le SDK officiel anthropic lève des exceptions typées — n'écris jamais de matching sur le message d'erreur.

Exception SDKCode HTTP sourceRetryable ?Quoi faire
BadRequestError400NonBug dans ta requête — logge, ne retente pas
AuthenticationError401NonClé API invalide — alerte ops
PermissionDeniedError403NonModèle/feature non autorisé
NotFoundError404NonID de modèle erroné
RateLimitError429OuiBackoff (lis retry-after)
InternalServerError500OuiBackoff
OverloadedError529OuiBackoff, voire fallback de modèle
APIErrortouteBase — attrape en dernier

Le SDK retente déjà automatiquement les 429 et 5xx avec backoff exponentiel (max_retries=2 par défaut). Tu n'as donc pas à recoder un retry naïf — tu configures max_retries et tu gères ce qui reste après épuisement des retries.

Endpoint qui appelle l'agent, avec traduction d'erreurs propre

python
import os

import anthropic
from anthropic import AsyncAnthropic
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

# Un seul client, partagé, async — pas un nouveau client par requête.
client = AsyncAnthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    max_retries=3,          # le SDK gère le backoff sur 429/5xx
    timeout=60.0,           # plafond dur par requête
)


class AskBody(BaseModel):
    question: str


class AnswerBody(BaseModel):
    answer: str


@app.post("/ask")
async def ask(body: AskBody) -> AnswerBody:
    message = await client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        thinking={"type": "adaptive"},
        messages=[{"role": "user", "content": body.question}],
    )

    # ⚠️ TOUJOURS vérifier stop_reason AVANT de lire content[0]
    if message.stop_reason == "refusal":
        # Le modèle a refusé pour raison de safety. content peut être vide.
        raise LLMRefused("The model declined to answer this request.")

    text = "".join(b.text for b in message.content if b.type == "text")
    return AnswerBody(answer=text)

Et les exceptions + handlers dédiés à la couche LLM :

python
class LLMError(DomainError):
    status_code = 502        # Bad Gateway : c'est l'upstream qui a un souci
    code = "llm_error"


class LLMUnavailable(LLMError):
    status_code = 503        # retryable côté client
    code = "llm_unavailable"


class LLMRefused(LLMError):
    status_code = 422        # ce n'est pas une panne : la requête est refusée
    code = "llm_refused"


@app.exception_handler(anthropic.APIStatusError)
async def handle_anthropic_error(
    request: Request, exc: anthropic.APIStatusError
) -> JSONResponse:
    # On classe par type, pas par parsing de string
    if isinstance(exc, (anthropic.RateLimitError, anthropic.OverloadedError)):
        mapped = LLMUnavailable("Upstream model is overloaded, retry shortly.")
    elif isinstance(exc, anthropic.AuthenticationError):
        logger.error("anthropic_auth_failed")          # alerte ops, pas le client
        mapped = LLMError("Upstream auth error.")
    else:
        mapped = LLMError(f"Upstream error ({exc.status_code}).")

    body = ErrorBody(
        code=mapped.code,
        message=mapped.message,
        request_id=request.headers.get("x-request-id"),
    )
    # On peut propager le retry-after de l'upstream au client
    headers = {}
    if isinstance(exc, anthropic.RateLimitError):
        retry_after = exc.response.headers.get("retry-after")
        if retry_after:
            headers["retry-after"] = retry_after
    return JSONResponse(
        status_code=mapped.status_code, content=body.model_dump(), headers=headers
    )

Le point qui sépare junior et senior : un endpoint qui appelle un LLM doit toujours vérifier stop_reason avant de lire message.content[0]. Sur un refus de safety, content peut être un tableau vide — content[0].text lèverait un IndexError que tu finirais par servir en 500 mystérieux. La vérification stop_reason == "refusal" est la première ligne de défense.

Streaming SSE : les erreurs après le premier octet

Quand tu streames les tokens de l'agent (Server-Sent Events), un nouveau problème apparaît : une fois que tu as commencé à streamer, tu as déjà envoyé 200 OK. Tu ne peux plus changer le status code. Une erreur en plein flux ne peut donc pas se traduire en 503 ; elle doit s'envoyer dans le flux sous forme d'événement d'erreur.

python
import json
from collections.abc import AsyncIterator

from fastapi.responses import StreamingResponse


@app.post("/ask/stream")
async def ask_stream(body: AskBody) -> StreamingResponse:
    async def event_source() -> AsyncIterator[str]:
        try:
            async with client.messages.stream(
                model="claude-opus-4-8",
                max_tokens=2048,
                thinking={"type": "adaptive"},
                messages=[{"role": "user", "content": body.question}],
            ) as stream:
                async for text in stream.text_stream:
                    yield f"event: token\ndata: {json.dumps({'text': text})}\n\n"

                final = await stream.get_final_message()
                if final.stop_reason == "refusal":
                    yield f"event: error\ndata: {json.dumps({'code': 'llm_refused'})}\n\n"
                else:
                    yield "event: done\ndata: {}\n\n"

        except anthropic.APIStatusError as exc:
            # Trop tard pour un status HTTP — on émet un événement d'erreur
            logger.exception("stream_failed", extra={"status": exc.status_code})
            payload = json.dumps({"code": "llm_unavailable"})
            yield f"event: error\ndata: {payload}\n\n"

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

Le client front doit donc écouter un event: error dans le flux, en plus de gérer un éventuel 503 initial (si l'erreur survient avant le premier token). C'est exactement la dualité qu'attend un intervieweur senior : « où peut tomber l'erreur — avant ou après le premier octet ? — et qu'est-ce que ça change ? »


⚙️ En production

Modes de défaillance à anticiper

  • Le handler Exception n'attrape pas tout. Les exceptions dans un BackgroundTask s'exécutent après la réponse — elles ne passent jamais par tes handlers. Logge-les explicitement dans la tâche. De même, une erreur dans le générateur d'un StreamingResponse ne peut plus modifier le status (cf. SSE ci-dessus).
  • Ordre middleware vs handlers. Si tu ajoutes un middleware custom qui appelle call_next() dans un try/except, tu peux court-circuiter tes exception_handlers. Règle : laisse les handlers gérer la traduction, garde les middlewares pour le cross-cutting (logging, request-id) et re-raise ce que tu attrapes.
  • HTTPException vs handler custom. FastAPI a déjà un handler pour HTTPException et RequestValidationError. Si tu veux un format unifié, remplace-les (@app.exception_handler(StarletteHTTPException)), sinon tu auras deux formats qui coexistent.
  • Timeouts en cascade. Ton endpoint a un timeout, ton client HTTP a un timeout, le SDK Anthropic a un timeout, ton load balancer a un timeout. S'ils ne sont pas ordonnés (LB > endpoint > LLM), tu auras des 504 du LB pendant que ton serveur travaille encore — gaspillage de tokens facturés pour une réponse jetée. Ordonne-les : timeout_LLM < timeout_endpoint < timeout_LB.

Performance

  • Un seul AsyncAnthropic, partagé. Créer un client par requête recrée le pool de connexions HTTP — coûteux et tu perds le keep-alive. Instancie-le une fois (au démarrage / via dépendance singleton).
  • Tout en async. Un client.messages.create() synchrone dans un endpoint async bloque l'event loop et fait écrouler le débit sous charge. Utilise AsyncAnthropic et await.
  • Streame dès que max_tokens est élevé. Au-delà de ~16K tokens de sortie, un appel non-streamé risque le timeout HTTP du SDK. Le streaming évite ça et améliore le time-to-first-token perçu.

Sécurité

  • Ne fuite jamais l'interne. str(exc) peut contenir des requêtes SQL, des chemins, des bouts de prompt système. Le client reçoit un code stable + un message neutre ; les détails vont dans les logs serveur uniquement, corrélés par incident_id/request_id.
  • N'écho jamais le payload d'entrée brut dans l'erreur. Sur une erreur de validation, renvoyer la valeur fautive peut refléter une injection ou exposer des données d'un autre champ.
  • Clé API jamais dans le code. os.environ["ANTHROPIC_API_KEY"], et une AuthenticationError du SDK déclenche une alerte ops — pas un message au client (qui n'y peut rien).

Observabilité

  • request_id de bout en bout. Génère/propage un x-request-id, mets-le dans chaque log et dans ErrorBody. Quand un client te dit « ça a planté », tu retrouves la trace en une recherche.
  • Logge usage même sur erreur partielle. Sur un refus mid-stream, tu es facturé pour les tokens déjà produits. Logge message.usage pour ne pas découvrir la facture en fin de mois.
  • Métriques par code. Compte les erreurs par code (llm_unavailable, quota_exceeded…). Un pic de 529/OverloadedError te dit qu'il faut un fallback de modèle ; un pic de validation_error pointe un client qui a cassé son intégration.

Les arbitrages senior

  • Granularité des exceptions. Trop d'exceptions = bureaucratie ; pas assez = tu reviens au except Exception. La bonne maille : une exception par décision client distincte. Si le client réagit pareil à deux erreurs, fusionne-les.
  • Retryable : qui décide ? Le SDK retente les 429/5xx upstream. Mais ton endpoint, lui, ne doit pas être retenté à l'aveugle par le client sur un 400. Le code HTTP que tu choisis est le signal de retryabilité — choisis-le avec ça en tête.
  • Fallback de modèle. Sur OverloadedError répété, un senior bascule sur un modèle moins chargé (ex. claude-haiku-4-5) plutôt que de faire échouer la requête. C'est un arbitrage coût/latence/qualité, pas un réflexe.

🏋️ Exercices

Exercice 1 — Centraliser (implémenter)

Objectif : prendre une API avec 4 endpoints qui font chacun leur try/except: return {"error": ...} et les refactorer pour qu'aucun endpoint ne contienne de try/except. Toutes les erreurs passent par une hiérarchie DomainError + un handler unique, avec un schéma ErrorBody Pydantic.

Indice/Solution : définis DomainError avec status_code/code en attributs de classe, une sous-classe par cas (NotFound, Conflict, …), un seul @app.exception_handler(DomainError). Vérifie qu'ajouter une 5ᵉ erreur ne demande de toucher aucun endpoint.

Exercice 2 — Aligner validation et 500 (implémenter)

Objectif : faire en sorte que toutes les sorties d'erreur de ton API — validation 422, HTTPException, DomainError, et 500 non géré — partagent exactement le même schéma JSON (code, message, request_id).

Indice/Solution : enregistre des handlers pour RequestValidationError, StarletteHTTPException, DomainError et Exception. Pour le 500, génère un incident_id, logge logger.exception(...), et ne renvoie que l'incident_id au client. Teste avec un endpoint qui fait volontairement 1 / 0.

Exercice 3 — Classer les erreurs LLM (rendre production-grade)

Objectif : écrire un endpoint /ask qui appelle Claude et traduit correctement chaque exception du SDK : RateLimitError/OverloadedError503 avec header retry-after propagé ; AuthenticationError500 + alerte log (rien d'utile au client) ; BadRequestError500 (c'est ton bug). Vérifie stop_reason == "refusal" avant de lire content.

Indice/Solution : un @app.exception_handler(anthropic.APIStatusError) avec des isinstance du plus spécifique au plus général. Propage exc.response.headers.get("retry-after"). Pour le refus, lève une LLMRefused(422) avant d'accéder à content[0].

Exercice 4 — Erreur en plein streaming (casser puis réparer)

Objectif : monter un endpoint SSE qui streame les tokens de Claude. Simule (1) une erreur avant le premier token et (2) une erreur après le premier token. Montre que le cas (1) peut renvoyer un 503, mais que le cas (2) doit émettre un event: error dans le flux. Côté « client », écris un petit consommateur qui distingue les deux.

Indice/Solution : dans le générateur, place le try/except autour du async with client.messages.stream(...). Si l'exception tombe avant le premier yield, tu peux lever pour obtenir un 503 ; après le premier yield, le status est figé — émets event: error. Pour simuler, lève une exception conditionnelle après N tokens.

Exercice 5 — Idempotence et conflits (production-grade, difficile)

Objectif : ajouter un endpoint POST /jobs qui crée un job de génération via l'agent. Gère le 409 Conflict proprement : si le même Idempotency-Key est rejoué pendant que le premier job tourne encore, renvoie un 409 typé (pas un doublon, pas un 500). Assure-toi que le timeout du LLM (< endpoint < LB) est cohérent.

Indice/Solution : un store dict[idempotency_key, status], lève JobInProgress(DomainError, status_code=409) si la clé est déjà running. Mets timeout= sur le client Anthropic en dessous du timeout de l'endpoint. Logge request_id + idempotency_key ensemble.

Exercice 6 — Fallback de modèle sous surcharge (architecture)

Objectif : quand OverloadedError survient après les retries du SDK, basculer automatiquement de claude-opus-4-8 vers claude-haiku-4-5 une seule fois, en loggant le fallback. Au-delà, renvoyer 503.

Indice/Solution : enveloppe l'appel dans une fonction call_with_fallback(models: list[str]) qui itère sur la liste, attrape OverloadedError, et ne renvoie LLMUnavailable que si tous les modèles échouent. Compte les fallbacks en métrique — un taux élevé signale un problème de capacité, pas un détail.


🎤 En entretien

Q : Pourquoi ne pas mettre un try/except Exception dans chaque endpoint ? Parce que ça duplique la logique de traduction, produit des formats d'erreur incohérents, et risque de fuiter des détails internes ou de renvoyer un 200 sur une erreur. La bonne approche est de lever des exceptions métier typées et de les traduire dans des exception handlers centralisés — une seule source de vérité pour le format et le status code.

Q : Quelle est la différence entre HTTPException et une exception métier custom ?HTTPException couple ta logique au HTTP — utile pour un endpoint trivial. Une exception métier (UserNotFound) ne connaît pas le HTTP : elle exprime un fait du domaine, et c'est la frontière (le handler) qui décide du code 404. Ça garde le code métier réutilisable et testable hors contexte web.

Q : Une erreur survient pendant que tu streames la réponse d'un agent — que se passe-t-il ? Le status 200 est déjà parti, on ne peut plus renvoyer un 503. L'erreur doit être émise dans le flux (un event: error en SSE) et le client doit savoir l'interpréter. Seule une erreur survenant avant le premier octet peut encore se traduire en status HTTP classique.

Q : Comment décides-tu si une erreur d'appel LLM doit être retentée ? Par le type d'exception du SDK, jamais par le message : RateLimitError, InternalServerError, OverloadedError sont retryables (le SDK le fait déjà avec backoff) ; BadRequestError, AuthenticationError, et un refus de safety (stop_reason == "refusal") ne le sont pas — retenter ne changera rien et brûlera des tokens. Le code HTTP que j'expose ensuite encode cette retryabilité pour mon propre client.

Bibliothèque tech perso — Achref