Un modèle entraîné qui dort dans un fichier joblib n’a aucune valeur opérationnelle. Pour qu’il serve, il faut l’exposer derrière une API que les applications consommatrices appellent — site web, mobile, microservice métier. FastAPI s’est imposé en quelques années comme la solution de référence pour cela en Python : performances proches de Node.js, génération automatique de la documentation OpenAPI, validation des entrées et sorties via Pydantic, et un code remarquablement concis. Ce tutoriel construit pas-à-pas une API qui sert un modèle scikit-learn, depuis la première route jusqu’à un déploiement Docker sécurisé.
Pour le contexte général, voir le guide principal sur la stack data 2026. On suppose qu’un modèle est déjà entraîné et sérialisé en joblib — voir les tutoriels sur le feature engineering et la classification.
Prérequis
- Python 3.10+, fastapi, uvicorn, pydantic 2.x, joblib, scikit-learn 1.8+
- Connaissance basique d’HTTP (GET, POST, JSON)
- Un modèle sauvegardé en .joblib
- Docker installé (pour la dernière étape)
- Temps estimé : 2 heures
Étape 1 — Installer FastAPI et un serveur ASGI
FastAPI est le framework, mais il a besoin d’un serveur ASGI pour tourner. Uvicorn est le choix par défaut, basé sur uvloop. Pour la production, on placera Gunicorn devant Uvicorn pour gérer plusieurs workers, mais en développement Uvicorn seul suffit largement.
pip install "fastapi[standard]" joblib scikit-learn pandas
fastapi --version
Le tag [standard] installe Uvicorn, Pydantic et les dépendances usuelles en un coup. Depuis la version 0.110, FastAPI fournit aussi une CLI fastapi qui simplifie le démarrage. Si l’installation échoue sur Pydantic, c’est probablement un conflit avec une version 1.x déjà présente : créer un environnement virtuel propre règle le problème.
Étape 2 — Première API minimale
Avant d’ajouter le modèle, on construit une API « hello world » pour valider l’installation et comprendre la structure. Un fichier Python avec une instance FastAPI et une route GET suffit. C’est aussi le bon moment pour vérifier que la documentation OpenAPI est accessible automatiquement à /docs — c’est l’un des arguments décisifs du framework.
# app.py
from fastapi import FastAPI
app = FastAPI(title="API ML — démarrage")
@app.get("/health")
def health():
return {"status": "ok"}
On démarre avec fastapi dev app.py. La sortie affiche l’URL http://127.0.0.1:8000. Visiter http://127.0.0.1:8000/docs ouvre Swagger UI avec la route health documentée automatiquement, prête à être testée depuis le navigateur. Cette documentation vivante évite la dérive classique entre code et documentation manuelle.
Étape 3 — Charger le modèle au démarrage
Le modèle ne doit être chargé qu’une seule fois, au démarrage de l’application — pas à chaque requête, sous peine de latence catastrophique. FastAPI propose un événement lifespan dédié pour les ressources globales : on y charge le modèle, on le stocke dans l’état de l’application, et on le libère proprement à l’arrêt si nécessaire.
from contextlib import asynccontextmanager
from fastapi import FastAPI
import joblib
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.model = joblib.load("model_paiement.joblib")
yield
app.state.model = None
app = FastAPI(title="API ML — paiement", lifespan=lifespan)
Tout ce qui précède le yield s’exécute au démarrage, tout ce qui suit s’exécute à l’arrêt. C’est le bon endroit pour ouvrir et fermer une connexion à une base de données, un client S3, ou tout autre client à durée de vie longue. Le modèle est désormais accessible depuis n’importe quelle route via request.app.state.model.
Étape 4 — Définir le schéma d’entrée avec Pydantic
Pydantic valide automatiquement les entrées contre un schéma typé. Définir une classe Pydantic pour les features attendues garantit que toute requête mal formée est rejetée avant d’atteindre le modèle. C’est aussi cette classe qui alimente la documentation OpenAPI générée automatiquement, ce qui évite de la maintenir séparément.
from pydantic import BaseModel, Field
from typing import Literal
class Transaction(BaseModel):
montant: float = Field(gt=0, description="Montant de la commande, > 0")
nb_articles: int = Field(ge=1, le=500, description="Nombre d'articles")
canal_paiement: Literal["carte", "especes", "mobile", "virement"]
age: int = Field(ge=18, le=110)
code_postal: str = Field(min_length=5, max_length=5)
class Prediction(BaseModel):
classe: int
probabilite_positive: float
seuil_utilise: float
Les contraintes gt, ge, le et Literal sont vérifiées par Pydantic à la réception. Une requête avec montant=-5 ou canal_paiement="bitcoin" est rejetée avec un code 422 et un message explicite, sans atteindre la fonction métier. C’est une couche de sécurité gratuite et précieuse — le modèle ne voit que des entrées valides.
Étape 5 — Implémenter la route de prédiction
La route POST reçoit un objet Transaction, le convertit en DataFrame d’une seule ligne, appelle le modèle, et retourne la prédiction. Le seuil de décision est volontairement explicite — c’est lui qu’on ajustera selon les besoins métier sans réentraîner le modèle.
import pandas as pd
from fastapi import Request, HTTPException
SEUIL = 0.40 # ajusté en regardant la courbe précision-rappel
@app.post("/predire", response_model=Prediction)
def predire(transaction: Transaction, request: Request):
model = request.app.state.model
if model is None:
raise HTTPException(status_code=503, detail="Modèle indisponible")
df = pd.DataFrame([transaction.model_dump()])
proba = float(model.predict_proba(df)[0, 1])
classe = int(proba >= SEUIL)
return Prediction(classe=classe, probabilite_positive=round(proba, 4), seuil_utilise=SEUIL)
La méthode model_dump() de Pydantic 2.x convertit l’objet en dict — équivalent à l’ancien .dict(). Le DataFrame d’une seule ligne se construit en passant ce dict à pd.DataFrame dans une liste. Le pipeline scikit-learn complet (preprocessing + modèle) gère le reste : encodage des catégorielles, normalisation, prédiction.
Étape 6 — Tester l’API avec curl et httpie
L’API est désormais fonctionnelle. On la teste depuis la ligne de commande pour s’assurer qu’elle répond correctement à des requêtes valides et invalides. La documentation OpenAPI à /docs permet aussi de tester depuis le navigateur, mais la ligne de commande est plus rapide et scriptable.
curl -X POST http://127.0.0.1:8000/predire \
-H "Content-Type: application/json" \
-d '{"montant": 12500, "nb_articles": 3, "canal_paiement": "mobile",
"age": 32, "code_postal": "12345"}'
# Avec httpie (plus lisible)
http POST :8000/predire montant=12500 nb_articles:=3 \
canal_paiement=mobile age:=32 code_postal=12345
La réponse JSON renvoie classe, probabilité et seuil. Avec une requête volontairement invalide (champ manquant, valeur hors borne), on doit obtenir un code 422 et un message qui pointe précisément le champ fautif. Cette boucle de tests rapides confirme que la couche Pydantic fait son travail avant de se concentrer sur le métier.
Étape 7 — Ajouter les batchs
Une API qui n’accepte qu’une transaction à la fois est inadaptée à des intégrations qui doivent scorer mille lignes en une requête. Une route batch acceptant une liste règle ce problème efficacement, tout en restant simple à implémenter — Pydantic gère naturellement les listes typées.
from typing import List
class BatchRequest(BaseModel):
transactions: List[Transaction]
class BatchResponse(BaseModel):
predictions: List[Prediction]
@app.post("/predire_batch", response_model=BatchResponse)
def predire_batch(payload: BatchRequest, request: Request):
model = request.app.state.model
df = pd.DataFrame([t.model_dump() for t in payload.transactions])
proba = model.predict_proba(df)[:, 1]
preds = [
Prediction(classe=int(p >= SEUIL), probabilite_positive=round(float(p), 4), seuil_utilise=SEUIL)
for p in proba
]
return BatchResponse(predictions=preds)
Le calcul vectorisé sur le DataFrame complet est nettement plus efficace qu’une boucle d’appels unitaires : scikit-learn exploite NumPy en interne. Pour une centaine de lignes, le batch s’exécute en moins de 50 ms typiquement. Pour des batchs énormes (plus de 10 000 lignes), on les découpe en lots de 1000 côté client pour éviter les timeouts.
Étape 8 — Authentification par clé API
Une API ouverte au monde sans authentification est un risque immédiat. Pour un service interne, l’authentification par clé API dans un header HTTP est suffisante et triviale à implémenter. Pour des cas plus avancés (multi-utilisateurs, scopes, rotation), on bascule sur OAuth 2.0 ou JWT — natifs dans FastAPI mais plus longs à mettre en place.
import os
from fastapi import Depends, HTTPException
from fastapi.security import APIKeyHeader
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
# Lecture stricte : si la variable n'est pas définie, l'application refuse de démarrer
API_KEY_VALIDE = os.environ["API_KEY"]
if len(API_KEY_VALIDE) < 16:
raise RuntimeError("API_KEY trop courte — minimum 16 caractères aléatoires")
def verifier_cle(cle: str = Depends(api_key_header)):
if cle != API_KEY_VALIDE:
raise HTTPException(status_code=401, detail="Clé API invalide ou manquante")
return cle
@app.post("/predire", response_model=Prediction, dependencies=[Depends(verifier_cle)])
def predire(transaction: Transaction, request: Request):
... # implémentation comme à l'étape 5
La clé doit être stockée en variable d’environnement, jamais dans le code. La lecture stricte (os.environ["API_KEY"]) garantit que l’application refuse de démarrer si la variable n’est pas définie — c’est exactement ce qu’on veut, car une API ouverte par défaut serait pire qu’un service indisponible. La validation de longueur ajoute une dernière défense contre les clés faibles. Le header X-API-Key est passé à chaque requête côté client. En production, on génère plusieurs clés (une par client) et on les stocke dans une base avec leurs droits — mais pour une API interne avec un seul consommateur, une simple clé en environnement suffit.
Étape 9 — Logger et observer
Sans logging, on ne sait pas ce qui se passe en production. On veut au minimum tracer chaque requête, sa latence, et la prédiction renvoyée — anonymisée si nécessaire. Le module standard logging couplé à un middleware FastAPI fait l’affaire pour démarrer ; pour un usage avancé, on basculerait sur structlog avec format JSON et un agrégateur comme Loki.
import logging, time
from fastapi import Request
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("api_ml")
@app.middleware("http")
async def log_requete(request: Request, call_next):
debut = time.perf_counter()
reponse = await call_next(request)
duree_ms = (time.perf_counter() - debut) * 1000
log.info(f"{request.method} {request.url.path} → {reponse.status_code} ({duree_ms:.1f} ms)")
return reponse
Le middleware s’exécute à chaque requête et mesure la latence totale. Une ligne de log par requête suffit à détecter rapidement les anomalies en production : pic de latence, taux d’erreur 5xx, route particulièrement lente. Pour la traçabilité distribuée (un appel qui traverse plusieurs services), on ajouterait OpenTelemetry — natif dans FastAPI via une instrumentation auto.
Étape 10 — Conteneuriser avec Docker
Le déploiement final passe par Docker. Le Dockerfile reste minimaliste : une image Python slim, copie du code, installation des dépendances figées, exposition du port. La séparation en deux étapes (builder + runtime) réduit la taille finale et expose moins de dépendances potentiellement vulnérables.
# Dockerfile
FROM python:3.13-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt
FROM python:3.13-slim
RUN useradd -m -u 1001 app
WORKDIR /app
COPY --from=builder /root/.local /home/app/.local
COPY --chown=app:app . .
RUN chown -R app:app /home/app/.local
USER app
ENV PATH=/home/app/.local/bin:$PATH
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]
L’image se construit avec docker build -t api-ml . et se lance avec docker run -p 8000:8000 -e API_KEY=$(openssl rand -hex 24) -v $(pwd)/model_paiement.joblib:/app/model_paiement.joblib api-ml. Le conteneur tourne en utilisateur non-root (app, UID 1001) — bonne pratique sécurité qui évite l’escalade de privilèges si une vulnérabilité est exploitée. Le volume mappe le modèle depuis l’hôte, ce qui permet de mettre à jour le modèle sans reconstruire l’image. En production on pousserait l’image vers un registre (Docker Hub, GitHub Container Registry) et on déploierait via Kubernetes ou un PaaS comme Coolify ou Dokku.
Erreurs fréquentes
| Erreur | Cause | Solution |
|---|---|---|
| Modèle rechargé à chaque requête | joblib.load dans la fonction de route | Charger dans le lifespan, accéder via app.state |
| Latence brutalement haute | Workers Uvicorn = 1 sur multi-CPU | –workers 2 ou plus, gunicorn pour scale plus loin |
| 422 Unprocessable Entity | Schéma Pydantic non respecté | Lire le détail dans la réponse, corriger le client |
| Pickle error au load | Versions sklearn divergentes | Geler les versions, image Docker reproductible |
| Image Docker énorme (1+ Go) | Pas de séparation builder/runtime | Multi-stage build sur python:3.13-slim |
| Clé API dans le code Git | Hardcodage | Toujours via variables d’environnement |
| Timeout sur batch très large | Pas de découpe côté client | Lots de 1000 max et pagination |
Tutoriels associés
- Régression et classification avec scikit-learn
- Tracking d’expériences MLflow
- Dashboard métier avec Streamlit
- 🔝 Retour au guide principal : Sciences de données : la stack pratique 2026
Ressources officielles
- Documentation FastAPI
- Documentation Pydantic v2
- Documentation Uvicorn
- Image Docker python officielle
FAQ
FastAPI ou Flask ?
FastAPI pour un projet neuf : asynchrone natif, Pydantic intégré, OpenAPI automatique, performances supérieures. Flask reste pertinent pour des intégrations legacy ou des cas très simples sans besoin de validation typée.
Faut-il du async pour servir un modèle scikit-learn ?
Non, pas obligatoirement. scikit-learn est CPU-bound et synchrone. On définit les routes en def classique, FastAPI les exécute dans un threadpool. L’async devient utile si l’on appelle des bases de données ou des APIs externes pendant la requête.
Comment scaler horizontalement ?
Plusieurs conteneurs derrière un load balancer (Nginx, HAProxy, ou ingress Kubernetes). Chaque conteneur charge sa copie du modèle. La latence est limitée par les CPU disponibles, pas par la mémoire — un modèle scikit-learn fait quelques Mo.
Comment tester l’API ?
FastAPI fournit TestClient qui simule des requêtes HTTP sans démarrer le serveur. Couplé à pytest, on automatise les tests d’intégration : envoi de payload, vérification de la réponse et du code statut. Tester avec un modèle factice (mock) évite de charger le vrai modèle pour chaque test.