Aller au contenu
English

Configuration et journalisation

  • Leçon 17
  • 45 min
  • Module 6 : Tester et renforcer le service

Ce que vous saurez faire

Quand un utilisateur signale « J’ai reçu une erreur », trouvez cette requête précise dans vos journaux en quelques secondes. Exécutez aussi le même code avec différents paramètres : sur votre ordinateur, dans les tests et sur le serveur, sans modifier le code.

Avant de commencerCe qu’il faut savoir et avoir sous la mainEnviron 45 minutes

Ce que vous devez savoir

Ce qu’il vous faut sur votre ordinateur

Testé avec Python 3.14.6 et FastAPI 0.142.2 sur macOS.

L’API lit maintenant trois variables d’environnement dans main.py (ALLOWED_ORIGINS, SHOW_DOCS), à deux endroits différents. Le classificateur est fixé dans le code. Comme dans le cours de Python, les paramètres appartiennent à un seul module, avec des valeurs par défaut sûres :

ticket_api/config.py
import os
from dataclasses import dataclass
@dataclass(frozen=True)
class Settings:
allowed_origins: list[str]
classifier_mode: str
log_level: str
show_docs: bool
def load_settings() -> Settings:
"""Read the settings from environment variables, with safe defaults."""
origins = os.environ.get("ALLOWED_ORIGINS", "")
return Settings(
allowed_origins=[o.strip() for o in origins.split(",") if o.strip()],
classifier_mode=os.environ.get("CLASSIFIER_MODE", "keywords"),
log_level=os.environ.get("LOG_LEVEL", "INFO").upper(),
show_docs=os.environ.get("SHOW_DOCS", "true").lower() == "true",
)
Variable Valeur par défaut Signification
ALLOWED_ORIGINS aucune Pages Web qui peuvent appeler l’API depuis un navigateur (module 3).
CLASSIFIER_MODE keywords keywords (mots-clés) est le vrai classificateur. unavailable (indisponible) simule un classificateur en panne.
LOG_LEVEL INFO DEBUG (débogage), INFO (information), WARNING (avertissement) ou ERROR (erreur).
SHOW_DOCS true false (faux) désactive /docs, /redoc et /openapi.json (module 4).

CLASSIFIER_MODE=unavailable vous permet d’observer le comportement de l’API quand sa dépendance est en panne. Vous le faites sur votre ordinateur et, au module 7, sur le serveur. Ce mode utilise un petit classificateur qui échoue toujours, à la fin de classifier.py :

ticket_api/classifier.py (add at the end)
class UnavailableClassifier:
"""Simulates a classifier that is down."""
version = "unavailable"
def predict(self, text: str) -> Prediction:
raise ClassifierUnavailable("the classifier is not available")
def make_classifier(mode: str) -> KeywordClassifier | UnavailableClassifier:
classifiers = {"keywords": KeywordClassifier, "unavailable": UnavailableClassifier}
if mode not in classifiers:
raise ValueError(f"unknown CLASSIFIER_MODE {mode!r}")
return classifiers[mode]()

Un mode inconnu arrête l’application au démarrage, avec un message clair. C’est mieux que de démarrer avec le mauvais classificateur.

Jusqu’ici, main.py créait l’application quand Python l’importait, avec les paramètres du moment. Un test qui veut d’autres paramètres, comme classifier_mode="unavailable", ne peut pas les modifier. La solution est une fabrique : une fonction qui crée une nouvelle application à partir des paramètres que vous lui passez.

ticket_api/main.py (the structure)
def create_app(settings: Settings | None = None) -> FastAPI:
settings = settings or load_settings()
# ... make the app with these settings, add middleware, handlers and routes
return app
app = create_app()

fastapi dev ticket_api/main.py trouve toujours app, créée à partir des variables d’environnement. Les tests appellent create_app(...) avec leurs propres paramètres et obtiennent une nouvelle application chaque fois.

