TP3 - Construire une API de prediction avec FastAPI

Lab Pratique 90 min Intermediaire

Objectifs

A la fin de ce lab, vous serez capable de :

Charger un modele ML serialise du Module 2 dans une application FastAPI
Definir des schemas Pydantic pour la validation des requetes et la serialisation des reponses
Implementer un endpoint /predict qui sert des predictions en temps reel
Implementer un endpoint /health pour la surveillance du service
Ajouter une gestion d'erreurs appropriee pour les scenarios d'echec courants
Tester l'API en utilisant uvicorn et l'interface Swagger UI auto-generee

Prerequis

TP2 termine (vous devriez avoir un fichier de modele serialise model_v1.joblib)
Python 3.10+ installe
Comprehension de base des API REST (concepts du Module 3)

Pas de modele du TP2 ?

Si vous n'avez pas termine le TP2, executez ce script pour creer un modele d'exemple :

# create_sample_model.py
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
import joblib

X, y = make_classification(
    n_samples=1000, n_features=5, n_informative=4,
    n_redundant=1, random_state=42,
)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
joblib.dump(model, "models/model_v1.joblib")
print("Model saved to models/model_v1.joblib")

Apercu de l'architecture

Etape 1 — Configuration du projet

1.1 Creer la structure du projet

mkdir -p fastapi-ml-api/app
mkdir -p fastapi-ml-api/models
cd fastapi-ml-api

1.2 Creer un environnement virtuel

python -m venv venv

# Windows
venv\Scripts\activate

# macOS/Linux
source venv/bin/activate

1.3 Installer les dependances

pip install fastapi uvicorn pydantic scikit-learn joblib numpy

Creez requirements.txt :

fastapi>=0.100.0
uvicorn>=0.23.0
pydantic>=2.0.0
scikit-learn>=1.3.0
joblib>=1.3.0
numpy>=1.24.0

1.4 Copier votre modele

Copiez le fichier de modele du TP2 dans le repertoire models/ :

cp /path/to/tp2/model_v1.joblib models/model_v1.joblib

Etape 2 — Definir les schemas Pydantic

Creez app/schemas.py :

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class PredictionInput(BaseModel):
    """Input features for the ML model."""

    age: int = Field(
        ...,
        ge=18,
        le=120,
        description="Applicant age in years",
        examples=[35],
    )
    income: float = Field(
        ...,
        gt=0,
        description="Annual income in USD",
        examples=[55000.0],
    )
    credit_score: int = Field(
        ...,
        ge=300,
        le=850,
        description="Credit score (FICO)",
        examples=[720],
    )
    employment_years: float = Field(
        ...,
        ge=0,
        description="Years of employment",
        examples=[8.5],
    )
    loan_amount: float = Field(
        ...,
        gt=0,
        description="Requested loan amount in USD",
        examples=[25000.0],
    )

    class Config:
        json_schema_extra = {
            "example": {
                "age": 35,
                "income": 55000.0,
                "credit_score": 720,
                "employment_years": 8.5,
                "loan_amount": 25000.0,
            }
        }


class PredictionOutput(BaseModel):
    """Prediction result from the ML model."""

    prediction: str = Field(..., description="Predicted class label")
    probability: float = Field(
        ...,
        ge=0,
        le=1,
        description="Prediction confidence (0 to 1)",
    )
    model_version: str = Field(..., description="Version of the model used")
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="UTC timestamp of the prediction",
    )


class HealthResponse(BaseModel):
    """Health check response."""

    status: str = Field(..., description="Service status")
    model_loaded: bool = Field(..., description="Whether the model is loaded")
    model_version: str = Field(..., description="Current model version")
    timestamp: str = Field(..., description="Current UTC time")


class ErrorResponse(BaseModel):
    """Standard error response."""

    error_code: str = Field(..., description="Machine-readable error code")
    message: str = Field(..., description="Human-readable error message")
    details: Optional[list] = Field(None, description="Additional error details")

Pourquoi definir des schemas ?

Validation : FastAPI rejette automatiquement les requetes qui ne correspondent pas au schema
Documentation : L'interface Swagger UI affiche les descriptions, types et contraintes des champs
Serialisation : Les donnees de reponse sont automatiquement formatees selon le schema de sortie

Etape 3 — Creer le service ML

Creez app/ml_service.py :

import joblib
import numpy as np
from pathlib import Path


class MLService:
    """Handles model loading and inference."""

    def __init__(self):
        self.model = None
        self.model_version = "unknown"
        self.feature_names = [
            "age", "income", "credit_score",
            "employment_years", "loan_amount",
        ]

    def load_model(self, model_path: str) -> None:
        """Load a serialized model from disk."""
        path = Path(model_path)
        if not path.exists():
            raise FileNotFoundError(
                f"Model file not found: {model_path}"
            )

        self.model = joblib.load(path)
        self.model_version = path.stem
        print(f"[MLService] Model loaded: {self.model_version}")

    def predict(self, features: dict) -> dict:
        """
        Run inference on input features.
        Returns prediction label and probability.
        """
        if self.model is None:
            raise RuntimeError("Model is not loaded")

        feature_array = np.array([[
            features["age"],
            features["income"],
            features["credit_score"],
            features["employment_years"],
            features["loan_amount"],
        ]])

        prediction = self.model.predict(feature_array)[0]
        probabilities = self.model.predict_proba(feature_array)[0]
        confidence = float(max(probabilities))

        label = "approved" if prediction == 1 else "denied"

        return {
            "prediction": label,
            "probability": round(confidence, 4),
            "model_version": self.model_version,
        }

    @property
    def is_ready(self) -> bool:
        return self.model is not None


