Avancé
Architecture Flask
Pour une API maintenable, il faut séparer les responsabilités. Cette page présente l'architecture Blueprint + Controller + Repository, adaptée aux APIs de données.
Pas de Clean Architecture ici
Contrairement à un cours de génie logiciel, on ne vise pas une architecture en couches (Domain, Application, Infrastructure). Blueprint, Controller, Repository : c'est suffisant pour la majorité des APIs de données.
Les trois couches
Chargement du diagramme…
| Couche | Rôle | Fichier |
|---|---|---|
| Blueprint (Routes) | Déclare les endpoints, extrait les paramètres de la requête | routes/ventes.py |
| Controller | Valide les paramètres, appelle le repository, formate la réponse | controllers/ventes_controller.py |
| Repository | Exécute les requêtes SQL, retourne les données brutes | repositories/ventes_repository.py |
Structure de fichiers
mon-api/
├── app.py # Point d'entrée Flask
├── routes/
│ ├── __init__.py
│ └── ventes.py # Blueprint des ventes
├── controllers/
│ ├── __init__.py
│ └── ventes_controller.py # Logique métier des ventes
├── repositories/
│ ├── __init__.py
│ └── ventes_repository.py # Requêtes SQL
├── dtos/
│ ├── __init__.py
│ └── ventes_dto.py # Structures de réponse
├── database.py # Connexion SQLite
└── requirements.txt
DTO ?
Un DTO (Data Transfer Object) est un simple dictionnaire ou dataclass qui définit la structure des données envoyées au client. C'est un contrat : le client sait exactement quels champs attendre.
Exemple complet : ressource ventes
1. Point d'entrée (app.py)
python
from flask import Flask
from routes.ventes import ventes_bp
def create_app():
app = Flask(__name__)
# Enregistrer les Blueprints
app.register_blueprint(ventes_bp)
return app
if __name__ == '__main__':
app = create_app()
app.run(debug=True)
2. Routes (routes/ventes.py)
Le Blueprint déclare les endpoints et délègue au controller.
python
from flask import Blueprint, request, jsonify
from controllers.ventes_controller import VentesController
ventes_bp = Blueprint('ventes', __name__, url_prefix='/api/v1/ventes')
controller = VentesController()
@ventes_bp.route('', methods=['GET'])
def lister():
"""Liste paginée des ventes."""
params = {
'region': request.args.get('region'),
'annee': request.args.get('annee', type=int),
'limit': request.args.get('limit', 20, type=int),
'offset': request.args.get('offset', 0, type=int),
'sort_by': request.args.get('sort_by', 'date'),
'order': request.args.get('order', 'desc'),
}
result = controller.lister_ventes(params)
return jsonify(result), 200
@ventes_bp.route('/<int:vente_id>', methods=['GET'])
def detail(vente_id):
"""Détail d'une vente."""
result = controller.detail_vente(vente_id)
if result is None:
return jsonify({"error": "Vente non trouvée"}), 404
return jsonify(result), 200
@ventes_bp.route('/stats', methods=['GET'])
def statistiques():
"""Statistiques agrégées."""
params = {
'region': request.args.get('region'),
'annee': request.args.get('annee', type=int),
}
result = controller.statistiques(params)
return jsonify(result), 200
3. Controller (controllers/ventes_controller.py)
Le controller valide les paramètres et appelle le repository.
python
from repositories.ventes_repository import VentesRepository
class VentesController:
def __init__(self):
self.repository = VentesRepository()
def lister_ventes(self, params):
"""Valide les paramètres et retourne la liste paginée."""
# Validation
limit = min(params['limit'], 100) # Maximum 100 par page
offset = max(params['offset'], 0)
# Appel au repository
ventes = self.repository.find_all(
region=params['region'],
annee=params['annee'],
limit=limit,
offset=offset,
sort_by=params['sort_by'],
order=params['order'],
)
total = self.repository.count(
region=params['region'],
annee=params['annee'],
)
# Formatage de la réponse (enveloppe)
return {
"data": ventes,
"total": total,
"limit": limit,
"offset": offset,
}
def detail_vente(self, vente_id):
"""Retourne le détail d'une vente ou None."""
return self.repository.find_by_id(vente_id)
def statistiques(self, params):
"""Retourne les statistiques agrégées."""
return self.repository.get_stats(
region=params['region'],
annee=params['annee'],
)
4. Repository (repositories/ventes_repository.py)
Le repository encapsule les requêtes SQL.
python
import sqlite3
class VentesRepository:
def __init__(self, db_path='data.db'):
self.db_path = db_path
def _get_connection(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def find_all(self, region=None, annee=None, limit=20, offset=0,
sort_by='date', order='desc'):
"""Récupère les ventes avec filtres, pagination et tri."""
conn = self._get_connection()
query = "SELECT * FROM ventes WHERE 1=1"
params = []
if region:
query += " AND region = ?"
params.append(region)
if annee:
query += " AND strftime('%Y', date) = ?"
params.append(str(annee))
# Tri (colonnes autorisées uniquement)
colonnes_autorisees = ['date', 'montant', 'region']
if sort_by in colonnes_autorisees:
direction = 'DESC' if order == 'desc' else 'ASC'
query += f" ORDER BY {sort_by} {direction}"
query += " LIMIT ? OFFSET ?"
params.extend([limit, offset])
rows = conn.execute(query, params).fetchall()
conn.close()
return [dict(row) for row in rows]
def count(self, region=None, annee=None):
"""Compte le nombre total de ventes (pour la pagination)."""
conn = self._get_connection()
query = "SELECT COUNT(*) as total FROM ventes WHERE 1=1"
params = []
if region:
query += " AND region = ?"
params.append(region)
if annee:
query += " AND strftime('%Y', date) = ?"
params.append(str(annee))
result = conn.execute(query, params).fetchone()
conn.close()
return result['total']
def find_by_id(self, vente_id):
"""Récupère une vente par son ID."""
conn = self._get_connection()
row = conn.execute(
"SELECT * FROM ventes WHERE id = ?", (vente_id,)
).fetchone()
conn.close()
return dict(row) if row else None
def get_stats(self, region=None, annee=None):
"""Statistiques agrégées des ventes."""
conn = self._get_connection()
query = """
SELECT
COUNT(*) as nombre_ventes,
SUM(montant) as total_montant,
AVG(montant) as montant_moyen,
MIN(montant) as montant_min,
MAX(montant) as montant_max
FROM ventes WHERE 1=1
"""
params = []
if region:
query += " AND region = ?"
params.append(region)
if annee:
query += " AND strftime('%Y', date) = ?"
params.append(str(annee))
result = conn.execute(query, params).fetchone()
conn.close()
return dict(result)
Flux d'une requête
Chargement du diagramme…
Quel est le rôle du Repository dans cette architecture ?
Pourquoi limiter la valeur de 'limit' dans le Controller (ex: max 100) ?
À retenir
Points clés
- L'architecture Blueprint + Controller + Repository suffit pour une API de données
- Le Blueprint déclare les routes, le Controller valide et orchestre, le Repository accède aux données
- Les DTOs définissent la structure des réponses
- Toujours valider et plafonner les paramètres de pagination dans le Controller