Quand un utilisateur signale une erreur, les journaux doivent montrer cette requête précise. L’API donne à chaque requête un court identifiant aléatoire. Elle le place dans l’en-tête de réponse X-Request-ID et dans chaque corps d’erreur, puis l’écrit dans sa ligne de journal. L’utilisateur copie l’identifiant de l’erreur, et vous le cherchez dans les journaux.

Un intergiciel s’exécute avant et après chaque requête, comme l’intergiciel CORS. Voici le nouveau main.py au complet :

ticket_api/main.py
import logging
import time
import uuid
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from ticket_api.classifier import make_classifier
from ticket_api.config import Settings, load_settings
from ticket_api.errors import add_error_handlers
from ticket_api.models import Health
from ticket_api.routes import router
logger = logging.getLogger("ticket_api")
DESCRIPTION = """
Classifies support tickets into a category and a priority.
The classifier is a deterministic mock: it uses keywords, so the same text
always gives the same result. Every error has the same shape:
`{"error": {"code": ..., "message": ..., "request_id": ..., "fields": [...]}}`.
"""
def create_app(settings: Settings | None = None) -> FastAPI:
settings = settings or load_settings()
logging.basicConfig(
level=settings.log_level, format="%(levelname)s %(name)s: %(message)s"
)
app = FastAPI(
title="Ticket Classifier API",
version="1.0.0",
description=DESCRIPTION,
openapi_tags=[
{"name": "tickets", "description": "Classify support tickets."},
{"name": "health", "description": "Check that the service runs."},
],
docs_url="/docs" if settings.show_docs else None,
redoc_url="/redoc" if settings.show_docs else None,
openapi_url="/openapi.json" if settings.show_docs else None,
)
app.state.settings = settings
app.state.classifier = make_classifier(settings.classifier_mode)
@app.middleware("http")
async def request_id_and_log(request: Request, call_next):
request.state.request_id = uuid.uuid4().hex[:12]
started = time.perf_counter()
response = await call_next(request)
response.headers["X-Request-ID"] = request.state.request_id
logger.info(
"%s %s %d %.0fms id=%s",
request.method,
request.url.path,
response.status_code,
(time.perf_counter() - started) * 1000,
request.state.request_id,
)
return response
if settings.allowed_origins:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_methods=["GET", "POST"],
allow_headers=["Content-Type"],
)
add_error_handlers(app)
app.include_router(router)
@app.get("/health", tags=["health"], summary="Check that the service runs")
def health() -> Health:
return Health(status="ok")
return app
app = create_app()
Partie Raison
uuid.uuid4().hex[:12] Un identifiant aléatoire, comme 5f5cb5124f1a. 12 caractères suffisent pour qu’il soit unique dans vos journaux.
request.state.request_id Conserve l’identifiant avec cette requête pour que les gestionnaires d’erreurs puissent l’utiliser.
await call_next(request) Exécute le reste de l’application : la route et les gestionnaires. L’intergiciel reçoit la réponse.
La ligne de journal Méthode, chemin, statut, durée et identifiant. Pas le corps, ni les en-têtes.

L’intergiciel journalise seulement des faits sur la requête, jamais son contenu. Le corps d’un billet de soutien peut contenir des données personnelles d’un client, et un en-tête peut contenir une clé d’API. Les journaux sont conservés longtemps et lus par beaucoup de gens. Ils ne doivent donc contenir ni l’un ni l’autre.

La structure d’erreur reçoit un champ de plus, dans models.py :

ticket_api/models.py (in ErrorDetail)
request_id: str | None = Field(
default=None, description="Give this ID when you report a problem."
)

Dans errors.py, error_response reçoit la requête comme premier paramètre et y lit l’identifiant. Chaque gestionnaire passe request :

ticket_api/errors.py (changes)
def error_response(
request: Request, status: int, code: str, message: str, fields=()
) -> JSONResponse:
detail = ErrorDetail(
code=code,
message=message,
request_id=getattr(request.state, "request_id", None),
fields=list(fields),
)
return JSONResponse(
status_code=status, content=ErrorResponse(error=detail).model_dump()
)

