ML Backend Development with FastAPI

Learn how to build high-performance ML APIs with FastAPI.

Basic Setup

1. Installation

pip install fastapi uvicorn pydantic numpy pandas scikit-learn

2. Project Structure

# 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"
)

# Data models
class PredictionRequest(BaseModel):
    features: List[float]
    model_type: Optional[str] = "default"

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

API Endpoints

1. Basic Routes

@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. Prediction Endpoints

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # Load the model
        model = load_model(request.model_type)
        
        # Prediction
        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))

Model Management

1. Loading and Management

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. Data Validation

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 and Security

1. Custom Middleware

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. Authentication

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"
        )

Testing and Documentation

1. Unit Tests

# 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. API Documentation

# Adding custom documentation
@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"
                }
            }
        }
    }

Best Practices

1. Performance

   • Model caching
   • Efficient validation
   • Memory management
   • Monitoring

2. Security

   • Authentication
   • Input validation
   • Rate limiting
   • Logging

3. Maintenance

   • Automated testing
   • Documentation
   • Versioning
   • Monitoring

Conclusion

Key takeaways:

• High-performance API
• Strengthened security
• Complete documentation
• Automated testing

Recommendations:

• Follow REST standards
• Implement security
• Keep the documentation up to date
• Automate testing
• Keep learning FastAPI