# Singleton instance
ml_service = MLService()

Etape 4 — Construire l'application FastAPI

Creez app/main.py :

from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware

from app.schemas import (
    PredictionInput,
    PredictionOutput,
    HealthResponse,
    ErrorResponse,
)
from app.ml_service import ml_service


# --- Lifespan: load model at startup ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        ml_service.load_model("models/model_v1.joblib")
    except FileNotFoundError as e:
        print(f"[WARNING] {e}. API will start in degraded mode.")
    yield
    print("[INFO] Shutting down API...")


# --- FastAPI App ---
app = FastAPI(
    title="Loan Prediction API",
    description="ML-powered loan approval prediction service built in TP3",
    version="1.0.0",
    lifespan=lifespan,
    openapi_tags=[
        {
            "name": "Predictions",
            "description": "Submit features and receive ML predictions",
        },
        {
            "name": "System",
            "description": "Health checks and service monitoring",
        },
    ],
)

# --- CORS Middleware ---
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


# --- Health Check ---
@app.get(
    "/health",
    response_model=HealthResponse,
    tags=["System"],
    summary="Check service health",
)
def health_check():
    """Returns the current health status of the API and model."""
    return HealthResponse(
        status="healthy" if ml_service.is_ready else "degraded",
        model_loaded=ml_service.is_ready,
        model_version=ml_service.model_version,
        timestamp=datetime.utcnow().isoformat(),
    )


# --- Prediction Endpoint ---
@app.post(
    "/api/v1/predict",
    response_model=PredictionOutput,
    responses={
        422: {"model": ErrorResponse, "description": "Validation error"},
        503: {"model": ErrorResponse, "description": "Model not available"},
        500: {"model": ErrorResponse, "description": "Prediction failed"},
    },
    tags=["Predictions"],
    summary="Get a loan approval prediction",
)
def predict(input_data: PredictionInput):
    """
    Submit loan application features and receive a prediction.

    The model returns:
    - **prediction**: "approved" or "denied"
    - **probability**: confidence score between 0 and 1
    - **model_version**: which model version produced the result
    """
    # Check model availability
    if not ml_service.is_ready:
        raise HTTPException(
            status_code=503,
            detail="Model is not loaded. Service is in degraded mode.",
        )

    # Run prediction
    try:
        features = input_data.model_dump()
        result = ml_service.predict(features)

        return PredictionOutput(
            prediction=result["prediction"],
            probability=result["probability"],
            model_version=result["model_version"],
            timestamp=datetime.utcnow(),
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Prediction failed: {str(e)}",
        )


# --- Root ---
@app.get("/", tags=["System"])
def root():
    """API root — returns basic service information."""
    return {
        "service": "Loan Prediction API",
        "version": "1.0.0",
        "docs": "/docs",
        "health": "/health",
    }

Etape 5 — Lancer et tester

5.1 Demarrer le serveur

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

Vous devriez voir :

[MLService] Model loaded: model_v1
INFO:     Uvicorn running on http://0.0.0.0:8000
INFO:     Started reloader process

5.2 Acceder a la documentation Swagger

Ouvrez votre navigateur et naviguez vers : http://localhost:8000/docs

Vous devriez voir l'interface Swagger UI interactive avec :

Tag Predictions → POST /api/v1/predict
Tag System → GET /health, GET /

5.3 Tester l'endpoint de sante

curl http://localhost:8000/health

Reponse attendue :

{
  "status": "healthy",
  "model_loaded": true,
  "model_version": "model_v1",
  "timestamp": "2026-02-23T14:30:00.000000"
}

5.4 Tester l'endpoint de prediction

curl -X POST http://localhost:8000/api/v1/predict \
  -H "Content-Type: application/json" \
  -d '{
    "age": 35,
    "income": 55000,
    "credit_score": 720,
    "employment_years": 8.5,
    "loan_amount": 25000
  }'

Reponse attendue :

{
  "prediction": "approved",
  "probability": 0.87,
  "model_version": "model_v1",
  "timestamp": "2026-02-23T14:30:05.123456"
}

5.5 Tester les erreurs de validation

Envoyez des donnees invalides pour verifier la validation Pydantic :

curl -X POST http://localhost:8000/api/v1/predict \
  -H "Content-Type: application/json" \
  -d '{
    "age": -5,
    "income": 55000,
    "credit_score": 720,
    "employment_years": 8,
    "loan_amount": 25000
  }'