Chaque appel passe de error_response(422, ...) à error_response(request, 422, ...). Les deux lignes de journal des gestionnaires reçoivent aussi l’identifiant : logger.warning("classifier unavailable: %s id=%s", exc, request.state.request_id) et logger.exception("unexpected error id=%s", request.state.request_id).

Un fichier nommé conftest.py dans tests contient des fixtures (fonctions de préparation) utilisables par chaque fichier de tests. Il donne à chaque test une nouvelle application avec des paramètres de test, et un moyen de modifier un paramètre :

tests/conftest.py
from dataclasses import replace
import pytest
from fastapi.testclient import TestClient
from ticket_api.config import Settings
from ticket_api.main import create_app
TEST_SETTINGS = Settings(
allowed_origins=[],
classifier_mode="keywords",
log_level="WARNING",
show_docs=True,
)
@pytest.fixture
def make_client():
"""Return a function that makes a test client with changed settings."""
def make(**changes) -> TestClient:
app = create_app(replace(TEST_SETTINGS, **changes))
return TestClient(app, raise_server_exceptions=False)
return make
@pytest.fixture
def client(make_client) -> TestClient:
return make_client()

replace (de dataclasses) copie les paramètres figés en modifiant certains champs. make_client renvoie une fonction, ce qui permet à un test de demander les paramètres dont il a besoin. La plupart des tests demandent seulement client.

Les tests changent de trois petites façons :

  1. Retirez client = TestClient(app) et ses importations des deux fichiers de tests. Donnez plutôt un paramètre client à chaque fonction de test : def test_health(client):.

  2. Remplacez la fixture du classificateur en panne par le paramètre, ce qui est plus simple :

    tests/test_classify.py
    def test_unavailable_classifier_gives_503(make_client):
    client = make_client(classifier_mode="unavailable")
    response = client.post("/v1/classify", json={"subject": "Refund"})
    assert response.status_code == 503
    assert response.json()["error"]["code"] == "classifier_unavailable"
    assert "Traceback" not in response.text
    def test_every_response_has_a_request_id(client):
    ok = client.get("/health")
    error = client.post("/v1/classify", json={"subject": ""})
    assert len(ok.headers["X-Request-ID"]) == 12
    assert error.json()["error"]["request_id"] == error.headers["X-Request-ID"]
  3. Dans export_contract.py, importez create_app au lieu de app, et utilisez create_app().openapi().

Amira exécute les tests. Le test du contrat échoue. La différence montre le nouveau champ : "request_id": ... dans la description et dans le schéma ErrorDetail. Un nouveau champ facultatif dans une erreur ne rompt pas la compatibilité. Elle exécute donc python -m ticket_api.export_contract, et les 13 tests réussissent.

Elle envoie ensuite un billet invalide à l’API en cours d’exécution :

Response body
{"error":{"code":"invalid_request","message":"The request is not valid.","request_id":"5f5cb5124f1a","fields":["subject"]}}
API terminal
INFO ticket_api: POST /v1/classify 422 1ms id=5f5cb5124f1a

L’identifiant dans la réponse et celui dans le journal sont identiques. Avec CLASSIFIER_MODE=unavailable, le journal indique aussi pourquoi :

API terminal
WARNING ticket_api: classifier unavailable: the classifier is not available id=073a3ced6d65
INFO ticket_api: POST /v1/classify 503 2ms id=073a3ced6d65

(Uvicorn écrit aussi sa propre ligne pour chaque requête, qui commence par INFO: 127.0.0.1:.... Vos lignes commencent par INFO ticket_api:.)

Exercice guidé : paramètres et identifiants de requête

