Développement Backend ML avec FastAPI

Découvrez comment créer des APIs ML performantes avec FastAPI.

Configuration de Base

1. Installation

pip install fastapi uvicorn pydantic numpy pandas scikit-learn

2. Structure du Projet

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(
    title="API ML",
    description="API pour prédictions ML",
    version="1.0.0"
)

# Modèles de données
class PredictionRequest(BaseModel):
    features: List[float]
    model_type: Optional[str] = "default"

class PredictionResponse(BaseModel):
    prediction: float
    confidence: float
    model_version: str

API Endpoints

1. Routes de Base

@app.get("/")
async def root():
    return {"message": "API ML active"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.get("/models")
async def list_models():
    return {
        "available_models": [
            "classification",
            "regression",
            "clustering"
        ]
    }

2. Endpoints de Prédiction

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # Chargement du modèle
        model = load_model(request.model_type)
        
        # Prédiction
        prediction = model.predict([request.features])[0]
        confidence = model.predict_proba([request.features])[0]
        
        return PredictionResponse(
            prediction=float(prediction),
            confidence=float(max(confidence)),
            model_version="1.0.0"
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Gestion des Modèles

1. Chargement et Gestion

from typing import Dict
import joblib
import os

class ModelManager:
    def __init__(self):
        self.models: Dict[str, object] = {}
        self.model_path = "models/"
        
    def load_model(self, model_type: str) -> object:
        if model_type not in self.models:
            path = os.path.join(self.model_path, f"{model_type}.pkl")
            if not os.path.exists(path):
                raise HTTPException(
                    status_code=404,
                    detail=f"Modèle {model_type} non trouvé"
                )
            self.models[model_type] = joblib.load(path)
        return self.models[model_type]

model_manager = ModelManager()

2. Validation des Données

from pydantic import validator
import numpy as np

class FeatureValidation(BaseModel):
    features: List[float]
    
    @validator('features')
    def validate_features(cls, v):
        if not v:
            raise ValueError("La liste des features ne peut pas être vide")
        if not all(isinstance(x, (int, float)) for x in v):
            raise ValueError("Toutes les features doivent être numériques")
        return v

Middleware et Sécurité

1. Middleware Personnalisé

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

2. Authentification

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token invalide"
        )

Tests et Documentation

1. Tests Unitaires

# test_api.py
from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_predict():
    response = client.post(
        "/predict",
        json={"features": [1.0, 2.0, 3.0]}
    )
    assert response.status_code == 200
    assert "prediction" in response.json()

2. Documentation API

# Ajout de documentation personnalisée
@app.get("/docs")
async def custom_docs():
    return {
        "endpoints": {
            "/predict": {
                "description": "Endpoint de prédiction",
                "parameters": {
                    "features": "Liste de features numériques",
                    "model_type": "Type de modèle à utiliser"
                }
            }
        }
    }

Bonnes Pratiques

1. Performance

   • Cache des modèles
   • Validation efficace
   • Gestion de la mémoire
   • Monitoring

2. Sécurité

   • Authentification
   • Validation des entrées
   • Rate limiting
   • Logging

3. Maintenance

   • Tests automatisés
   • Documentation
   • Versioning
   • Monitoring

Conclusion

Points clés à retenir :

• API performante
• Sécurité renforcée
• Documentation complète
• Tests automatisés

Recommandations :

• Suivre les standards REST
• Implémenter la sécurité
• Maintenir la documentation
• Automatiser les tests
• Se former sur FastAPI