Attendu : 422 Unprocessable Entity avec les details sur le champ age.

curl -X POST http://localhost:8000/api/v1/predict \
  -H "Content-Type: application/json" \
  -d '{
    "age": 35,
    "income": 55000
  }'

Attendu : 422 avec les details sur les champs obligatoires manquants.

Etape 6 — Tester avec Swagger UI

Ouvrez http://localhost:8000/docs dans votre navigateur
Cliquez sur POST /api/v1/predict
Cliquez sur Try it out
Le JSON d'exemple est pre-rempli a partir de votre schema
Cliquez sur Execute
Observez le code de reponse, le corps et les en-tetes

Utilisez Swagger pour des tests rapides

Pendant le developpement, Swagger UI est plus rapide que d'ecrire des commandes curl. Utilisez-le pour :

Tester differentes entrees rapidement
Voir les formats exacts de requete/reponse
Verifier les reponses d'erreur
Partager la documentation API avec vos collegues

Etape 7 — Structure finale du projet

Votre projet termine devrait ressembler a :

fastapi-ml-api/
├── app/
│   ├── __init__.py          # vide
│   ├── main.py              # Application FastAPI
│   ├── schemas.py           # Modeles Pydantic
│   └── ml_service.py        # Chargement du modele et inference
├── models/
│   └── model_v1.joblib      # Modele ML serialise
├── requirements.txt
└── venv/

Liste de verification

Avant de marquer ce lab comme termine, verifiez :

uvicorn demarre sans erreurs
GET /health retourne {"status": "healthy", "model_loaded": true}
POST /api/v1/predict avec des donnees valides retourne une prediction
Des donnees invalides (age negatif, champs manquants) retournent 422
Swagger UI a /docs affiche tous les endpoints avec les schemas
La reponse inclut model_version et timestamp

Defis bonus

Defi 1 : Ajouter un endpoint de prediction par lot

Ajoutez un endpoint POST /api/v1/predict/batch qui accepte une liste d'entrees :

from typing import List

class BatchInput(BaseModel):
    inputs: List[PredictionInput] = Field(..., min_length=1, max_length=50)

class BatchOutput(BaseModel):
    predictions: List[PredictionOutput]
    total: int

@app.post("/api/v1/predict/batch", response_model=BatchOutput, tags=["Predictions"])
def predict_batch(batch: BatchInput):
    results = []
    for item in batch.inputs:
        features = item.model_dump()
        result = ml_service.predict(features)
        results.append(PredictionOutput(
            prediction=result["prediction"],
            probability=result["probability"],
            model_version=result["model_version"],
        ))
    return BatchOutput(predictions=results, total=len(results))

Defi 2 : Ajouter un middleware de chronometrage des requetes

import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        ms = (time.perf_counter() - start) * 1000
        response.headers["X-Response-Time-Ms"] = f"{ms:.2f}"
        return response

app.add_middleware(TimingMiddleware)

Defi 3 : Ajouter l'authentification par cle API

from fastapi import Depends, Header, HTTPException

API_KEYS = {"sk_test_abc123", "sk_test_def456"}

async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key not in API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

@app.post("/api/v1/predict", dependencies=[Depends(verify_api_key)])
def predict(input_data: PredictionInput):
    ...

Testez avec :

curl -X POST http://localhost:8000/api/v1/predict \
  -H "Content-Type: application/json" \
  -H "X-API-Key: sk_test_abc123" \
  -d '{"age": 35, "income": 55000, "credit_score": 720, "employment_years": 8, "loan_amount": 25000}'

Problemes courants

Probleme	Solution
`ModuleNotFoundError: app.schemas`	Assurez-vous que `app/__init__.py` existe (peut etre vide)
`FileNotFoundError: model_v1.joblib`	Verifiez que le fichier de modele est dans `models/` par rapport a l'endroit ou vous lancez uvicorn
Le port 8000 est deja utilise	Utilisez `--port 8001` ou arretez le processus existant
Les changements ne sont pas reflets	Assurez-vous que le flag `--reload` est defini avec uvicorn
Erreurs 422 sur des donnees apparemment valides	Verifiez les types de champs — Pydantic est strict (ex. `"35"` n'est pas un `int`)

Objectifs​

Prerequis​

Apercu de l'architecture​

Etape 1 — Configuration du projet​

1.1 Creer la structure du projet​

1.2 Creer un environnement virtuel​

1.3 Installer les dependances​

1.4 Copier votre modele​

Etape 2 — Definir les schemas Pydantic​

Etape 3 — Creer le service ML​

Etape 4 — Construire l'application FastAPI​

Etape 5 — Lancer et tester​

5.1 Demarrer le serveur​

5.2 Acceder a la documentation Swagger​

5.3 Tester l'endpoint de sante​

5.4 Tester l'endpoint de prediction​

5.5 Tester les erreurs de validation​

Etape 6 — Tester avec Swagger UI​

Etape 7 — Structure finale du projet​

Liste de verification​

Defis bonus​

Problemes courants​