Section intitulée « Exercice guidé : paramètres et identifiants de requête »
  1. Créez config.py, ajoutez UnavailableClassifier et make_classifier à classifier.py, puis remplacez main.py.
  2. Ajoutez request_id à ErrorDetail et modifiez errors.py.
  3. Créez tests/conftest.py, puis modifiez les tests et export_contract.py.
  4. Exécutez les tests. Lisez la différence du contrat, exportez le contrat, puis exécutez les tests de nouveau. Tous réussissent.
  5. Envoyez une requête valide et une requête invalide à l’API en cours d’exécution. Trouvez l’identifiant de chacune dans la réponse (en-tête ou corps) et dans le terminal de l’API.
  6. Arrêtez l’API, puis démarrez-la avec CLASSIFIER_MODE=unavailable ($env:CLASSIFIER_MODE = "unavailable" dans PowerShell). Envoyez un billet. Vous obtenez 503, et le journal contient un WARNING avec le même identifiant. Redémarrez-la sans la variable.
  7. Démarrez-la avec CLASSIFIER_MODE=magic. Elle s’arrête immédiatement avec ValueError: unknown CLASSIFIER_MODE 'magic' (valeur de CLASSIFIER_MODE inconnue : « magic »).

Un utilisateur écrit : « La classification a échoué à 14 h 02, l’erreur indiquait request_id 9b1c0e7a44d2 ». Notez les étapes exactes que vous suivriez pour comprendre ce qui s’est passé, avec vos journaux actuels. Nommez ensuite une information que les journaux ne donnent pas, et dites si son ajout serait sûr.

Afficher une réponse

Cherchez 9b1c0e7a44d2 dans les journaux de l’API. La ligne INFO donne la méthode, le chemin, le code de statut et la durée. Pour un 503, une ligne WARNING avec le même identifiant indique que le classificateur était indisponible. Pour un 500, une ligne ERROR contient la trace complète de l’erreur.

Les journaux ne contiennent pas le texte du billet. Vous ne pouvez donc pas reproduire la requête exacte. Le journaliser n’est pas sûr par défaut : les billets peuvent contenir des données personnelles. Au besoin, demandez à l’utilisateur d’envoyer le texte. Sinon, journalisez seulement sa longueur ou une catégorie, jamais le contenu.

Erreur courante

Un secret dans les journaux

Symptôme : un collègue lit les journaux pour trouver un bogue et y voit des clés d’API ou des messages de clients.

Cause : quelqu’un a ajouté une ligne de journal contenant toute la requête pour déboguer un problème, puis l’a laissée. Par exemple : logger.info("request %s %s", request.headers, body). Les journaux sont copiés vers des services de journalisation, conservés pendant des mois et lus par beaucoup de gens. Chaque secret qu’ils contiennent a été divulgué.

Solution :

  1. Retirez la ligne de journal. Journalisez seulement des faits : méthode, chemin, statut, durée, identifiant.
  2. Traitez chaque clé apparue dans les journaux comme divulguée : révoquez-la et créez-en une nouvelle.
  3. Si vous avez besoin de plus de détails pour déboguer, utilisez une ligne DEBUG avec seulement des champs sûrs. Retirez-la après la correction.

Vérifiez votre compréhension

Pourquoi create_app() reçoit-elle les paramètres en argument, au lieu de toujours appeler load_settings()?

Afficher la réponse

Pour que chaque appelant puisse décider d’où viennent les paramètres. Le serveur utilise les variables d’environnement (create_app() sans argument). Un test passe ses propres paramètres, par exemple classifier_mode="unavailable", sans définir de variables d’environnement qui affecteraient d’autres tests. Chaque test reçoit une nouvelle application. Aucun état ne passe donc d’un test au suivant.

Récapitulatif

  • Lisez tous les paramètres dans config.py, avec des valeurs par défaut sûres. Une valeur inconnue arrête l’application au démarrage.
  • create_app(settings) crée une nouvelle application. Le serveur et les tests passent chacun leurs propres paramètres.
  • Un intergiciel donne un identifiant à chaque requête. Il le place dans l’en-tête X-Request-ID et le corps d’erreur, puis le journalise.
  • Journalisez des faits sur les requêtes, jamais leur contenu, leurs en-têtes ou leurs secrets.
  • conftest.py contient des fixtures partagées, comme un client avec des paramètres modifiés.

Ensuite, vous protégez l’API : une clé d’API, une limite de taille du corps et une limite de durée pour le classificateur.

Leçon suivante : Protection de base