TP4 - Construire une API de prediction avec Flask
Lab Pratique 60 min Intermediaire
Objectifs
A la fin de ce lab, vous serez capable de :
- Construire une application Flask qui sert des predictions ML
- Analyser et valider les donnees de requete JSON manuellement
- Implementer des endpoints de prediction et de verification de sante
- Gerer les erreurs de maniere elegante avec les codes de statut HTTP appropries
- Comparer votre implementation Flask avec la version FastAPI du TP3
Prerequis
- TP2 termine (modele serialise
model_v1.joblib) - TP3 termine (pour comparaison)
- Python 3.10+ installe
Meme modele, framework different
Ce lab construit exactement la meme API que le TP3 (memes endpoints, meme format d'entree/sortie) mais utilise Flask au lieu de FastAPI. Cela vous permet de comparer directement les deux approches.
Apercu de l'architecture
Etape 1 — Configuration du projet
1.1 Creer la structure du projet
mkdir -p flask-ml-api/app
mkdir -p flask-ml-api/models
cd flask-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 flask flask-cors scikit-learn joblib numpy
Creez requirements.txt :
flask>=3.0.0
flask-cors>=4.0.0
scikit-learn>=1.3.0
joblib>=1.3.0
numpy>=1.24.0
1.4 Copier votre modele
cp /path/to/tp2/model_v1.joblib models/model_v1.joblib
Etape 2 — Creer le validateur d'entree
Puisque Flask n'a pas de validation integree comme Pydantic, creez app/validators.py :
def validate_prediction_input(data):
"""
Validate prediction input data.
Returns (validated_data, None) on success
or (None, errors) on failure.
"""
if data is None:
return None, [{"message": "Request body must be valid JSON"}]
errors = []
validated = {}
schema = {
"age": {
"type": int,
"required": True,
"min": 18,
"max": 120,
"description": "Applicant age (18-120)",
},
"income": {
"type": float,
"required": True,
"min": 0,
"description": "Annual income (> 0)",
},
"credit_score": {
"type": int,
"required": True,
"min": 300,
"max": 850,
"description": "Credit score (300-850)",
},
"employment_years": {
"type": float,
"required": True,
"min": 0,
"description": "Years of employment (>= 0)",
},
"loan_amount": {
"type": float,
"required": True,
"min": 0,
"description": "Loan amount (> 0)",
},
}
for field, rules in schema.items():
if field not in data:
if rules["required"]:
errors.append({
"field": field,
"message": f"Missing required field: {field}",
"expected": rules["description"],
})
continue
try:
value = rules["type"](data[field])
except (ValueError, TypeError):
errors.append({
"field": field,
"message": f"Must be {rules['type'].__name__}",
"received": str(data[field]),
})
continue
if "min" in rules and value < rules["min"]:
errors.append({
"field": field,
"message": f"Must be >= {rules['min']}",
"received": value,
})
elif "max" in rules and value > rules["max"]:
errors.append({
"field": field,
"message": f"Must be <= {rules['max']}",
"received": value,
})
else:
validated[field] = value
if errors:
return None, errors
return validated, None
Comparez avec FastAPI
Dans le TP3, toute cette logique de validation etait geree par un seul modele Pydantic (~20 lignes). Ici, vous avez besoin de ~70 lignes de code de validation manuel. C'est le compromis cle de Flask.
Etape 3 — Creer le service ML
Creez app/ml_service.py (identique au TP3) :
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"
def load_model(self, model_path: str) -> None:
path = Path(model_path)
if not path.exists():
raise FileNotFoundError(f"Model 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:
if self.model is None:
raise RuntimeError("Model is not loaded")
arr = np.array([[
features["age"],
features["income"],
features["credit_score"],
features["employment_years"],
features["loan_amount"],
]])
prediction = self.model.predict(arr)[0]
probabilities = self.model.predict_proba(arr)[0]
return {
"prediction": "approved" if prediction == 1 else "denied",
"probability": round(float(max(probabilities)), 4),
"model_version": self.model_version,
}
@property
def is_ready(self) -> bool:
return self.model is not None
ml_service = MLService()
Etape 4 — Construire l'application Flask
Creez app/main.py :
from flask import Flask, request, jsonify
from flask_cors import CORS
from datetime import datetime
from app.ml_service import ml_service
from app.validators import validate_prediction_input
def create_app():
"""Application factory."""
app = Flask(__name__)
# CORS
CORS(app, origins=["http://localhost:3000"])
# Load model at startup
try:
ml_service.load_model("models/model_v1.joblib")
except FileNotFoundError as e:
print(f"[WARNING] {e}. Starting in degraded mode.")
# --- Routes ---
@app.route("/", methods=["GET"])
def root():
return jsonify({
"service": "Loan Prediction API (Flask)",
"version": "1.0.0",
"health": "/health",
})
@app.route("/health", methods=["GET"])
def health():
return jsonify({
"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(),
})
@app.route("/api/v1/predict", methods=["POST"])
def predict():
# Parse JSON
data = request.get_json(silent=True)
# Validate input
validated, errors = validate_prediction_input(data)
if errors:
return jsonify({
"error_code": "VALIDATION_ERROR",
"message": "Invalid input data",
"details": errors,
}), 422
# Check model
if not ml_service.is_ready:
return jsonify({
"error_code": "MODEL_NOT_LOADED",
"message": "Model is not available. Service is degraded.",
}), 503
# Run prediction
try:
result = ml_service.predict(validated)
return jsonify({
"prediction": result["prediction"],
"probability": result["probability"],
"model_version": result["model_version"],
"timestamp": datetime.utcnow().isoformat(),
})
except Exception as e:
return jsonify({
"error_code": "PREDICTION_FAILED",
"message": f"Prediction failed: {str(e)}",
}), 500
# --- Error Handlers ---
@app.errorhandler(404)
def not_found(error):
return jsonify({
"error_code": "NOT_FOUND",
"message": "The requested endpoint does not exist",
}), 404
@app.errorhandler(405)
def method_not_allowed(error):
return jsonify({
"error_code": "METHOD_NOT_ALLOWED",
"message": "This HTTP method is not allowed for this endpoint",
}), 405
@app.errorhandler(500)
def internal_error(error):
return jsonify({
"error_code": "INTERNAL_ERROR",
"message": "An unexpected error occurred",
}), 500
return app
Creez run.py a la racine du projet :
from app.main import create_app
app = create_app()
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)
Etape 5 — Lancer et tester
5.1 Demarrer le serveur
python run.py
Vous devriez voir :
[MLService] Model loaded: model_v1
* Running on http://0.0.0.0:5000
* Restarting with stat
* Debugger is active!
5.2 Tester l'endpoint de sante
curl http://localhost:5000/health
Attendu :
{
"status": "healthy",
"model_loaded": true,
"model_version": "model_v1",
"timestamp": "2026-02-23T14:30:00.000000"
}
5.3 Tester la prediction
curl -X POST http://localhost:5000/api/v1/predict \
-H "Content-Type: application/json" \
-d '{
"age": 35,
"income": 55000,
"credit_score": 720,
"employment_years": 8.5,
"loan_amount": 25000
}'
5.4 Tester les erreurs de validation
# Champs manquants
curl -X POST http://localhost:5000/api/v1/predict \
-H "Content-Type: application/json" \
-d '{"age": 35}'
# Valeurs invalides
curl -X POST http://localhost:5000/api/v1/predict \
-H "Content-Type: application/json" \
-d '{"age": -5, "income": 55000, "credit_score": 720, "employment_years": 8, "loan_amount": 25000}'
# Pas du JSON
curl -X POST http://localhost:5000/api/v1/predict \
-H "Content-Type: text/plain" \
-d 'this is not json'
Etape 6 — Comparer TP3 (FastAPI) vs TP4 (Flask)
Maintenant que vous avez construit la meme API dans les deux frameworks, comparez-les :
| Aspect | TP3 (FastAPI) | TP4 (Flask) |
|---|---|---|
| Total de fichiers | 3 (main, schemas, ml_service) | 4 (main, validators, ml_service, run) |
| Code de validation | ~25 lignes (modele Pydantic) | ~70 lignes (validateur manuel) |
| Code de route | ~15 lignes par route | ~25 lignes par route |
| Documentation Swagger | Auto-generee a /docs | Non disponible (necessite une extension) |
| Commande serveur | uvicorn app.main:app --reload | python run.py |
| Format d'erreur | Details 422 auto-generes | Format d'erreur personnalise |
| Surete des types | Validation complete au runtime | Conversion de type manuelle |
| Demarrage | Gestionnaire de contexte lifespan | Dans la factory create_app() |
Liste de verification
- Le serveur Flask demarre sur le port 5000
-
GET /healthretourne un statut sain -
POST /api/v1/predictavec des donnees valides retourne une prediction - Les champs manquants retournent 422 avec les details d'erreur specifiques
- Les valeurs invalides (age negatif, credit_score=1000) retournent 422
- Un corps non-JSON retourne 422
- Demander un endpoint inexistant retourne 404
Defis bonus
Defi 1 : Ajouter Flask-RESTX pour la documentation Swagger
pip install flask-restx
Remplacez l'application Flask basique par Flask-RESTX pour obtenir la documentation Swagger a /docs :
from flask_restx import Api, Resource, fields
api = Api(app, title="Loan Prediction API", version="1.0",
doc="/docs")
ns = api.namespace("api/v1", description="Predictions")
input_model = api.model("PredictionInput", {
"age": fields.Integer(required=True, min=18, max=120),
"income": fields.Float(required=True, min=0),
"credit_score": fields.Integer(required=True, min=300, max=850),
"employment_years": fields.Float(required=True, min=0),
"loan_amount": fields.Float(required=True, min=0),
})
@ns.route("/predict")
class Predict(Resource):
@ns.expect(input_model, validate=True)
def post(self):
"""Get a loan approval prediction."""
data = api.payload
result = ml_service.predict(data)
return result
Defi 2 : Ajouter la journalisation des requetes
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ml-api")
@app.before_request
def log_request():
logger.info(f"[{datetime.utcnow()}] {request.method} {request.path}")
@app.after_request
def log_response(response):
logger.info(f"[{datetime.utcnow()}] Response: {response.status_code}")
return response
Problemes courants
| Probleme | Solution |
|---|---|
ImportError: cannot import name 'create_app' | Assurez-vous que app/__init__.py existe |
jsonify retourne une page d'erreur HTML | Vous avez oublie les gestionnaires d'erreurs personnalises |
request.get_json() retourne None | L'en-tete Content-Type: application/json est manquant |
| Conflit du port 5000 (macOS) | macOS Monterey+ utilise le port 5000 pour AirPlay. Utilisez --port 5001 |
| Modele non trouve | Lancez python run.py depuis la racine du projet (ou se trouve models/) |