EnglishExposer un modèle : FastAPI, Pydantic, Jinja
Un modèle entraîné dans un notebook ne sert à personne : il faut que d'autres programmes — ou d'autres humains — puissent lui poser des questions. Ce module construit ce pont : une API HTTP avec FastAPI, des données d'entrée validées avec Pydantic, et une petite page web générée avec Jinja pour donner une interface à ceux qui ne parlent pas JSON.
📋 Au programme
1. Pourquoi une API
Trois façons de rendre un modèle utilisable. Le script batch : on lit un fichier, on écrit les prédictions dans un autre. C'est simple, robuste, et suffisant dans une majorité de cas — personne n'a besoin d'un score de churn à la milliseconde. La bibliothèque : on publie un paquet Python que les autres importent ; élégant, mais cela impose à tous le même langage et le même environnement. L'API : le modèle tourne quelque part, et n'importe quel programme lui parle par le réseau, quel que soit son langage.
L'API s'impose quand la prédiction doit être fraîche (au moment de l'action, pas la veille au soir), quand plusieurs applications hétérogènes doivent y accéder, ou quand on veut découpler le cycle de vie du modèle de celui des applications qui l'utilisent — remplacer le modèle sans redéployer le site web qui l'appelle.
2. Le minimum vital sur HTTP
Une API web repose sur un échange requête-réponse. Le client envoie une requête : une méthode, un chemin, des en-têtes, parfois un corps. Le serveur renvoie une réponse : un code de statut, des en-têtes, un corps.
MÉTHODES
GET lire une ressource, sans effet de bord. Paramètres dans l'URL.
POST envoyer des données, créer, déclencher un calcul. Données dans le CORPS.
PUT remplacer une ressource DELETE supprimer PATCH modifier
CODES DE STATUT
2xx succès 200 OK, 201 Créé, 204 Pas de contenu
4xx faute du CLIENT 400 requête malformée 401 non authentifié
403 interdit 404 introuvable
422 données invalides 429 trop de requêtes
5xx faute du SERVEUR 500 erreur interne 503 indisponible
FORMAT JSON dans la quasi-totalité des cas : {"surface": 72.5, "pieces": 3}
En-tête Content-Type: application/jsonPour une prédiction, on utilise POST plutôt que GET : les paramètres sont nombreux et structurés, ce qui tient mal dans une URL, et le corps de requête permet des types imbriqués. On réserve GET aux lectures simples, comme un /health.
3. Pydantic : valider ce qui entre
Le problème
Dès qu'une donnée vient de l'extérieur — une requête HTTP, un fichier de configuration, une réponse d'API tierce — on ne contrôle plus rien. Un client peut envoyer une surface sous forme de chaîne de caractères, oublier un champ, ajouter un champ inconnu, ou envoyer une valeur absurde comme une surface de -40 m². Sans validation, ces valeurs traversent le code et finissent dans le modèle, qui produit alors une prédiction absurde — ou une exception incompréhensible dix fonctions plus loin.
Écrire cette validation à la main donne une cinquantaine de lignes de if imbriqués par point d'entrée, qu'il faut maintenir en double avec la documentation. C'est exactement ce que Pydantic supprime.
Le principe
On décrit la forme attendue des données comme une classe annotée — très proche d'une dataclass (module 1) — et Pydantic s'occupe du reste : il vérifie les types à l'exécution, convertit ce qui peut l'être (la chaîne "72.5" devient le flottant 72.5), refuse le reste avec un message précis, et fournit une description exploitable par les outils.
def predire(donnees: dict):
if "surface" not in donnees:
raise ValueError("champ 'surface' manquant")
try:
surface = float(donnees["surface"])
except (TypeError, ValueError):
raise ValueError("'surface' doit être un nombre")
if surface <= 0 or surface > 10_000:
raise ValueError("'surface' hors bornes")
# … et la même chose pour chacun des douze autres champsfrom typing import Literal
from pydantic import BaseModel, Field, field_validator
class Annonce(BaseModel):
surface: float = Field(gt=0, le=10_000, description="surface habitable en m²")
n_pieces: int = Field(ge=1, le=20)
code_postal: str = Field(pattern=r"^\d{5}$")
type_bien: Literal["appartement", "maison"] # valeurs autorisées
etage: int | None = None # champ facultatif
ascenseur: bool = False # avec valeur par défaut
@field_validator("code_postal")
@classmethod
def departement_couvert(cls, v: str) -> str: # règle MÉTIER, pas de type
if not v.startswith(("38", "69", "73", "74")):
raise ValueError("département non couvert par le modèle")
return v
a = Annonce(surface="72.5", n_pieces=3, code_postal="38000", type_bien="appartement")
print(a.surface, type(a.surface)) # 72.5 <class 'float'> → converti automatiquementCe qui se passe à l'instanciation : chaque champ est vérifié, converti si possible, et une erreur groupée est levée si quelque chose ne va pas. Le message est directement exploitable :
Annonce(surface=-5, n_pieces=0, code_postal="7500", type_bien="chalet")
# ValidationError: 4 validation errors for Annonce
# surface Input should be greater than 0 [input_value=-5]
# n_pieces Input should be greater than or equal to 1 [input_value=0]
# code_postal String should match pattern '^\d{5}$' [input_value='7500']
# type_bien Input should be 'appartement' or 'maison' [input_value='chalet']Les quatre erreurs sont signalées d'un coup, avec le champ fautif et la valeur reçue — au lieu d'un KeyError anonyme au milieu du traitement.
from pydantic import BaseModel, Field, ConfigDict, model_validator
from datetime import date
class AnnonceStricte(BaseModel):
model_config = ConfigDict(
extra="forbid", # REFUSE les champs inconnus (défaut : les ignorer)
frozen=True, # immuable, comme une dataclass frozen (module 1)
str_strip_whitespace=True,
)
ville: str
prix: float = Field(gt=0)
date_publication: date # "2026-03-14" → objet date automatiquement
photos: list[str] = Field(default_factory=list)
@model_validator(mode="after") # validation qui porte sur PLUSIEURS champs
def coherence(self):
if self.prix > 5_000_000 and not self.photos:
raise ValueError("un bien à ce prix sans photo est suspect")
return self
# sérialisation dans les deux sens
a = AnnonceStricte.model_validate(dictionnaire) # depuis un dict
a = AnnonceStricte.model_validate_json(chaine_json) # depuis du JSON brut
a.model_dump() # → dict
a.model_dump_json() # → chaîne JSONPydantic ou dataclass ?
La règle est nette. Dataclass pour les objets internes, dont on maîtrise la construction : aucune vérification n'est nécessaire, et le coût d'exécution est nul. Pydantic à chaque frontière avec l'extérieur : requête HTTP, fichier de configuration, réponse d'API, ligne scrapée. La validation a un coût — quelques microsecondes par objet, le cœur de Pydantic v2 étant écrit en Rust — parfaitement acceptable pour une requête, mais qu'on ne veut pas payer un million de fois dans une boucle interne.
Cas particulier utile : pydantic-settings lit la configuration depuis les variables d'environnement en la validant, ce qui remplace avantageusement les os.environ.get éparpillés — et fait échouer le démarrage immédiatement si une variable manque, plutôt qu'au premier appel.
4. FastAPI : premier serveur
FastAPI relie tout cela : les routes sont des fonctions Python décorées (module 1), les entrées et sorties sont des modèles Pydantic, et la documentation interactive est générée automatiquement à partir des types.
from fastapi import FastAPI
app = FastAPI(title="API de scoring immobilier", version="1.0.0")
@app.get("/health") # décorateur : "cette fonction répond au GET /health"
def health() -> dict:
return {"status": "ok"} # le dict est converti en JSON automatiquementuv run uvicorn mon_projet.api:app --reload # --reload : redémarre à chaque sauvegarde # └─ module:objet # → http://127.0.0.1:8000/health # → http://127.0.0.1:8000/docs : documentation interactive, générée toute seule
La page /docs mérite qu'on s'y arrête : elle liste les routes, les schémas d'entrée et de sortie, et permet d'envoyer une requête de test depuis le navigateur. Elle est produite à partir des annotations de type, donc elle ne peut pas se désynchroniser du code — contrairement à une documentation écrite à la main.
from fastapi import FastAPI, Query, Path, HTTPException
@app.get("/annonces/{annonce_id}") # 1. dans le CHEMIN
def lire(annonce_id: int = Path(ge=1)) -> dict:
annonce = base.get(annonce_id)
if annonce is None:
raise HTTPException(status_code=404, detail="annonce introuvable")
return annonce
@app.get("/annonces") # 2. en QUERY : /annonces?ville=Lyon
def chercher(ville: str, prix_max: float = Query(default=1e9, gt=0),
limite: int = Query(default=20, le=100)) -> list[dict]:
...
@app.post("/predict") # 3. dans le CORPS de la requête
def predire(annonce: Annonce) -> Reponse: # ← modèle Pydantic
...Le troisième cas est le plus intéressant : FastAPI voit que le paramètre est un modèle Pydantic, en déduit qu'il faut le lire dans le corps de la requête, le valide, et renvoie automatiquement un code 422 avec le détail des erreurs si les données ne conviennent pas. Aucune ligne de validation à écrire.
5. Servir un modèle de ML
from contextlib import asynccontextmanager
import anyio, joblib, numpy as np, pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
MODELE_VERSION = "1.4.2"
etat: dict = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# ── exécuté UNE FOIS au démarrage ──
etat["modele"] = joblib.load(f"models/gbm-{MODELE_VERSION}.joblib")
yield
# ── exécuté à l'arrêt ──
etat.clear()
app = FastAPI(title="API de scoring immobilier", version=MODELE_VERSION, lifespan=lifespan)
class Annonce(BaseModel):
surface: float = Field(gt=0, le=10_000)
n_pieces: int = Field(ge=1, le=20)
code_postal: str = Field(pattern=r"^\d{5}$")
type_bien: str
class Prediction(BaseModel):
prix_estime: float
intervalle_bas: float
intervalle_haut: float
version_modele: str
@app.get("/health")
def health() -> dict:
return {"status": "ok", "modele_charge": "modele" in etat, "version": MODELE_VERSION}
@app.post("/predict", response_model=Prediction)
async def predire(annonce: Annonce) -> Prediction:
X = pd.DataFrame([annonce.model_dump()]) # le pipeline attend un DataFrame
try:
# predict est BLOQUANT : on le sort de la boucle d'événements (module 2)
y = await anyio.to_thread.run_sync(etat["modele"].predict, X)
except Exception as exc:
raise HTTPException(status_code=500, detail="échec de l'inférence") from exc
prix = float(y[0])
return Prediction(prix_estime=prix,
intervalle_bas=prix * 0.9, intervalle_haut=prix * 1.1,
version_modele=MODELE_VERSION)
@app.post("/predict/batch", response_model=list[Prediction])
async def predire_lot(annonces: list[Annonce]) -> list[Prediction]:
"""Un seul appel pour N annonces : bien plus efficace que N appels."""
X = pd.DataFrame([a.model_dump() for a in annonces])
y = await anyio.to_thread.run_sync(etat["modele"].predict, X)
return [Prediction(prix_estime=float(p), intervalle_bas=float(p) * 0.9,
intervalle_haut=float(p) * 1.1, version_modele=MODELE_VERSION)
for p in y]Les cinq points qui reviennent systématiquement en production
- Charger le modèle au démarrage, via
lifespan. Le charger dans la fonction de prédiction ajouterait des centaines de millisecondes à chaque appel, et multiplierait la mémoire utilisée. - Sortir le calcul de la boucle d'événements. Un
predictscikit-learn est bloquant ; appelé directement dans une coroutine, il gèle tout le serveur pendant son exécution (module 2).anyio.to_thread.run_syncle délègue à un thread. - Valider strictement les entrées. Les contraintes Pydantic rejettent les valeurs aberrantes avant qu'elles n'atteignent le modèle : c'est la première ligne de défense contre les prédictions absurdes sur des données hors domaine.
- Renvoyer la version du modèle. Sans cela, impossible de relier a posteriori une prédiction contestée à l'artefact qui l'a produite (module 26).
- Prévoir une route par lot. Cent appels unitaires paient cent fois le coût du réseau et de la conversion ; un appel pour cent annonces est souvent dix fois plus rapide.
6. Jinja : une interface web
À quoi sert un moteur de templates
Une API renvoie du JSON, parfait pour un programme, illisible pour un humain. Pour donner un petit formulaire à un collègue non technicien, il faut produire du HTML — et le produire par concaténation de chaînes en Python devient vite ingérable et dangereux (une valeur contenant du code HTML casserait la page, voire injecterait un script).
Un moteur de templates sépare la structure de la page et les données. On écrit un fichier HTML normal, dans lequel on place des marqueurs ; le moteur remplit ces marqueurs avec les valeurs fournies. Jinja est le moteur de référence en Python — celui de Flask, et il s'utilise aussi bien avec FastAPI.
{{ expression }} affiche une valeur → {{ annonce.prix }}
{% instruction %} logique → {% if %} {% for %} {% block %}
{# commentaire #} n'apparaît pas dans la page finale
filtres : une valeur passée dans une transformation
{{ prix | round(2) }} {{ nom | upper }}
{{ liste | length }} {{ texte | default("inconnu") }}<!DOCTYPE html>
<html lang="fr">
<head><meta charset="utf-8"><title>Estimation de prix</title></head>
<body>
<h1>Estimer le prix d'un bien</h1>
<form method="post" action="/estimer">
<label>Surface (m²) <input type="number" name="surface" step="0.1" required></label>
<label>Pièces <input type="number" name="n_pieces" min="1" required></label>
<label>Code postal <input type="text" name="code_postal" pattern="\d{5}" required></label>
<button type="submit">Estimer</button>
</form>
{# ce bloc ne s'affiche que si une estimation a été calculée #}
{% if prix %}
<p>Prix estimé : <strong>{{ prix | round(0) }} €</strong>
({{ (prix / surface) | round(0) }} €/m²)</p>
{% endif %}
{% if annonces_comparables %}
<h2>Biens comparables</h2>
<ul>
{% for a in annonces_comparables %}
<li>{{ a.ville }} — {{ a.surface }} m² — {{ a.prix | round(0) }} €</li>
{% else %}
<li>aucun bien comparable trouvé</li>
{% endfor %}
</ul>
{% endif %}
</body></html>from fastapi import Form, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
templates = Jinja2Templates(directory="templates")
@app.get("/", response_class=HTMLResponse)
def formulaire(request: Request):
return templates.TemplateResponse("formulaire.html", {"request": request})
@app.post("/estimer", response_class=HTMLResponse)
async def estimer(request: Request,
surface: float = Form(...), n_pieces: int = Form(...),
code_postal: str = Form(...)):
annonce = Annonce(surface=surface, n_pieces=n_pieces,
code_postal=code_postal, type_bien="appartement")
X = pd.DataFrame([annonce.model_dump()])
prix = float((await anyio.to_thread.run_sync(etat["modele"].predict, X))[0])
# les clés du dictionnaire deviennent les variables disponibles dans le template
return templates.TemplateResponse("formulaire.html", {
"request": request, "prix": prix, "surface": surface,
"annonces_comparables": chercher_comparables(annonce),
})Deux mécanismes de Jinja méritent d'être connus. L'échappement automatique : une valeur contenant <script> est affichée comme du texte et non exécutée, ce qui bloque par défaut les injections de code. Et l'héritage de templates : un fichier base.html définit la structure commune avec des {% block contenu %}, et chaque page fait {% extends "base.html" %} puis remplit les blocs — exactement le principe qui évite de dupliquer l'en-tête et le pied de page sur toutes les pages.
7. Tester une API
FastAPI fournit un client de test qui appelle l'application en mémoire, sans démarrer de serveur ni ouvrir de port : les tests sont donc rapides et utilisables en intégration continue.
from fastapi.testclient import TestClient
from mon_projet.api import app
client = TestClient(app) # déclenche aussi le lifespan (chargement du modèle)
def test_health():
r = client.get("/health")
assert r.status_code == 200 and r.json()["modele_charge"] is True
def test_prediction_valide():
r = client.post("/predict", json={"surface": 72.5, "n_pieces": 3,
"code_postal": "38000", "type_bien": "appartement"})
assert r.status_code == 200
corps = r.json()
assert corps["prix_estime"] > 0
assert corps["version_modele"] == "1.4.2"
def test_entree_invalide_renvoie_422():
r = client.post("/predict", json={"surface": -5, "n_pieces": 3,
"code_postal": "38000", "type_bien": "appartement"})
assert r.status_code == 422
assert "surface" in r.text # le champ fautif est nommé dans la réponse
def test_champ_manquant():
r = client.post("/predict", json={"surface": 72.5})
assert r.status_code == 4228. Mise en production
# en développement uv run uvicorn mon_projet.api:app --reload # en production : plusieurs processus workers, chacun avec son propre GIL (module 2) uv run uvicorn mon_projet.api:app --host 0.0.0.0 --port 8000 --workers 4 # dans un conteneur, la commande du Dockerfile (module 5) CMD ["uvicorn", "mon_projet.api:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
Chaque worker charge sa propre copie du modèle : quatre workers avec un modèle de 2 Go consomment 8 Go de RAM. C'est l'arbitrage principal à faire, entre débit et mémoire. Au-delà, on regroupe les requêtes par lots côté serveur (dynamic batching), ce que font Triton ou vLLM (modules 22 et 25).
from fastapi.middleware.cors import CORSMiddleware
from fastapi import Depends, Header, HTTPException
import logging, time, uuid
# 1. CORS : quels sites web ont le droit d'appeler cette API depuis un navigateur
app.add_middleware(CORSMiddleware, allow_origins=["https://mon-site.fr"],
allow_methods=["POST", "GET"], allow_headers=["*"])
# 2. authentification simple par clé d'API
def verifier_cle(x_api_key: str = Header(...)):
if x_api_key != os.environ["API_KEY"]:
raise HTTPException(status_code=401, detail="clé invalide")
@app.post("/predict", dependencies=[Depends(verifier_cle)])
async def predire(...): ...
# 3. journalisation : identifiant de requête, durée, version du modèle
@app.middleware("http")
async def journaliser(request, call_next):
rid, debut = str(uuid.uuid4())[:8], time.perf_counter()
reponse = await call_next(request)
logging.info("%s %s %s %.1fms", rid, request.url.path,
reponse.status_code, (time.perf_counter() - debut) * 1000)
reponse.headers["X-Request-ID"] = rid
return reponseS'y ajoutent, selon le contexte : une limitation de débit (rate limiting) pour éviter qu'un client ne sature le service, et l'exposition de métriques au format Prometheus pour le suivi (module 26). Le Depends utilisé ci-dessus est le système d'injection de dépendances de FastAPI : une fonction déclarée comme dépendance est exécutée avant la route, et son résultat est fourni à celle-ci — c'est le mécanisme standard pour l'authentification, l'ouverture d'une session de base de données ou la lecture de la configuration.
Ce qu'il faut retenir du module
- Une API se justifie quand la prédiction doit être fraîche ou partagée entre applications ; sinon, un batch suffit souvent.
- Pydantic valide et convertit les données à l'exécution : c'est la frontière entre le monde extérieur et le code.
- Dataclass à l'intérieur, Pydantic aux frontières.
- FastAPI déduit la documentation et la validation des annotations de type ;
/docsne peut pas se désynchroniser. - Modèle chargé une fois au démarrage, inférence déportée dans un thread, version renvoyée dans la réponse, route par lot prévue.
- Jinja sépare structure et données, échappe automatiquement le HTML et permet l'héritage de gabarits.
- Tester avec
TestClient, y compris les entrées invalides qui doivent renvoyer 422. - En production : plusieurs workers (attention à la mémoire), CORS, authentification, journalisation avec identifiant de requête.