Construire des API avec Flask
Qu'est-ce que Flask ?
Flask est un micro-framework web leger pour Python cree par Armin Ronacher en 2010. Il est appele un framework "micro" car il n'inclut pas d'outils integres pour l'abstraction de base de donnees, la validation de formulaires ou l'authentification — vous choisissez et ajoutez les extensions dont vous avez besoin.
Flask a ete le framework standard pour deployer des modeles ML pendant des annees et reste extremement populaire grace a sa simplicite et son ecosysteme massif.
La philosophie de Flask
Voir l'architecture Flask
Flask est souvent compare a Django (framework complet). Pour les API ML, Flask est prefere car vous n'avez pas besoin du panneau d'administration, de l'ORM ou du moteur de templates de Django. Vous avez juste besoin d'un routage HTTP leger.
Flask vs FastAPI — Quand utiliser lequel ?
C'est l'une des questions les plus courantes dans le deploiement ML. Voici une comparaison complete :
| Fonctionnalite | Flask | FastAPI |
|---|---|---|
| Annee de sortie | 2010 | 2018 |
| Architecture | WSGI (synchrone) | ASGI (asynchrone) |
| Annotations de type | Optionnelles, non utilisees par le framework | Obligatoires, pilotent la validation |
| Validation des donnees | Manuelle ou via extensions | Integree via Pydantic |
| Documentation API | Flask-RESTX / Flasgger | Auto-generee (Swagger + ReDoc) |
| Support async | Limite (Flask 2.0+) | Natif, de premiere classe |
| Performance | Bonne | Excellente |
| Courbe d'apprentissage | Tres faible | Faible |
| Ecosysteme | Massif (15+ ans) | En croissance rapide |
| Taille de la communaute | Tres grande | Grande et active |
| Pret pour la production | Eprouve au combat | Prouve en production |
Matrice de decision
- Choisissez Flask quand vous voulez une simplicite maximale, avez une base de code Flask existante ou avez besoin d'extensions matures.
- Choisissez FastAPI pour les nouveaux projets qui beneficient de la documentation automatique, de la validation et de la performance async.
- Les deux sont excellents pour servir des modeles ML. Le "meilleur" choix depend de votre equipe et de votre projet.
Installation et configuration
pip install flask flask-cors
pip install scikit-learn joblib numpy pandas
Structure du projet
flask-ml-api/
├── app/
│ ├── __init__.py # Factory de l'application Flask
│ ├── routes/
│ │ ├── __init__.py
│ │ └── predictions.py
│ ├── services/
│ │ ├── __init__.py
│ │ └── ml_service.py
│ └── utils/
│ ├── __init__.py
│ └── validators.py
├── models/
│ └── model_v1.joblib
├── config.py
├── run.py
└── requirements.txt
Votre premiere API Flask
Exemple minimal
from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/")
def home():
return jsonify({"message": "ML Prediction API is running"})
@app.route("/health")
def health():
return jsonify({"status": "healthy"})
if __name__ == "__main__":
app.run(debug=True, port=5000)
Lancez-le :
python run.py
Votre API est disponible a http://localhost:5000.
Analyse des requetes dans Flask
Contrairement a FastAPI, Flask n'a pas de validation Pydantic integree. Vous analysez les donnees de requete manuellement depuis l'objet request.
Analyse de donnees JSON
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/api/v1/predict", methods=["POST"])
def predict():
# Get JSON from request body
data = request.get_json()
if data is None:
return jsonify({"error": "Request body must be JSON"}), 400
# Manual validation
required_fields = ["age", "income", "credit_score",
"employment_years", "loan_amount"]
missing = [f for f in required_fields if f not in data]
if missing:
return jsonify({
"error": "Missing required fields",
"missing_fields": missing,
}), 422
# Type validation
try:
age = int(data["age"])
income = float(data["income"])
credit_score = int(data["credit_score"])
except (ValueError, TypeError) as e:
return jsonify({
"error": f"Invalid data type: {str(e)}",
}), 422
# Range validation
if not (18 <= age <= 120):
return jsonify({
"error": "Age must be between 18 and 120",
}), 422
# Proceed with prediction...
return jsonify({"prediction": "approved", "probability": 0.87})
Remarquez la quantite de code repetitif necessaire pour la validation dans Flask. Avec FastAPI, c'est un seul modele Pydantic. Pour les API complexes, c'est une raison pour laquelle FastAPI permet d'economiser un temps de developpement significatif.
Utilisation d'un helper de validation
Pour reduire la repetition, creez un validateur reutilisable :
def validate_prediction_input(data):
"""Validate prediction input and return errors if any."""
errors = []
if data is None:
return None, [{"message": "Request body must be JSON"}]
schema = {
"age": {"type": int, "min": 18, "max": 120, "required": True},
"income": {"type": float, "min": 0, "required": True},
"credit_score": {"type": int, "min": 300, "max": 850, "required": True},
"employment_years": {"type": float, "min": 0, "required": True},
"loan_amount": {"type": float, "min": 0, "required": True},
}
validated = {}
for field, rules in schema.items():
if field not in data:
if rules["required"]:
errors.append({"field": field, "message": "Field is required"})
continue
try:
value = rules["type"](data[field])
except (ValueError, TypeError):
errors.append({
"field": field,
"message": f"Must be {rules['type'].__name__}",
})
continue
if "min" in rules and value < rules["min"]:
errors.append({
"field": field,
"message": f"Must be >= {rules['min']}",
})
elif "max" in rules and value > rules["max"]:
errors.append({
"field": field,
"message": f"Must be <= {rules['max']}",
})
else:
validated[field] = value
if errors:
return None, errors
return validated, None
Chargement et service d'un modele ML
Service de modele
import joblib
import numpy as np
from pathlib import Path
class MLService:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.model = None
cls._instance.version = "unknown"
return cls._instance
def load_model(self, model_path: str):
path = Path(model_path)
if not path.exists():
raise FileNotFoundError(f"Model not found: {model_path}")
self.model = joblib.load(path)
self.version = path.stem
def predict(self, features: dict) -> dict:
if self.model is None:
raise RuntimeError("Model 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": float(max(probabilities)),
"model_version": self.version,
}
Pattern Application Factory
L'application factory est une bonne pratique Flask. Au lieu de creer l'application dans le scope global, vous utilisez une fonction qui la cree et la configure.
from flask import Flask
from flask_cors import CORS
from app.services.ml_service import MLService
def create_app(config=None):
app = Flask(__name__)
if config:
app.config.update(config)
CORS(app, origins=["http://localhost:3000"])
ml_service = MLService()
ml_service.load_model("models/model_v1.joblib")
app.ml_service = ml_service
from app.routes.predictions import predictions_bp
app.register_blueprint(predictions_bp, url_prefix="/api/v1")
return app
Blueprints — Organisation des routes
Les Blueprints Flask vous permettent d'organiser les routes en modules logiques — similaire a l'utilisation de APIRouter par FastAPI.
from flask import Blueprint, request, jsonify, current_app
from datetime import datetime
predictions_bp = Blueprint("predictions", __name__)
@predictions_bp.route("/predict", methods=["POST"])
def predict():
data = request.get_json()
validated, errors = validate_prediction_input(data)
if errors:
return jsonify({"errors": errors}), 422
ml_service = current_app.ml_service
try:
result = ml_service.predict(validated)
return jsonify({
"prediction": result["prediction"],
"probability": result["probability"],
"model_version": result["model_version"],
"timestamp": datetime.utcnow().isoformat(),
})
except RuntimeError as e:
return jsonify({"error": str(e)}), 503
except Exception as e:
return jsonify({"error": f"Prediction failed: {str(e)}"}), 500
@predictions_bp.route("/health", methods=["GET"])
def health():
ml_service = current_app.ml_service
return jsonify({
"status": "healthy" if ml_service.model else "degraded",
"model_loaded": ml_service.model is not None,
"model_version": ml_service.version,
})
Architecture des Blueprints
Gestionnaires d'erreurs
Flask vous permet d'enregistrer des gestionnaires d'erreurs personnalises pour des codes de statut HTTP specifiques ou des types d'exception.
from werkzeug.exceptions import HTTPException
@app.errorhandler(404)
def not_found(error):
return jsonify({
"error_code": "NOT_FOUND",
"message": "The requested resource was not found",
}), 404
@app.errorhandler(422)
def validation_error(error):
return jsonify({
"error_code": "VALIDATION_ERROR",
"message": "Request validation failed",
"details": error.description if hasattr(error, "description") else str(error),
}), 422
@app.errorhandler(500)
def internal_error(error):
return jsonify({
"error_code": "INTERNAL_ERROR",
"message": "An unexpected error occurred",
}), 500
@app.errorhandler(Exception)
def handle_unexpected(error):
"""Catch-all for unhandled exceptions."""
if isinstance(error, HTTPException):
return jsonify({"error": error.description}), error.code
return jsonify({
"error_code": "UNEXPECTED_ERROR",
"message": "Something went wrong",
}), 500
Flask-RESTX pour la documentation API
Flask ne genere pas automatiquement la documentation Swagger. Flask-RESTX est une extension qui ajoute la documentation Swagger.
pip install flask-restx
from flask import Flask
from flask_restx import Api, Resource, fields
app = Flask(__name__)
api = Api(
app,
title="ML Prediction API",
version="1.0",
description="Loan approval prediction service",
doc="/docs",
)
ns = api.namespace("predictions", description="Prediction operations")
input_model = api.model("PredictionInput", {
"age": fields.Integer(required=True, min=18, max=120,
description="Applicant age"),
"income": fields.Float(required=True, min=0,
description="Annual income"),
"credit_score": fields.Integer(required=True, min=300, max=850,
description="FICO score"),
"employment_years": fields.Float(required=True, min=0,
description="Years employed"),
"loan_amount": fields.Float(required=True, min=0,
description="Loan amount"),
})
output_model = api.model("PredictionOutput", {
"prediction": fields.String(description="Predicted class"),
"probability": fields.Float(description="Confidence score"),
"model_version": fields.String(description="Model version"),
"timestamp": fields.DateTime(description="Prediction timestamp"),
})
@ns.route("/predict")
class Predict(Resource):
@ns.expect(input_model, validate=True)
@ns.marshal_with(output_model, code=200)
@ns.response(422, "Validation Error")
@ns.response(500, "Internal Server Error")
def post(self):
"""Submit features for loan approval prediction."""
data = api.payload
result = ml_service.predict(data)
return result
Visitez http://localhost:5000/docs pour voir l'interface Swagger UI.
API ML Flask complete
Voici une application Flask complete et executable :
from flask import Flask, request, jsonify
from flask_cors import CORS
from datetime import datetime
import joblib
import numpy as np
# --- App Setup ---
app = Flask(__name__)
CORS(app, origins=["http://localhost:3000"])
# --- Model Loading ---
model = None
model_version = "unknown"
def load_model():
global model, model_version
model = joblib.load("models/model_v1.joblib")
model_version = "v1.0"
# --- Routes ---
@app.route("/health", methods=["GET"])
def health():
return jsonify({
"status": "healthy" if model else "degraded",
"model_version": model_version,
})
@app.route("/api/v1/predict", methods=["POST"])
def predict():
data = request.get_json()
if not data:
return jsonify({"error": "JSON body required"}), 400
required = ["age", "income", "credit_score",
"employment_years", "loan_amount"]
missing = [f for f in required if f not in data]
if missing:
return jsonify({"error": "Missing fields", "fields": missing}), 422
try:
features = np.array([[
int(data["age"]),
float(data["income"]),
int(data["credit_score"]),
float(data["employment_years"]),
float(data["loan_amount"]),
]])
except (ValueError, TypeError) as e:
return jsonify({"error": f"Invalid data: {str(e)}"}), 422
if model is None:
return jsonify({"error": "Model not loaded"}), 503
try:
pred = model.predict(features)[0]
proba = model.predict_proba(features)[0]
return jsonify({
"prediction": "approved" if pred == 1 else "denied",
"probability": round(float(max(proba)), 4),
"model_version": model_version,
"timestamp": datetime.utcnow().isoformat(),
})
except Exception as e:
return jsonify({"error": f"Prediction failed: {str(e)}"}), 500
# --- Startup ---
if __name__ == "__main__":
load_model()
app.run(debug=True, host="0.0.0.0", port=5000)
Cote a cote : Flask vs FastAPI
Le meme endpoint de prediction dans les deux frameworks :
Version Flask
@app.route("/api/v1/predict", methods=["POST"])
def predict():
data = request.get_json()
if not data:
return jsonify({"error": "JSON required"}), 400
required = ["age", "income", "credit_score"]
missing = [f for f in required if f not in data]
if missing:
return jsonify({"error": f"Missing: {missing}"}), 422
result = ml_service.predict(data)
return jsonify(result)
Version FastAPI
@app.post("/api/v1/predict", response_model=PredictionOutput)
def predict(data: PredictionInput):
result = ml_service.predict(data.model_dump())
return result
| Aspect | Flask | FastAPI |
|---|---|---|
| Lignes de code | ~12 lignes | ~4 lignes |
| Validation | Manuelle, verbeuse | Automatique via Pydantic |
| Messages d'erreur | Personnalises pour chaque verification | Auto-generes, detailles |
| Documentation | Necessite Flask-RESTX | Auto-generee |
| Surete des types | Aucune au runtime | Validation complete au runtime |
Resume
| Sujet | Point cle |
|---|---|
| Flask | Micro-framework leger et eprouve au combat |
| Analyse des requetes | Manuelle avec request.get_json() |
| Validation | Manuelle ou via extensions (plus de code repetitif) |
| Blueprints | Organiser les routes en modules |
| Gestionnaires d'erreurs | Enregistrer par code de statut ou type d'exception |
| Flask-RESTX | Ajoute la documentation Swagger a Flask |
| vs FastAPI | Plus simple mais plus manuel ; FastAPI automatise la validation et la doc |
Reference rapide Flask
| Action | Code |
|---|---|
| Creer l'application | app = Flask(__name__) |
| Route GET | @app.route("/path", methods=["GET"]) |
| Route POST | @app.route("/path", methods=["POST"]) |
| Obtenir le corps JSON | request.get_json() |
| Retourner du JSON | jsonify({"key": "value"}) |
| Retourner avec statut | return jsonify({...}), 422 |
| Enregistrer un blueprint | app.register_blueprint(bp, url_prefix="/api") |
| Lancer l'application | app.run(debug=True, port=5000) |
| Ajouter CORS | CORS(app, origins=[...]) |
| Gestionnaire d'erreur | @app.errorhandler(404) |