# [PYTHON] Formation Flask — Module 1
## De Zéro à Professionnel : Fondations & Base de données
### Parties I & II — Chapitres 1 à 6

---

> [OBJECTIF] **Projet Fil Rouge : `UrbanPulse`**
> Une plateforme communautaire urbaine où les habitants d'une ville signalent des incidents (nids-de-poule, éclairage défectueux, graffitis...), votent pour prioriser les réparations, et suivent l'état de résolution par les services municipaux.
>
> Ce projet sera construit **chapitre par chapitre**, en ajoutant des briques réelles à chaque étape.

---

## [WORLD_MAP] Table des matières

- [Chapitre 1 — Introduction à Flask](#chapitre-1)
- [Chapitre 2 — Installation et Configuration](#chapitre-2)
- [Chapitre 3 — Concepts de Base](#chapitre-3)
- [Chapitre 4 — Introduction aux ORM](#chapitre-4)
- [Chapitre 5 — Migrations](#chapitre-5)
- [Chapitre 6 — Relations entre Modèles](#chapitre-6)

---

<a name="chapitre-1"></a>
# [GUIDE] Chapitre 1 — Introduction à Flask

## 1.1 Qu'est-ce que Flask ?

Flask est un **microframework web Python**, créé par Armin Ronacher en 2010. Le terme "micro" ne signifie pas qu'il est limité ou incomplet — il signifie qu'il fournit **le minimum indispensable** pour construire une application web, laissant le développeur libre de choisir ses outils pour tout le reste.

Flask repose sur deux bibliothèques fondamentales :
- **Werkzeug** : une boîte à outils WSGI (Web Server Gateway Interface) qui gère les requêtes/réponses HTTP, le routing, et le serveur de développement.
- **Jinja2** : un moteur de templates puissant et sécurisé qui permet de générer du HTML dynamique.

### Pourquoi utiliser Flask ?

Flask brille dans des contextes précis :

**Avantages :**
- Courbe d'apprentissage douce — on peut avoir une app fonctionnelle en 5 lignes
- Liberté architecturale totale — pas d'impositions sur la structure du projet
- Parfait pour les APIs REST, les microservices, les outils internes
- Communauté massive et écosystème d'extensions riche
- Idéal pour l'apprentissage : on comprend ce qu'on fait à chaque étape

**Inconvénients (relatifs) :**
- Pas de composants "batteries included" comme Django (pas d'admin panel intégré, pas d'auth intégrée...)
- Nécessite plus de décisions architecturales au démarrage
- Peut devenir difficile à maintenir si mal structuré

---

## 1.2 Flask vs Django vs FastAPI

| Critère | Flask | Django | FastAPI |
|---|---|---|---|
| Type | Microframework | Full-stack | Microframework moderne |
| Courbe apprentissage | Facile | Moyenne | Facile-Moyenne |
| Performance | Bonne | Bonne | Excellente (async natif) |
| Admin Panel | [X] (extension) | [OK] intégré | [X] |
| ORM | Extension | Intégré | Extension |
| API REST | Très bon | Possible | Excellent (OpenAPI auto) |
| Maturité | Très mature (2010) | Très mature (2005) | Récent (2018) |
| Cas d'usage | APIs, microservices, sites | Applications complètes | APIs haute performance |

**Quand choisir Flask ?**
- Vous apprenez le développement web Python
- Vous construisez une API REST ou un microservice
- Vous avez besoin d'une grande flexibilité architecturale
- Votre équipe préfère assembler des pièces plutôt qu'utiliser un framework monolithique

---

## 1.3 Philosophie Flask : microframework + extensions

La philosophie de Flask s'exprime dans cette phrase de sa documentation officielle :

> *"Flask is a microframework for Python based on Werkzeug, Jinja2. It's designed to make getting started quick and easy, with the ability to scale up to complex applications."*

Le modèle mental à adopter :

```
Flask Core (requis)
    ├── Routing HTTP
    ├── Gestion Request/Response
    ├── Templates Jinja2
    └── Serveur de dev

Extensions (optionnelles, selon vos besoins)
    ├── Flask-SQLAlchemy -> Base de données ORM
    ├── Flask-Migrate -> Migrations de schéma
    ├── Flask-Login -> Authentification
    ├── Flask-WTF -> Formulaires avec protection CSRF
    ├── Flask-JWT-Extended -> JWT pour APIs
    ├── Flask-Mail -> Envoi d'emails
    ├── Flask-Limiter -> Rate limiting
    └── Flask-Caching -> Cache Redis/Memcached
```

---

## 1.4 Cas d'usage réels

- **Airbnb** a utilisé Flask pour plusieurs microservices internes
- **Netflix** utilise Flask pour certains outils de gestion
- **Reddit** a des composants Flask dans son infrastructure
- **Lyft** utilise Flask pour ses APIs internes

---

## 1.5 Notre projet UrbanPulse — Vision initiale

**UrbanPulse** est une application web qui permettra à des citoyens de :
1. **Signaler** des problèmes urbains (avec photos, localisation, catégorie)
2. **Voter** pour prioriser certains incidents
3. **Suivre** l'état de résolution (En attente, En cours, Résolu)
4. **Commenter** et interagir avec les signalements
5. **S'authentifier** pour éviter les abus
6. (Avancé) **Recevoir des notifications** par email lors des mises à jour

---

## [EDIT] Exercice 1.1 — Réflexion architecturale

> **Objectif** : Comprendre les choix technologiques avant de coder.

Répondez par écrit à ces questions (gardez vos réponses, vous y reviendrez en fin de formation) :

1. Selon vous, quels sont les 3 types d'utilisateurs de UrbanPulse ? (ex: citoyen, modérateur, admin)
2. Listez 5 "actions" que chaque type d'utilisateur peut faire.
3. Dessinez (même à la main) un schéma simple montrant les grandes entités de la base de données que vous imaginez (ex: User, Incident, Comment...).
4. Flask est-il le bon choix pour ce projet ? Justifiez en 3 phrases.

---

<a name="chapitre-2"></a>
# [GUIDE] Chapitre 2 — Installation et Configuration

## 2.1 Environnement de développement Python

### Pourquoi un environnement virtuel ?

Imaginez que vous ayez deux projets : l'un utilise Flask 2.3, l'autre Flask 1.1. Sans environnement virtuel, les deux partageraient la même installation Python système — impossible de faire cohabiter les deux versions. Un **environnement virtuel** isole les dépendances de chaque projet.

### Installation avec `venv` (outil standard Python)

```bash
# 1. Vérifiez votre version Python (besoin de 3.8+)
python --version
# ou
python3 --version

# 2. Créez le dossier de votre projet
mkdir urbanpulse
cd urbanpulse

# 3. Créez un environnement virtuel nommé "venv"
python -m venv venv

# 4. Activez l'environnement virtuel
# Sur Windows :
venv\Scripts\activate
# Sur macOS/Linux :
source venv/bin/activate

# Votre terminal affiche maintenant (venv) en préfixe
# (venv) $ 

# 5. Installez Flask
pip install flask

# 6. Vérifiez l'installation
python -c "import flask; print(flask.__version__)"
# Devrait afficher quelque chose comme : 3.0.3

# 7. Sauvegardez les dépendances
pip freeze > requirements.txt
```

### Alternative avec `poetry` (recommandé pour les projets sérieux)

```bash
# Installez poetry (une seule fois)
curl -sSL https://install.python-poetry.org | python3 -

# Dans votre dossier projet
poetry new urbanpulse
cd urbanpulse
poetry add flask

# Pour activer le shell
poetry shell
```

---

## 2.2 Structure minimale d'une application Flask

Créons notre premier fichier Flask :

```python
# app.py — Le fichier le plus simple possible

from flask import Flask  # Import de la classe Flask

# Création de l'instance de l'application
# __name__ est le nom du module Python actuel
# Flask l'utilise pour trouver les fichiers ressources (templates, static)
app = Flask(__name__)

# Décoration d'une fonction pour en faire une route
# Quand l'utilisateur visite "/", Flask appelle cette fonction
@app.route("/")
def hello():
    return "Hello, UrbanPulse ! [CITYSCAPE]"

# Point d'entrée du programme
if __name__ == "__main__":
    # debug=True active :
    # - Le rechargement automatique quand le code change
    # - Un debugger interactif dans le navigateur en cas d'erreur
    # [ATTENTION] JAMAIS en production !
    app.run(debug=True)
```

### Décryptage ligne par ligne

**`from flask import Flask`**
On importe la classe `Flask` du package `flask`. C'est la classe centrale qui représente votre application web.

**`app = Flask(__name__)`**
On crée une instance de Flask. `__name__` est une variable Python spéciale qui contient le nom du module courant. Quand vous lancez `python app.py`, `__name__` vaut `"__main__"`. Flask utilise cette information pour localiser les ressources de votre projet (templates, fichiers statiques).

**`@app.route("/")`**
C'est un **décorateur** Python. Il dit à Flask : "Quand quelqu'un fait une requête HTTP GET sur l'URL `/`, appelle la fonction qui suit." On appelle cela enregistrer une route.

**`def hello():`**
La fonction **view** (ou vue). Elle reçoit la requête et retourne la réponse. Ici, on retourne une simple chaîne de caractères.

**`app.run(debug=True)`**
Lance le serveur de développement intégré de Flask (Werkzeug). En production, on n'utilise JAMAIS cette commande — on utilise Gunicorn ou uWSGI.

---

## 2.3 Lancement et première visite

```bash
# Dans votre terminal (venv activé)
python app.py
```

Vous verrez :
```
 * Serving Flask app 'app'
 * Debug mode: on
WARNING: This is a development server. Do not use it in production.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
```

Ouvrez votre navigateur sur `http://127.0.0.1:5000` — vous verrez "Hello, UrbanPulse ! [CITYSCAPE]".

### Variables d'environnement Flask

Flask peut aussi être lancé avec la commande `flask` :

```bash
# Définir le fichier principal
export FLASK_APP=app.py     # Linux/macOS
set FLASK_APP=app.py        # Windows CMD
$env:FLASK_APP="app.py"     # Windows PowerShell

# Activer le mode debug
export FLASK_DEBUG=1

# Lancer
flask run
```

---

## 2.4 Structure professionnelle du projet UrbanPulse

Pour un projet réel, on ne met pas tout dans `app.py`. Voici la structure que nous allons construire progressivement :

```
urbanpulse/
├── app/                        # Package principal de l'application
│   ├── __init__.py             # Factory function de l'app Flask
│   ├── config.py               # Configuration (dev, prod, test)
│   ├── models/                 # Modèles de base de données
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── incident.py
│   ├── routes/                 # Routes organisées par Blueprint
│   │   ├── __init__.py
│   │   ├── main.py
│   │   ├── incidents.py
│   │   └── auth.py
│   ├── templates/              # Templates Jinja2
│   │   ├── base.html
│   │   ├── incidents/
│   │   └── auth/
│   └── static/                 # CSS, JS, images
│       ├── css/
│       ├── js/
│       └── uploads/
├── migrations/                 # Migrations Alembic (générées auto)
├── tests/                      # Tests automatisés
├── venv/                       # Environnement virtuel (jamais dans git)
├── .env                        # Variables d'environnement (jamais dans git)
├── .gitignore
├── requirements.txt
└── run.py                      # Point d'entrée de l'application
```

---

## 2.5 Le pattern Application Factory

Au lieu de créer `app` directement au niveau du module, on utilise une **factory function** — une fonction qui crée et configure l'application :

```python
# app/__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

# Extensions déclarées globalement mais pas encore liées à une app
db = SQLAlchemy()

def create_app(config_name="development"):
    """
    Application Factory Pattern.
    Crée et configure une instance Flask.
    
    Avantages :
    - Permet de créer plusieurs instances (utile pour les tests)
    - Évite les imports circulaires
    - Centralise la configuration
    """
    app = Flask(__name__)
    
    # Chargement de la configuration
    app.config.from_object(f"app.config.{config_name.capitalize()}Config")
    
    # Initialisation des extensions avec l'app
    db.init_app(app)
    
    # Enregistrement des blueprints (routes)
    from app.routes.main import main_bp
    app.register_blueprint(main_bp)
    
    return app
```

```python
# app/config.py

import os

class Config:
    """Configuration de base commune à tous les environnements."""
    SECRET_KEY = os.environ.get("SECRET_KEY") or "dev-secret-key-change-in-prod"
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(Config):
    """Configuration pour le développement."""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///urbanpulse_dev.db"

class ProductionConfig(Config):
    """Configuration pour la production."""
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL")

class TestingConfig(Config):
    """Configuration pour les tests."""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
```

```python
# run.py — Point d'entrée de l'application

from app import create_app

app = create_app("development")

if __name__ == "__main__":
    app.run()
```

---

## [EDIT] Exercice 2.1 — Mise en place du projet

> **Objectif** : Créer la structure de base du projet UrbanPulse.

1. Créez le dossier `urbanpulse` et naviguez dedans.
2. Créez et activez un environnement virtuel.
3. Installez Flask et créez `requirements.txt`.
4. Créez l'arborescence de fichiers décrite ci-dessus (les fichiers peuvent être vides pour l'instant).
5. Créez un fichier `app.py` minimal et vérifiez que `flask run` affiche "Hello, UrbanPulse !".
6. Modifiez votre application pour afficher le texte : `"UrbanPulse — La voix citoyenne de votre ville"`.

## [EDIT] Exercice 2.2 — Exploration du Debug Mode

> **Objectif** : Comprendre le rechargement automatique.

1. Lancez votre app avec `debug=True`.
2. Visitez `http://127.0.0.1:5000` dans le navigateur.
3. Sans arrêter le serveur, modifiez le texte retourné par `hello()`.
4. Rafraîchissez le navigateur — le changement est-il visible immédiatement ?
5. Maintenant introduisez une erreur Python dans le code (ex: `return 1/0`).
6. Visitez la page — que se passe-t-il ? Explorez le debugger interactif.

---

<a name="chapitre-3"></a>
# [GUIDE] Chapitre 3 — Concepts de Base

## 3.1 Routes et méthodes HTTP

### Le protocole HTTP en bref

HTTP (HyperText Transfer Protocol) est la langue que parlent les navigateurs et les serveurs. Chaque échange HTTP est une **requête** (client -> serveur) suivie d'une **réponse** (serveur -> client).

Les **méthodes HTTP** (ou "verbes") indiquent l'intention de la requête :

| Méthode | Sémantique | Exemple |
|---|---|---|
| **GET** | Récupérer une ressource | Afficher la liste des incidents |
| **POST** | Créer une nouvelle ressource | Soumettre un nouveau signalement |
| **PUT** | Remplacer complètement une ressource | Mettre à jour un incident entier |
| **PATCH** | Modifier partiellement une ressource | Changer le statut d'un incident |
| **DELETE** | Supprimer une ressource | Supprimer un commentaire |

### Routes Flask

```python
from flask import Flask, request, jsonify

app = Flask(__name__)

# Route simple GET (par défaut)
@app.route("/")
def index():
    return "Page d'accueil"

# Route avec méthodes explicites
@app.route("/incidents", methods=["GET", "POST"])
def incidents():
    if request.method == "GET":
        # Logique pour récupérer les incidents
        return "Liste des incidents"
    elif request.method == "POST":
        # Logique pour créer un incident
        return "Incident créé", 201  # 201 = Created

# Routes REST séparées (style Flask-MethodView)
@app.route("/incidents/<int:incident_id>", methods=["GET"])
def get_incident(incident_id):
    return f"Détail de l'incident #{incident_id}"

@app.route("/incidents/<int:incident_id>", methods=["PUT"])
def update_incident(incident_id):
    return f"Incident #{incident_id} mis à jour"

@app.route("/incidents/<int:incident_id>", methods=["DELETE"])
def delete_incident(incident_id):
    return f"Incident #{incident_id} supprimé", 204  # 204 = No Content
```

---

## 3.2 Paramètres d'URL et paramètres de requête

### Paramètres d'URL (URL Parameters)

Ce sont des parties **dynamiques** de l'URL, définies avec `<type:nom>` dans la route.

```python
# Paramètre simple (type string par défaut)
@app.route("/users/<username>")
def user_profile(username):
    return f"Profil de {username}"

# Paramètre de type entier
@app.route("/incidents/<int:incident_id>")
def incident_detail(incident_id):
    # incident_id est automatiquement converti en int
    return f"Incident numéro {incident_id}"

# Paramètre de type float
@app.route("/coordinates/<float:lat>/<float:lng>")
def map_point(lat, lng):
    return f"Coordonnées : {lat}, {lng}"

# Paramètre "path" accepte les slashes
@app.route("/files/<path:filepath>")
def serve_file(filepath):
    return f"Fichier : {filepath}"

# Convertisseurs disponibles :
# string : (défaut) toute chaîne sans slash
# int    : entiers positifs
# float  : nombres à virgule
# path   : comme string mais accepte les slashes
# uuid   : identifiants UUID
```

### Paramètres de requête (Query Parameters)

Ce sont des paramètres après `?` dans l'URL : `http://example.com/incidents?page=2&status=open`

```python
@app.route("/incidents")
def list_incidents():
    # Récupération des paramètres de requête
    page = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", 10, type=int)
    status = request.args.get("status", "all")  # "all" est la valeur par défaut
    category = request.args.get("category")  # None si absent
    
    # Exemple : /incidents?page=2&per_page=5&status=open&category=voirie
    
    return jsonify({
        "page": page,
        "per_page": per_page,
        "status": status,
        "category": category
    })
```

---

## 3.3 L'objet `request`

Flask met à disposition un objet `request` qui contient toutes les informations sur la requête HTTP entrante.

```python
from flask import request

@app.route("/incidents", methods=["POST"])
def create_incident():
    # Corps de la requête JSON
    data = request.json          # dict Python (si Content-Type: application/json)
    data = request.get_json()    # Plus robuste, retourne None si pas du JSON
    
    # Données de formulaire HTML
    title = request.form.get("title")
    description = request.form.get("description")
    
    # Fichiers uploadés
    photo = request.files.get("photo")
    
    # Headers HTTP
    auth_header = request.headers.get("Authorization")
    user_agent = request.headers.get("User-Agent")
    
    # Méthode HTTP
    method = request.method       # "GET", "POST", etc.
    
    # URL et chemin
    full_url = request.url        # "http://localhost:5000/incidents"
    path = request.path           # "/incidents"
    
    # Adresse IP du client
    client_ip = request.remote_addr
    
    return "Requête analysée"
```

---

## 3.4 L'objet `response` et codes HTTP

Flask génère automatiquement une réponse HTTP à partir de ce que retourne votre fonction view.

```python
from flask import Flask, jsonify, make_response, redirect, url_for

# Retourner une chaîne simple (code 200 par défaut)
return "Hello World"

# Retourner avec un code HTTP spécifique
return "Created", 201
return "Not Found", 404

# Retourner du JSON
return jsonify({"message": "Succès", "id": 42})

# Retourner un objet Response complet (plus de contrôle)
response = make_response(jsonify({"error": "Non autorisé"}), 401)
response.headers["X-Custom-Header"] = "UrbanPulse"
return response

# Redirection
return redirect("/incidents")
return redirect(url_for("incidents.list_incidents"))  # Avec url_for

# Codes HTTP importants à connaître
# 200 OK           — Succès
# 201 Created      — Ressource créée
# 204 No Content   — Succès sans corps de réponse (DELETE)
# 400 Bad Request  — Requête malformée
# 401 Unauthorized — Non authentifié
# 403 Forbidden    — Authentifié mais pas autorisé
# 404 Not Found    — Ressource inexistante
# 422 Unprocessable — Données invalides (validation échouée)
# 500 Internal Error — Erreur serveur
```

---

## 3.5 Templates Jinja2

Au lieu de retourner du HTML brut dans Python (horrible à maintenir), Flask utilise **Jinja2** — un moteur de templates qui sépare la logique de la présentation.

### Structure des templates

```
app/templates/
├── base.html           <- Template parent (layout commun)
├── index.html          <- Page d'accueil
└── incidents/
    ├── list.html       <- Liste des incidents
    ├── detail.html     <- Détail d'un incident
    └── create.html     <- Formulaire de création
```

### Template de base (héritage)

```html
<!-- app/templates/base.html -->
<!DOCTYPE html>
<html lang="fr">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <!-- block = zone que les enfants peuvent remplacer -->
    <title>{% block title %}UrbanPulse{% endblock %}</title>
    <link rel="stylesheet" href="{{ url_for('static', filename='css/main.css') }}">
    {% block extra_css %}{% endblock %}
</head>
<body>
    <!-- Navigation commune à toutes les pages -->
    <nav>
        <a href="{{ url_for('main.index') }}">[CITYSCAPE] UrbanPulse</a>
        <a href="{{ url_for('incidents.list') }}">Signalements</a>
        {% if current_user.is_authenticated %}
            <a href="{{ url_for('incidents.create') }}">+ Signaler</a>
            <a href="{{ url_for('auth.logout') }}">Déconnexion</a>
        {% else %}
            <a href="{{ url_for('auth.login') }}">Connexion</a>
        {% endif %}
    </nav>

    <!-- Messages flash (notifications) -->
    {% with messages = get_flashed_messages(with_categories=true) %}
        {% if messages %}
            {% for category, message in messages %}
                <div class="alert alert-{{ category }}">{{ message }}</div>
            {% endfor %}
        {% endif %}
    {% endwith %}

    <!-- Contenu principal — les enfants remplissent ce bloc -->
    <main>
        {% block content %}{% endblock %}
    </main>

    <footer>
        <p>© 2024 UrbanPulse — Ensemble pour une ville meilleure</p>
    </footer>
    
    <script src="{{ url_for('static', filename='js/main.js') }}"></script>
    {% block extra_js %}{% endblock %}
</body>
</html>
```

### Template enfant

```html
<!-- app/templates/incidents/list.html -->

{# On hérite du template parent #}
{% extends "base.html" %}

{# On redéfinit le titre #}
{% block title %}Signalements — UrbanPulse{% endblock %}

{# On remplit le contenu principal #}
{% block content %}
<div class="container">
    <h1>Signalements citoyens</h1>
    
    {# Boucle sur la liste des incidents #}
    {% for incident in incidents %}
        <div class="incident-card incident-{{ incident.status }}">
            <h2>{{ incident.title }}</h2>
            <p>{{ incident.description | truncate(150) }}</p>
            
            {# Affichage conditionnel #}
            {% if incident.status == "resolved" %}
                <span class="badge green">[OK] Résolu</span>
            {% elif incident.status == "in_progress" %}
                <span class="badge orange">[OUTIL] En cours</span>
            {% else %}
                <span class="badge red">[HOURGLASS_WITH_FLOWING_SAND] En attente</span>
            {% endif %}
            
            <p>
                Signalé par <strong>{{ incident.author.username }}</strong>
                le {{ incident.created_at | strftime("%d/%m/%Y") }}
            </p>
            
            <a href="{{ url_for('incidents.detail', id=incident.id) }}">
                Voir le détail ->
            </a>
        </div>
    {% else %}
        {# Affiché si la liste est vide #}
        <p>Aucun signalement pour le moment. Soyez le premier !</p>
    {% endfor %}
    
    {# Pagination #}
    <div class="pagination">
        {% if pagination.has_prev %}
            <a href="?page={{ pagination.prev_num }}"><- Précédent</a>
        {% endif %}
        <span>Page {{ pagination.page }} / {{ pagination.pages }}</span>
        {% if pagination.has_next %}
            <a href="?page={{ pagination.next_num }}">Suivant -></a>
        {% endif %}
    </div>
</div>
{% endblock %}
```

### Syntaxe Jinja2 résumée

```jinja
{# Commentaire (non affiché dans le HTML) #}

{# Variables — affichage #}
{{ variable }}
{{ user.name }}
{{ incident.title | upper }}    {# Filtre : met en majuscules #}
{{ text | truncate(100) }}      {# Filtre : tronque à 100 caractères #}
{{ price | round(2) }}          {# Filtre : arrondit #}
{{ html_content | safe }}       {# Filtre : désactive l'échappement HTML #}

{# Structures de contrôle #}
{% if condition %}
    ...
{% elif autre_condition %}
    ...
{% else %}
    ...
{% endif %}

{% for item in liste %}
    {{ loop.index }}     {# Index (commence à 1) #}
    {{ loop.index0 }}    {# Index (commence à 0) #}
    {{ loop.first }}     {# True si premier élément #}
    {{ loop.last }}      {# True si dernier élément #}
    {{ item }}
{% else %}
    Rien à afficher
{% endfor %}

{# Inclusion d'un autre template #}
{% include "components/navbar.html" %}

{# Héritage #}
{% extends "base.html" %}
{% block nom %}contenu{% endblock %}

{# Macro (comme une fonction réutilisable) #}
{% macro render_incident(incident) %}
    <div class="card">{{ incident.title }}</div>
{% endmacro %}

{# Appel de la macro #}
{{ render_incident(incident) }}
```

---

## 3.6 Rendu des templates depuis les routes

```python
from flask import Flask, render_template

app = Flask(__name__)

@app.route("/")
def index():
    # render_template cherche le fichier dans app/templates/
    return render_template("index.html")

@app.route("/incidents")
def list_incidents():
    # On passe des données au template
    # Les incidents viendraient normalement de la base de données
    incidents = [
        {"id": 1, "title": "Nid-de-poule rue Victor Hugo", "status": "open"},
        {"id": 2, "title": "Lampadaire cassé place de la Mairie", "status": "in_progress"},
    ]
    return render_template(
        "incidents/list.html",
        incidents=incidents,     # Disponible comme {{ incidents }} dans le template
        page_title="Signalements"
    )
```

---

## 3.7 Fichiers statiques

Les fichiers statiques (CSS, JavaScript, images) se placent dans `app/static/`.

```python
# Dans les routes Python :
from flask import url_for

url_for("static", filename="css/main.css")
# -> "/static/css/main.css"

url_for("static", filename="js/app.js")
# -> "/static/js/app.js"

url_for("static", filename="uploads/photo.jpg")
# -> "/static/uploads/photo.jpg"
```

```html
<!-- Dans les templates Jinja2 : -->
<link rel="stylesheet" href="{{ url_for('static', filename='css/main.css') }}">
<script src="{{ url_for('static', filename='js/main.js') }}"></script>
<img src="{{ url_for('static', filename='img/logo.png') }}" alt="Logo">
```

**[ATTENTION] Toujours utiliser `url_for` plutôt que des chemins hardcodés** — si vous déployez Flask sous un sous-chemin (`/app/`), les chemins absolus cassent mais `url_for` s'adapte automatiquement.

---

## 3.8 UrbanPulse — Premières routes

```python
# app/routes/main.py

from flask import Blueprint, render_template

# Blueprint = mini-application Flask avec son propre espace de noms
main_bp = Blueprint("main", __name__)

@main_bp.route("/")
def index():
    """Page d'accueil d'UrbanPulse."""
    # Statistiques fictives pour l'instant
    stats = {
        "total_incidents": 142,
        "resolved_this_month": 38,
        "active_users": 891
    }
    return render_template("index.html", stats=stats)

@main_bp.route("/about")
def about():
    """Page de présentation du projet."""
    return render_template("about.html")
```

---

## [EDIT] Exercice 3.1 — Premières routes UrbanPulse

> **Objectif** : Créer les routes et templates de base.

1. Créez les routes suivantes pour UrbanPulse :
   - `GET /` -> Page d'accueil
   - `GET /incidents` -> Liste fictive de 5 incidents (données codées en dur)
   - `GET /incidents/<int:id>` -> Détail d'un incident (retourner un JSON pour l'instant)
   - `GET /incidents?status=open` -> Même route, mais filtrer selon le paramètre

2. Créez `base.html` et `incidents/list.html` en utilisant l'héritage Jinja2.

3. Dans `list.html`, affichez les 5 incidents en utilisant une boucle `{% for %}`.

4. Ajoutez un message "Aucun signalement" si la liste est vide (testez en passant une liste vide).

## [EDIT] Exercice 3.2 — Exploration HTTP

> **Objectif** : Comprendre les requêtes HTTP en pratique.

1. Installez l'outil **HTTPie** (`pip install httpie`) ou utilisez **curl**.
2. Faites une requête GET sur votre route `/incidents` : `http GET localhost:5000/incidents`
3. Faites une requête POST (même si la route ne le supporte pas encore) : que se passe-t-il ?
4. Quelle est la différence entre `/incidents?page=2` et `/incidents/2` ? Quand utiliser l'un ou l'autre ?

---

<a name="chapitre-4"></a>
# [GUIDE] Chapitre 4 — Introduction aux ORM et SQLAlchemy

## 4.1 Qu'est-ce qu'un ORM ?

Un **ORM (Object-Relational Mapper)** est une bibliothèque qui fait le pont entre le monde des objets Python et le monde des bases de données relationnelles (SQL).

**Sans ORM :**
```python
import sqlite3
conn = sqlite3.connect("urbanpulse.db")
cursor = conn.cursor()
cursor.execute("""
    SELECT id, title, description, status 
    FROM incidents 
    WHERE status = ? AND created_at > ?
""", ("open", "2024-01-01"))
rows = cursor.fetchall()
# rows = [(1, "Nid-de-poule", "...", "open"), (2, ...)]
# Pas d'objet Python, juste des tuples
```

**Avec SQLAlchemy ORM :**
```python
incidents = Incident.query.filter(
    Incident.status == "open",
    Incident.created_at > datetime(2024, 1, 1)
).all()
# incidents = [<Incident 1>, <Incident 2>]
# Ce sont des objets Python avec des attributs !
incident = incidents[0]
print(incident.title)    # "Nid-de-poule"
print(incident.author.username)  # Accès aux relations !
```

**Avantages de l'ORM :**
- Code Python natif, plus lisible et maintenable
- Protection automatique contre l'injection SQL
- Portabilité (changez de SQLite à PostgreSQL sans modifier les requêtes)
- Gestion des relations entre tables (jointures automatiques)
- Migrations de schéma avec Alembic

**Inconvénients :**
- Légère perte de performance sur les requêtes très complexes
- Peut générer des requêtes SQL non optimales si mal utilisé
- Courbe d'apprentissage initiale

---

## 4.2 Installation et configuration Flask-SQLAlchemy

```bash
pip install flask-sqlalchemy
pip freeze > requirements.txt
```

```python
# app/__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

# Déclaration globale — sera initialisée plus tard avec l'app
db = SQLAlchemy()

def create_app(config_name="development"):
    app = Flask(__name__)
    
    # Configuration de la base de données
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///urbanpulse.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    # Option utile en développement pour voir les requêtes SQL générées :
    app.config["SQLALCHEMY_ECHO"] = True  # Affiche les requêtes SQL dans le terminal
    
    # Liaison de db avec l'application
    db.init_app(app)
    
    return app
```

### Types de bases de données supportées

```python
# SQLite (développement — pas de serveur à installer)
SQLALCHEMY_DATABASE_URI = "sqlite:///urbanpulse.db"
SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"  # En mémoire (pour les tests)

# PostgreSQL (recommandé pour la production)
SQLALCHEMY_DATABASE_URI = "postgresql://user:password@localhost/urbanpulse"

# MySQL/MariaDB
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://user:password@localhost/urbanpulse"
```

---

## 4.3 Définition des modèles

Un **modèle** est une classe Python qui représente une table de base de données. Chaque instance du modèle est une ligne de la table.

```python
# app/models/user.py

from datetime import datetime
from app import db  # Import de l'instance db créée dans __init__.py

class User(db.Model):
    """
    Modèle représentant un utilisateur de UrbanPulse.
    Table SQL correspondante : 'user'
    """
    # Nom de table explicite (optionnel, Flask génère "user" automatiquement)
    __tablename__ = "users"
    
    # ==================== COLONNES ====================
    
    # Clé primaire — identifiant unique, auto-incrémenté
    id = db.Column(db.Integer, primary_key=True)
    
    # Email — unique et requis
    email = db.Column(
        db.String(120),
        unique=True,         # Pas deux utilisateurs avec le même email
        nullable=False,      # Ne peut pas être NULL
        index=True           # Index pour accélérer les recherches par email
    )
    
    # Nom d'utilisateur — unique et requis
    username = db.Column(
        db.String(80),
        unique=True,
        nullable=False
    )
    
    # Mot de passe hashé (JAMAIS le mot de passe en clair !)
    password_hash = db.Column(db.String(256), nullable=False)
    
    # Rôle de l'utilisateur
    role = db.Column(
        db.String(20),
        nullable=False,
        default="citizen"    # Valeur par défaut : citoyen
        # Valeurs possibles : "citizen", "moderator", "admin"
    )
    
    # Statut du compte
    is_active = db.Column(db.Boolean, default=True, nullable=False)
    
    # Timestamps automatiques
    created_at = db.Column(
        db.DateTime,
        default=datetime.utcnow,  # Valeur par défaut côté Python
        nullable=False
    )
    updated_at = db.Column(
        db.DateTime,
        default=datetime.utcnow,
        onupdate=datetime.utcnow  # Mis à jour automatiquement à chaque modification
    )
    
    # ==================== RELATIONS (défini dans chapitre 6) ====================
    # incidents = db.relationship("Incident", backref="author", lazy="dynamic")
    
    # ==================== MÉTHODES ====================
    
    def set_password(self, password):
        """Hash et stocke le mot de passe (utilise werkzeug)."""
        from werkzeug.security import generate_password_hash
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        """Vérifie si le mot de passe fourni correspond au hash stocké."""
        from werkzeug.security import check_password_hash
        return check_password_hash(self.password_hash, password)
    
    def to_dict(self):
        """Sérialise l'utilisateur en dictionnaire (pour les APIs JSON)."""
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "role": self.role,
            "created_at": self.created_at.isoformat()
            # Note : on n'inclut JAMAIS password_hash dans to_dict !
        }
    
    def __repr__(self):
        """Représentation textuelle pour le debugging."""
        return f"<User {self.username} ({self.role})>"
```

```python
# app/models/incident.py

from datetime import datetime
from app import db

class Incident(db.Model):
    """
    Modèle représentant un signalement citoyen.
    Table SQL correspondante : 'incidents'
    """
    __tablename__ = "incidents"
    
    # Statuts possibles (constantes de classe)
    STATUS_OPEN = "open"
    STATUS_IN_PROGRESS = "in_progress"
    STATUS_RESOLVED = "resolved"
    STATUS_REJECTED = "rejected"
    
    VALID_STATUSES = [STATUS_OPEN, STATUS_IN_PROGRESS, STATUS_RESOLVED, STATUS_REJECTED]
    
    # Catégories possibles
    CATEGORY_ROAD = "voirie"
    CATEGORY_LIGHTING = "eclairage"
    CATEGORY_WASTE = "dechets"
    CATEGORY_GREENSPACE = "espaces_verts"
    CATEGORY_OTHER = "autre"
    
    # ==================== COLONNES ====================
    
    id = db.Column(db.Integer, primary_key=True)
    
    title = db.Column(db.String(200), nullable=False)
    
    description = db.Column(db.Text, nullable=False)
    
    category = db.Column(
        db.String(50),
        nullable=False,
        default=CATEGORY_OTHER
    )
    
    status = db.Column(
        db.String(20),
        nullable=False,
        default=STATUS_OPEN
    )
    
    # Adresse textuelle du signalement
    address = db.Column(db.String(300))
    
    # Coordonnées GPS (optionnelles)
    latitude = db.Column(db.Float)
    longitude = db.Column(db.Float)
    
    # Nombre de votes (upvotes citoyens pour prioriser)
    vote_count = db.Column(db.Integer, default=0, nullable=False)
    
    # Photo du problème (chemin vers le fichier uploadé)
    photo_url = db.Column(db.String(500))
    
    # Clé étrangère vers l'utilisateur qui a créé le signalement
    author_id = db.Column(
        db.Integer,
        db.ForeignKey("users.id"),  # Référence la colonne "id" de la table "users"
        nullable=False
    )
    
    # Timestamps
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    resolved_at = db.Column(db.DateTime)  # Rempli quand le statut passe à "resolved"
    
    # ==================== MÉTHODES ====================
    
    def resolve(self):
        """Marque l'incident comme résolu."""
        self.status = self.STATUS_RESOLVED
        self.resolved_at = datetime.utcnow()
    
    def is_open(self):
        """Retourne True si l'incident est encore ouvert."""
        return self.status == self.STATUS_OPEN
    
    def upvote(self):
        """Incrémente le compteur de votes."""
        self.vote_count += 1
    
    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "category": self.category,
            "status": self.status,
            "address": self.address,
            "vote_count": self.vote_count,
            "author": self.author.username if self.author else None,
            "created_at": self.created_at.isoformat()
        }
    
    def __repr__(self):
        return f"<Incident #{self.id} '{self.title}' [{self.status}]>"
```

---

## 4.4 Types de colonnes SQLAlchemy

```python
# Types courants
db.Column(db.Integer)           # Entier — clés primaires, compteurs
db.Column(db.BigInteger)        # Grand entier
db.Column(db.String(n))         # VARCHAR(n) — chaîne de longueur max n
db.Column(db.Text)              # TEXT — chaîne longue sans limite
db.Column(db.Boolean)           # Booléen
db.Column(db.Float)             # Nombre à virgule flottante
db.Column(db.Numeric(10, 2))    # Nombre décimal précis (montants)
db.Column(db.DateTime)          # Date et heure
db.Column(db.Date)              # Date seulement
db.Column(db.JSON)              # JSON (PostgreSQL natif, SQLite émulé)
db.Column(db.Enum("val1", "val2", name="enum_name"))  # Énumération
```

---

## 4.5 CRUD — Opérations de base

**CRUD = Create, Read, Update, Delete** — les 4 opérations fondamentales sur les données.

```python
# ============================================================
# CREATE — Créer des enregistrements
# ============================================================

# Créer un nouvel utilisateur
user = User(
    email="alice@example.com",
    username="alice_martin",
    role="citizen"
)
user.set_password("MotDePasse123!")

db.session.add(user)      # Ajoute à la "session" (pas encore en base)
db.session.commit()       # Confirme et écrit en base de données

print(user.id)  # Maintenant l'ID est disponible car la ligne est créée

# Créer plusieurs objets en une fois
users = [
    User(email="bob@ex.com", username="bob", role="citizen"),
    User(email="carol@ex.com", username="carol", role="moderator"),
]
for u in users:
    u.set_password("pass123")
db.session.add_all(users)
db.session.commit()

# ============================================================
# READ — Lire des enregistrements
# ============================================================

# Récupérer par clé primaire (retourne None si inexistant)
user = User.query.get(1)
# Ou plus explicite :
user = db.session.get(User, 1)  # Style SQLAlchemy 2.x

# Récupérer tous
all_users = User.query.all()  # Liste de tous les User

# Avec filtre
alice = User.query.filter_by(username="alice_martin").first()
alice = User.query.filter(User.username == "alice_martin").first()

# Plusieurs conditions
open_road_incidents = Incident.query.filter(
    Incident.status == "open",
    Incident.category == "voirie"
).all()

# Opérateurs de comparaison
Incident.query.filter(Incident.vote_count >= 10).all()
Incident.query.filter(Incident.title.contains("nid")).all()
Incident.query.filter(Incident.title.ilike("%nid%")).all()  # Insensible à la casse

# Tri
Incident.query.order_by(Incident.vote_count.desc()).all()
Incident.query.order_by(Incident.created_at.asc()).all()

# Limiter et paginer
Incident.query.limit(10).all()
Incident.query.offset(20).limit(10).all()  # Page 3 avec 10 par page

# Pagination avec Flask-SQLAlchemy
pagination = Incident.query.paginate(page=2, per_page=10)
incidents = pagination.items    # Les incidents de cette page
total = pagination.total        # Nombre total d'incidents

# Compter
count = Incident.query.filter_by(status="open").count()

# ============================================================
# UPDATE — Mettre à jour des enregistrements
# ============================================================

# Modifier un objet existant
incident = Incident.query.get(1)
incident.status = "in_progress"
incident.updated_at = datetime.utcnow()
db.session.commit()

# Mise à jour en masse (plus efficace)
Incident.query.filter_by(status="open").update({
    "status": "in_progress"
})
db.session.commit()

# ============================================================
# DELETE — Supprimer des enregistrements
# ============================================================

# Supprimer un objet spécifique
incident = Incident.query.get(1)
db.session.delete(incident)
db.session.commit()

# Suppression en masse
Incident.query.filter_by(status="rejected").delete()
db.session.commit()
```

---

## 4.6 Gestion des sessions et transactions

```python
# La "session" SQLAlchemy est comme un panier de modifications
# en attente d'être confirmées (commit) ou annulées (rollback)

from flask import jsonify
from sqlalchemy.exc import IntegrityError

def create_incident(data):
    try:
        incident = Incident(
            title=data["title"],
            description=data["description"],
            author_id=data["author_id"]
        )
        db.session.add(incident)
        db.session.commit()
        return incident
        
    except IntegrityError:
        # Erreur de contrainte (ex: clé étrangère invalide, unicité violée)
        db.session.rollback()  # Annule toutes les modifications en attente
        raise ValueError("Données invalides")
        
    except Exception as e:
        db.session.rollback()
        raise e
```

---

## [EDIT] Exercice 4.1 — Définir les modèles UrbanPulse

> **Objectif** : Créer et utiliser les modèles de base.

1. Créez le fichier `app/models/user.py` avec le modèle `User` complet.
2. Créez le fichier `app/models/incident.py` avec le modèle `Incident`.
3. Créez un fichier `app/models/__init__.py` qui importe les deux modèles.
4. Dans un script Python interactif (`flask shell`), créez les tables : `db.create_all()`
5. Créez 3 utilisateurs et 5 incidents avec des données fictives.
6. Testez toutes les opérations CRUD : récupérer tous les incidents, filtrer par statut, mettre à jour, supprimer.

## [EDIT] Exercice 4.2 — Flask Shell

> **Objectif** : Maîtriser l'exploration interactive.

```bash
flask shell
```

1. Importez vos modèles : `from app.models.user import User`
2. Créez un utilisateur et sauvegardez-le.
3. Récupérez-le par username.
4. Modifiez son rôle et committez.
5. Comptez le nombre total d'utilisateurs.
6. **Bonus** : Installez `flask-shell-ipython` pour un shell plus agréable.

---

<a name="chapitre-5"></a>
# [GUIDE] Chapitre 5 — Migrations de Base de Données

## 5.1 Pourquoi les migrations ?

Imaginez que votre application est en production avec 10 000 utilisateurs. Vous réalisez que vous devez ajouter une colonne `phone_number` à la table `users`. Que faites-vous ?

**Sans migrations :** Vous modifiez le modèle Python, mais la base de données en production ne change pas. L'application plante.

**Avec migrations (Alembic/Flask-Migrate) :** Vous générez un script de migration qui modifie la structure de la base de données sans perdre les données existantes.

Une migration = **un script versionné** qui décrit comment passer d'une version du schéma à la suivante (et comment revenir en arrière).

---

## 5.2 Installation et initialisation

```bash
pip install flask-migrate
pip freeze > requirements.txt
```

```python
# app/__init__.py — Ajout de Flask-Migrate

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

def create_app(config_name="development"):
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///urbanpulse.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    
    db.init_app(app)
    migrate.init_app(app, db)  # Lie Migrate avec app et db
    
    return app
```

### Initialisation du dossier de migrations (une seule fois)

```bash
flask db init
```

Cela crée le dossier `migrations/` avec :
```
migrations/
├── alembic.ini
├── env.py          <- Configuration d'Alembic
├── README
├── script.py.mako  <- Template pour les scripts de migration
└── versions/       <- Tous vos fichiers de migration
```

---

## 5.3 Workflow complet des migrations

```bash
# ÉTAPE 1 : Vous avez modifié ou créé des modèles Python

# ÉTAPE 2 : Générer une migration automatiquement
# Flask-Migrate compare vos modèles avec l'état actuel de la base
flask db migrate -m "Ajout table incidents et users"
# Crée un fichier dans migrations/versions/xxx_ajout_table_incidents.py

# ÉTAPE 3 : Vérifiez le fichier généré !
# (Très important — Alembic n'est pas parfait, vérifiez toujours)

# ÉTAPE 4 : Appliquer la migration
flask db upgrade
# La base de données est maintenant mise à jour

# Autres commandes utiles
flask db downgrade          # Revenir à la migration précédente
flask db downgrade -2       # Revenir 2 versions en arrière
flask db current            # Afficher la version courante
flask db history            # Voir l'historique des migrations
flask db heads              # Voir la dernière migration
```

---

## 5.4 Anatomie d'un fichier de migration

```python
# migrations/versions/20240115_1043_ajout_table_users_et_incidents.py

"""Ajout table users et incidents

Revision ID: a1b2c3d4e5f6
Revises: 
Create Date: 2024-01-15 10:43:21.123456
"""

from alembic import op
import sqlalchemy as sa

# Identifiant unique de cette migration
revision = 'a1b2c3d4e5f6'
# Identifiant de la migration précédente (None = première migration)
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    """Ce que fait la migration (vers l'avant)."""
    # Création de la table users
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(length=120), nullable=False),
        sa.Column('username', sa.String(length=80), nullable=False),
        sa.Column('password_hash', sa.String(length=256), nullable=False),
        sa.Column('role', sa.String(length=20), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.Column('updated_at', sa.DateTime()),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
        sa.UniqueConstraint('username')
    )
    
    # Création de la table incidents
    op.create_table(
        'incidents',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(length=200), nullable=False),
        sa.Column('description', sa.Text(), nullable=False),
        sa.Column('status', sa.String(length=20), nullable=False),
        sa.Column('author_id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['author_id'], ['users.id']),
        sa.PrimaryKeyConstraint('id')
    )


def downgrade():
    """Comment annuler cette migration (vers l'arrière)."""
    op.drop_table('incidents')
    op.drop_table('users')
```

### Exemple de migration d'ajout de colonne

```python
# Une migration typique : ajouter une colonne à une table existante

def upgrade():
    # Ajouter la colonne phone_number à la table users
    op.add_column(
        'users',
        sa.Column('phone_number', sa.String(length=20), nullable=True)
    )
    # nullable=True est important ! Si False, il faut une valeur par défaut
    # pour les lignes existantes.

def downgrade():
    op.drop_column('users', 'phone_number')
```

---

## 5.5 Cas complexes de migration

```python
# Renommer une colonne
def upgrade():
    op.alter_column('incidents', 'vote_count', new_column_name='upvotes')

def downgrade():
    op.alter_column('incidents', 'upvotes', new_column_name='vote_count')

# Ajouter un index
def upgrade():
    op.create_index('ix_incidents_status', 'incidents', ['status'])

def downgrade():
    op.drop_index('ix_incidents_status')

# Migration de données (pas juste de schéma)
def upgrade():
    # Ajouter une colonne
    op.add_column('users', sa.Column('full_name', sa.String(200)))
    
    # Peupler la colonne avec des données existantes
    op.execute("""
        UPDATE users 
        SET full_name = username
        WHERE full_name IS NULL
    """)
    
    # Rendre la colonne NOT NULL maintenant que toutes les lignes ont une valeur
    op.alter_column('users', 'full_name', nullable=False)
```

---

## 5.6 Bonnes pratiques de migration

**1. Toujours versionner les migrations dans Git**
```bash
git add migrations/
git commit -m "Migration: ajout colonne phone_number à users"
```

**2. Nommer les migrations clairement**
```bash
flask db migrate -m "Ajout phone_number à users"
# Pas : flask db migrate -m "fix"
```

**3. Tester localement avant la production**
```bash
# En local
flask db upgrade
# Testez que tout fonctionne
flask db downgrade  # Vérifiez que le rollback fonctionne aussi
flask db upgrade    # Repartez de l'avant

# En production seulement après validation
```

**4. Ne jamais modifier une migration déjà appliquée en production**
Créez toujours une nouvelle migration si vous devez corriger quelque chose.

**5. Script de déploiement**
```bash
#!/bin/bash
# deploy.sh
git pull origin main
pip install -r requirements.txt
flask db upgrade          # Toujours migrer avant de redémarrer l'app
gunicorn app:app --reload
```

---

## [EDIT] Exercice 5.1 — Cycle de vie d'une migration

> **Objectif** : Maîtriser le workflow migrations.

1. Initialisez Flask-Migrate dans votre projet.
2. Créez votre première migration : `flask db migrate -m "Initial migration"`
3. Ouvrez le fichier généré et vérifiez qu'il correspond à vos modèles.
4. Appliquez la migration : `flask db upgrade`
5. Vérifiez avec SQLite Browser (ou `flask shell`) que les tables existent.
6. Ajoutez une colonne `city = db.Column(db.String(100), default="Paris")` à `Incident`.
7. Créez et appliquez une nouvelle migration.
8. Testez le rollback avec `flask db downgrade`.

---

<a name="chapitre-6"></a>
# [GUIDE] Chapitre 6 — Relations entre Modèles

## 6.1 Concepts des relations en base de données

Les données réelles sont rarement isolées. Un incident appartient à un utilisateur. Un utilisateur peut avoir plusieurs incidents. Plusieurs utilisateurs peuvent voter pour le même incident.

Les trois types de relations :

| Relation | Exemple | Implémentation SQL |
|---|---|---|
| **One-to-Many** (1:N) | Un user a plusieurs incidents | Clé étrangère dans "plusieurs" |
| **Many-to-Many** (N:M) | Des users votent pour des incidents | Table de jointure |
| **One-to-One** (1:1) | Un user a un profil | Clé étrangère unique |

---

## 6.2 One-to-Many : User -> Incidents

```python
# app/models/user.py

class User(db.Model):
    __tablename__ = "users"
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    # ... autres colonnes ...
    
    # Relation : un User a plusieurs Incidents
    # backref="author" crée automatiquement Incident.author
    # lazy="dynamic" retourne une query plutôt qu'une liste (pour filtrer ensuite)
    incidents = db.relationship(
        "Incident",
        backref="author",       # Crée incident.author pour accéder à l'user
        lazy="dynamic",         # Permet : user.incidents.filter_by(status="open")
        cascade="all, delete-orphan"  # Supprime les incidents si l'user est supprimé
    )
    
    # Relation pour les commentaires
    comments = db.relationship(
        "Comment",
        backref="author",
        lazy="dynamic"
    )
```

```python
# app/models/incident.py

class Incident(db.Model):
    __tablename__ = "incidents"
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    
    # Clé étrangère — crée la relation SQL
    author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    
    # Grâce au backref="author" dans User.incidents :
    # incident.author -> retourne l'objet User correspondant (SANS requête supplémentaire si lazy=select)
    
    # Relation One-to-Many avec les commentaires
    comments = db.relationship(
        "Comment",
        backref="incident",
        lazy="dynamic",
        order_by="Comment.created_at.asc()"
    )
```

### Utilisation des relations

```python
# Créer un incident associé à un utilisateur
alice = User.query.filter_by(username="alice").first()
incident = Incident(
    title="Nid-de-poule dangereux",
    description="Rue Victor Hugo, à hauteur du n°45",
    category="voirie",
    author=alice   # Utiliser la relation directement (SQLAlchemy gère l'author_id)
    # Ou : author_id=alice.id
)
db.session.add(incident)
db.session.commit()

# Accéder aux incidents d'un utilisateur
alice = User.query.get(1)
alice.incidents.all()                              # Tous ses incidents
alice.incidents.filter_by(status="open").all()    # Seulement les ouverts
alice.incidents.count()                            # Combien d'incidents

# Accéder à l'auteur d'un incident
incident = Incident.query.get(1)
print(incident.author.username)   # "alice"
print(incident.author.email)      # "alice@example.com"
```

---

## 6.3 Many-to-Many : Users <-> Votes sur Incidents

Pour les votes, un utilisateur peut voter pour plusieurs incidents, et un incident peut recevoir des votes de plusieurs utilisateurs.

```python
# Table de jointure (association table)
# Simple : juste les deux clés étrangères
votes = db.Table(
    "votes",
    db.Column("user_id", db.Integer, db.ForeignKey("users.id"), primary_key=True),
    db.Column("incident_id", db.Integer, db.ForeignKey("incidents.id"), primary_key=True),
    db.Column("voted_at", db.DateTime, default=datetime.utcnow)  # Métadonnée du vote
)

# Dans le modèle User :
class User(db.Model):
    # ...
    voted_incidents = db.relationship(
        "Incident",
        secondary=votes,           # La table de jointure
        backref=db.backref("voters", lazy="dynamic"),  # Incident.voters = tous les votants
        lazy="dynamic"
    )
```

### Utilisation de la relation Many-to-Many

```python
# Voter pour un incident
alice = User.query.get(1)
incident = Incident.query.get(5)

# Vérifier si alice a déjà voté
has_voted = alice.voted_incidents.filter(
    Incident.id == incident.id
).first() is not None

if not has_voted:
    alice.voted_incidents.append(incident)
    incident.vote_count += 1
    db.session.commit()
    print("Vote enregistré !")
else:
    print("Alice a déjà voté pour cet incident")

# Voir tous les incidents pour lesquels alice a voté
alice.voted_incidents.all()

# Voir tous les utilisateurs qui ont voté pour cet incident
incident.voters.all()
```

---

## 6.4 Association Object Pattern (Plus riche que many-to-many simple)

Quand la table de jointure doit contenir plus qu'un simple lien (ex: timestamp, raison du vote...) :

```python
# app/models/vote.py

class Vote(db.Model):
    """
    Table d'association entre User et Incident pour les votes.
    Utilise le pattern "association object" pour stocker des données supplémentaires.
    """
    __tablename__ = "votes"
    
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), primary_key=True)
    incident_id = db.Column(db.Integer, db.ForeignKey("incidents.id"), primary_key=True)
    voted_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    # Relations vers les deux côtés
    user = db.relationship("User", backref="votes")
    incident = db.relationship("Incident", backref="votes")
    
    def __repr__(self):
        return f"<Vote user={self.user_id} incident={self.incident_id}>"
```

```python
# Usage
vote = Vote(user_id=alice.id, incident_id=incident.id)
db.session.add(vote)
incident.vote_count += 1
db.session.commit()

# Vérifier si alice a voté
existing_vote = Vote.query.filter_by(
    user_id=alice.id,
    incident_id=incident.id
).first()
```

---

## 6.5 Le modèle Comment

```python
# app/models/comment.py

class Comment(db.Model):
    __tablename__ = "comments"
    
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text, nullable=False)
    
    # Relations
    author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    incident_id = db.Column(db.Integer, db.ForeignKey("incidents.id"), nullable=False)
    
    # Commentaire parent (pour les réponses imbriquées)
    parent_id = db.Column(db.Integer, db.ForeignKey("comments.id"))
    replies = db.relationship(
        "Comment",
        backref=db.backref("parent", remote_side=[id]),
        lazy="dynamic"
    )
    
    is_official = db.Column(db.Boolean, default=False)  # Commentaire des services municipaux
    
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
```

---

## 6.6 Lazy Loading vs Eager Loading

L'un des aspects les plus importants pour les **performances** avec SQLAlchemy.

```python
# LAZY LOADING (par défaut) — Problème N+1 !
# Chaque accès à une relation déclenche une nouvelle requête SQL

incidents = Incident.query.all()  # 1 requête

for incident in incidents:
    print(incident.author.username)  # N requêtes supplémentaires !
    # Si 100 incidents -> 101 requêtes SQL au total

# EAGER LOADING — Solution au problème N+1
from sqlalchemy.orm import joinedload, subqueryload

# Joinedload : une seule requête avec JOIN
incidents = Incident.query.options(
    joinedload(Incident.author)    # Charge les auteurs en même temps
).all()
# 1 seule requête SQL (avec JOIN)

for incident in incidents:
    print(incident.author.username)  # Pas de requête supplémentaire !

# Subqueryload : deux requêtes mais souvent plus efficace pour les listes
incidents = Incident.query.options(
    subqueryload(Incident.comments)
).all()

# Charger plusieurs relations en même temps
incidents = Incident.query.options(
    joinedload(Incident.author),
    subqueryload(Incident.comments).subqueryload(Comment.author)
).all()
```

### Options de lazy dans la définition des relations

```python
# lazy="select"   (défaut) — Requête SQL au premier accès
# lazy="joined"   — Toujours eager loading avec JOIN
# lazy="subquery" — Toujours eager loading avec subquery
# lazy="dynamic"  — Retourne une Query plutôt qu'une liste (pour .filter())
# lazy="noload"   — Ne charge jamais automatiquement

comments = db.relationship("Comment", lazy="dynamic")    # user.comments.filter()
```

---

## 6.7 Schéma complet des modèles UrbanPulse

```
┌─────────────────┐       ┌─────────────────────┐
│     users       │       │      incidents       │
├─────────────────┤       ├─────────────────────┤
│ id (PK)         │[BLACK_LEFT-POINTING_POINTER]──┐   │ id (PK)              │
│ email           │   │   │ title                │
│ username        │   │   │ description          │
│ password_hash   │   │   │ category             │
│ role            │   │   │ status               │
│ is_active       │   └───│ author_id (FK)       │
│ created_at      │       │ vote_count           │
└─────────────────┘       │ photo_url            │
         │                │ address              │
         │                │ latitude/longitude   │
         │                │ created_at           │
         │                └──────────┬──────────┘
         │                           │
         │    ┌──────────────────────┘
         │    │
         │    [BLACK_DOWN-POINTING_TRIANGLE]
         │  ┌─────────────────────┐
         │  │       votes         │
         │  ├─────────────────────┤
         └──│ user_id (FK, PK)    │
            │ incident_id (FK, PK)│
            │ voted_at            │
            └─────────────────────┘
         │
         │    ┌─────────────────────┐
         │    │      comments       │
         └────┤─────────────────────┤
              │ id (PK)              │
              │ content              │
              │ author_id (FK)       │[BLACK_LEFT-POINTING_POINTER]── users.id
              │ incident_id (FK)     │[BLACK_LEFT-POINTING_POINTER]── incidents.id
              │ parent_id (FK)       │[BLACK_LEFT-POINTING_POINTER]── comments.id (auto-ref)
              │ is_official          │
              │ created_at           │
              └─────────────────────┘
```

---

## [EDIT] Exercice 6.1 — Relations complètes

> **Objectif** : Implémenter toutes les relations et les tester.

1. Ajoutez le modèle `Comment` au projet.
2. Ajoutez la table `votes` et la relation Many-to-Many.
3. Créez une migration et appliquez-la.
4. Dans `flask shell`, testez :
   - Créer un commentaire sur un incident
   - Faire voter un utilisateur pour un incident
   - Empêcher un double vote
   - Récupérer tous les incidents avec leurs auteurs (eager loading)
   - Trouver les 5 incidents les plus votés

## [EDIT] Exercice 6.2 — Performance et N+1

> **Objectif** : Comprendre et résoudre le problème N+1.

1. Créez 20 incidents liés à 5 utilisateurs différents.
2. Activez `SQLALCHEMY_ECHO = True` dans la config.
3. Récupérez tous les incidents et affichez le username de chaque auteur **sans** `joinedload` -> comptez les requêtes SQL.
4. Refaites la même opération **avec** `joinedload(Incident.author)` -> combien de requêtes maintenant ?
5. **Bonus** : Quel est l'impact sur le temps d'exécution ? Utilisez `timeit` pour mesurer.

---

## [TROPHEE] Récapitulatif Module 1

Vous maîtrisez maintenant :

[OK] La philosophie et l'écosystème Flask
[OK] L'installation et la configuration d'un projet professionnel
[OK] Le routing HTTP avec méthodes, paramètres URL et query parameters
[OK] Les templates Jinja2 avec héritage, boucles et conditions
[OK] Les modèles SQLAlchemy avec tous les types de colonnes
[OK] Les opérations CRUD complètes
[OK] Les migrations avec Flask-Migrate/Alembic
[OK] Les relations One-to-Many et Many-to-Many
[OK] L'optimisation des requêtes avec joinedload

### [PACKAGE] État du projet UrbanPulse à la fin du Module 1

```
urbanpulse/
├── app/
│   ├── __init__.py          [OK] Application factory
│   ├── config.py            [OK] Configs dev/prod/test
│   ├── models/
│   │   ├── __init__.py      [OK]
│   │   ├── user.py          [OK] Modèle User complet
│   │   ├── incident.py      [OK] Modèle Incident complet
│   │   ├── comment.py       [OK] Modèle Comment avec réponses
│   │   └── vote.py          [OK] Table votes
│   ├── routes/
│   │   └── main.py          [OK] Routes de base
│   └── templates/
│       ├── base.html        [OK] Layout principal
│       └── incidents/
│           └── list.html    [OK] Liste des incidents
├── migrations/              [OK] Migrations initialisées
├── requirements.txt         [OK]
└── run.py                   [OK]
```

**-> Module 2 : Vues, Templates avancés, Formulaires, Auth et Sécurité**

# [PYTHON] Formation Flask — Module 2
## Vues Avancées, Formulaires, Authentification & Sécurité
### Parties III & IV — Chapitres 7 à 11

---

> [OBJECTIF] **Rappel Projet UrbanPulse**
> Nous continuons à construire notre plateforme citoyenne. Ce module ajoute :
> la navigation modulaire (Blueprints), les formulaires de signalement,
> l'inscription/connexion des citoyens, et toutes les couches de sécurité.

---

## [WORLD_MAP] Table des matières

- [Chapitre 7 — Views et Routing Avancé](#chapitre-7)
- [Chapitre 8 — Templates Jinja2 Avancés](#chapitre-8)
- [Chapitre 9 — Forms et Validation](#chapitre-9)
- [Chapitre 10 — Authentification](#chapitre-10)
- [Chapitre 11 — Sécurité](#chapitre-11)

---

<a name="chapitre-7"></a>
# [GUIDE] Chapitre 7 — Views et Routing Avancé

## 7.1 Function-Based Views (FBV) — Rappel et approfondissement

Les **Function-Based Views** sont les fonctions décorées avec `@app.route()`. C'est l'approche la plus directe dans Flask.

```python
from flask import Flask, request, jsonify, render_template, redirect, url_for, flash, abort

app = Flask(__name__)

@app.route("/incidents/<int:incident_id>", methods=["GET", "POST", "DELETE"])
def incident_view(incident_id):
    """
    Vue multiméthode pour un incident spécifique.
    """
    # Récupérer l'incident ou retourner 404 si inexistant
    incident = Incident.query.get_or_404(incident_id)
    
    if request.method == "GET":
        return render_template("incidents/detail.html", incident=incident)
    
    elif request.method == "POST":
        # Mise à jour
        data = request.form
        incident.title = data.get("title", incident.title)
        incident.description = data.get("description", incident.description)
        db.session.commit()
        flash("Incident mis à jour avec succès !", "success")
        return redirect(url_for("incident_view", incident_id=incident_id))
    
    elif request.method == "DELETE":
        db.session.delete(incident)
        db.session.commit()
        return jsonify({"message": "Incident supprimé"}), 200
```

### `get_or_404` — Pattern indispensable

```python
# Au lieu de :
incident = Incident.query.get(incident_id)
if incident is None:
    abort(404)

# On écrit simplement :
incident = Incident.query.get_or_404(incident_id)
# Flask retourne automatiquement une page 404 si l'objet n'existe pas

# Version avec message personnalisé :
incident = Incident.query.get_or_404(
    incident_id,
    description="Cet incident n'existe pas ou a été supprimé."
)

# filter_by version
incident = Incident.query.filter_by(id=incident_id, status="open").first_or_404()
```

---

## 7.2 Class-Based Views (CBV) avec MethodView

Pour des vues plus complexes ou pour éviter la répétition, Flask propose `MethodView` :

```python
from flask.views import MethodView

class IncidentAPI(MethodView):
    """
    Vue basée sur une classe pour l'API REST des incidents.
    Les méthodes get(), post(), put(), delete() correspondent
    aux méthodes HTTP.
    """
    
    def get(self, incident_id=None):
        """GET /api/incidents ou GET /api/incidents/<id>"""
        if incident_id is None:
            # Liste tous les incidents
            page = request.args.get("page", 1, type=int)
            incidents = Incident.query.paginate(page=page, per_page=10)
            return jsonify({
                "incidents": [i.to_dict() for i in incidents.items],
                "total": incidents.total,
                "pages": incidents.pages,
                "current_page": incidents.page
            })
        else:
            # Retourne un incident spécifique
            incident = Incident.query.get_or_404(incident_id)
            return jsonify(incident.to_dict())
    
    def post(self):
        """POST /api/incidents — Créer un incident"""
        data = request.get_json()
        
        if not data:
            return jsonify({"error": "Pas de données JSON"}), 400
        
        incident = Incident(
            title=data.get("title"),
            description=data.get("description"),
            category=data.get("category", "autre"),
            author_id=1  # Sera remplacé par l'utilisateur connecté au chapitre 10
        )
        db.session.add(incident)
        db.session.commit()
        return jsonify(incident.to_dict()), 201
    
    def put(self, incident_id):
        """PUT /api/incidents/<id> — Remplacer un incident"""
        incident = Incident.query.get_or_404(incident_id)
        data = request.get_json()
        
        incident.title = data.get("title", incident.title)
        incident.description = data.get("description", incident.description)
        incident.status = data.get("status", incident.status)
        db.session.commit()
        
        return jsonify(incident.to_dict())
    
    def delete(self, incident_id):
        """DELETE /api/incidents/<id>"""
        incident = Incident.query.get_or_404(incident_id)
        db.session.delete(incident)
        db.session.commit()
        return "", 204  # 204 No Content


# Enregistrement de la vue avec une URL flexible
incident_view = IncidentAPI.as_view("incident_api")

app.add_url_rule(
    "/api/incidents",
    view_func=incident_view,
    methods=["GET", "POST"]
)
app.add_url_rule(
    "/api/incidents/<int:incident_id>",
    view_func=incident_view,
    methods=["GET", "PUT", "DELETE"]
)
```

---

## 7.3 Blueprints — Organisation modulaire

Un **Blueprint** est comme une mini-application Flask qui peut avoir ses propres routes, templates, et fichiers statiques. C'est LA solution pour organiser une grande application Flask.

### Pourquoi les Blueprints ?

```
Sans Blueprints :          Avec Blueprints :
───────────────            ─────────────────
app.py                     app/
  (500+ lignes)              routes/
                               main.py       (page accueil, about...)
                               incidents.py  (CRUD incidents)
                               auth.py       (login, register, logout)
                               admin.py      (tableau de bord admin)
                               api.py        (endpoints API REST)
```

### Création et enregistrement d'un Blueprint

```python
# app/routes/incidents.py

from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, abort
from app import db
from app.models.incident import Incident
from app.models.user import User

# Création du Blueprint
# "incidents" = nom du blueprint (utilisé dans url_for: "incidents.list")
# __name__   = pour localiser les templates/static de ce module
incidents_bp = Blueprint(
    "incidents",     # Nom du blueprint
    __name__,
    url_prefix="/incidents"   # Préfixe automatique pour toutes les routes
)

# Grâce à url_prefix="/incidents", cette route répond à GET /incidents
@incidents_bp.route("/")
@incidents_bp.route("")  # Accepte aussi /incidents sans slash final
def list():
    """Liste paginée des incidents, avec filtres."""
    page = request.args.get("page", 1, type=int)
    status = request.args.get("status", "all")
    category = request.args.get("category")
    
    # Construction de la requête avec filtres dynamiques
    query = Incident.query.options(
        joinedload(Incident.author)
    ).order_by(Incident.vote_count.desc(), Incident.created_at.desc())
    
    if status != "all":
        query = query.filter(Incident.status == status)
    
    if category:
        query = query.filter(Incident.category == category)
    
    pagination = query.paginate(page=page, per_page=10)
    
    return render_template(
        "incidents/list.html",
        incidents=pagination.items,
        pagination=pagination,
        current_status=status,
        current_category=category
    )

# Route répondant à GET/POST /incidents/create
@incidents_bp.route("/create", methods=["GET", "POST"])
def create():
    """Formulaire de création d'un incident."""
    if request.method == "POST":
        title = request.form.get("title", "").strip()
        description = request.form.get("description", "").strip()
        category = request.form.get("category", "autre")
        
        # Validation basique
        errors = []
        if not title or len(title) < 10:
            errors.append("Le titre doit faire au moins 10 caractères.")
        if not description or len(description) < 30:
            errors.append("La description doit faire au moins 30 caractères.")
        
        if errors:
            for error in errors:
                flash(error, "danger")
            return render_template("incidents/create.html"), 400
        
        # Création de l'incident (sans auth pour l'instant)
        incident = Incident(
            title=title,
            description=description,
            category=category,
            author_id=1  # TODO: remplacer par current_user.id au ch.10
        )
        db.session.add(incident)
        db.session.commit()
        
        flash("Votre signalement a été soumis avec succès !", "success")
        return redirect(url_for("incidents.detail", incident_id=incident.id))
    
    return render_template("incidents/create.html")

@incidents_bp.route("/<int:incident_id>")
def detail(incident_id):
    """Page de détail d'un incident."""
    incident = Incident.query.options(
        joinedload(Incident.author),
        subqueryload(Incident.comments).joinedload(Comment.author)
    ).get_or_404(incident_id)
    
    return render_template("incidents/detail.html", incident=incident)

@incidents_bp.route("/<int:incident_id>/vote", methods=["POST"])
def vote(incident_id):
    """Vote pour un incident."""
    incident = Incident.query.get_or_404(incident_id)
    incident.upvote()
    db.session.commit()
    
    # Réponse JSON pour les appels AJAX
    return jsonify({"vote_count": incident.vote_count})
```

```python
# app/__init__.py — Enregistrement des Blueprints

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

def create_app(config_name="development"):
    app = Flask(__name__)
    app.config.from_object(f"app.config.{config_name.capitalize()}Config")
    
    db.init_app(app)
    migrate.init_app(app, db)
    
    # Enregistrement des Blueprints
    from app.routes.main import main_bp
    from app.routes.incidents import incidents_bp
    from app.routes.auth import auth_bp
    
    app.register_blueprint(main_bp)
    app.register_blueprint(incidents_bp)   # url_prefix="/incidents" défini dans le blueprint
    app.register_blueprint(auth_bp, url_prefix="/auth")  # Ou défini ici
    
    return app
```

---

## 7.4 Converters d'URL et Routes nommées

```python
# Converters personnalisés
from werkzeug.routing import BaseConverter

class StatusConverter(BaseConverter):
    """Convertisseur qui valide les statuts d'incidents."""
    regex = "(open|in_progress|resolved|rejected)"

app.url_map.converters["status"] = StatusConverter

@app.route("/incidents/status/<status:status_filter>")
def incidents_by_status(status_filter):
    # status_filter est garanti d'être l'un des 4 statuts valides
    # Si une URL comme /incidents/status/unknown est visitée -> 404 automatique
    pass

# URL nommées avec url_for — TOUJOURS utiliser url_for plutôt que des strings
url_for("incidents.list")                           # -> /incidents/
url_for("incidents.detail", incident_id=5)          # -> /incidents/5
url_for("incidents.create")                         # -> /incidents/create
url_for("auth.login")                               # -> /auth/login
url_for("main.index")                               # -> /
url_for("static", filename="css/main.css")          # -> /static/css/main.css

# Générer une URL absolue (avec le domaine)
url_for("incidents.detail", incident_id=5, _external=True)
# -> http://localhost:5000/incidents/5
```

---

## 7.5 Before/After Request Hooks

Flask permet d'exécuter du code **avant ou après** chaque requête dans un Blueprint ou l'application entière.

```python
from flask import g, request
import time

# Exécuté avant chaque requête
@app.before_request
def before_request():
    # g = contexte global de la requête, réinitialisé à chaque requête
    g.start_time = time.time()
    g.db_query_count = 0  # Pour compter les requêtes SQL

# Exécuté après chaque requête (même si une exception est levée)
@app.after_request
def after_request(response):
    duration = time.time() - g.start_time
    # Ajouter des headers de debug
    response.headers["X-Request-Duration"] = f"{duration:.3f}s"
    return response

# Spécifique à un Blueprint
@incidents_bp.before_request
def load_user():
    """Charger l'utilisateur connecté avant chaque requête du blueprint incidents."""
    from flask_login import current_user
    # Vérifier si l'utilisateur est banni, etc.
    if current_user.is_authenticated and not current_user.is_active:
        abort(403)
```

---

## [EDIT] Exercice 7.1 — Architecture Blueprints

> **Objectif** : Refactoriser votre app en Blueprints.

1. Créez les fichiers `app/routes/main.py`, `app/routes/incidents.py`, et `app/routes/auth.py`.
2. Déplacez vos routes existantes dans le bon Blueprint.
3. Enregistrez tous les Blueprints dans `create_app()`.
4. Vérifiez que toutes les URLs fonctionnent encore avec `url_for`.
5. Ajoutez un `@incidents_bp.before_request` qui logge chaque accès au blueprint incidents.

## [EDIT] Exercice 7.2 — MethodView pour l'API

> **Objectif** : Créer une API REST propre avec MethodView.

1. Créez `app/routes/api.py` avec une `IncidentAPI(MethodView)`.
2. Implémentez GET (liste + détail), POST, PUT, DELETE.
3. Enregistrez les routes avec `add_url_rule`.
4. Testez avec HTTPie : `http GET localhost:5000/api/incidents`

---

<a name="chapitre-8"></a>
# [GUIDE] Chapitre 8 — Templates Jinja2 Avancés

## 8.1 Filters (Filtres) personnalisés

Les filtres transforment des données dans les templates. Jinja2 en propose beaucoup, et vous pouvez en créer.

```python
# app/__init__.py ou app/filters.py

from datetime import datetime

def register_filters(app):
    """Enregistre tous les filtres Jinja2 personnalisés."""
    
    @app.template_filter("datetime_fr")
    def datetime_fr(dt):
        """Formate une date en français : '15 janvier 2024 à 14h30'"""
        if dt is None:
            return "—"
        mois = [
            "janvier", "février", "mars", "avril", "mai", "juin",
            "juillet", "août", "septembre", "octobre", "novembre", "décembre"
        ]
        return f"{dt.day} {mois[dt.month-1]} {dt.year} à {dt.strftime('%Hh%M')}"
    
    @app.template_filter("relative_time")
    def relative_time(dt):
        """Retourne 'il y a X minutes/heures/jours'"""
        if dt is None:
            return "—"
        now = datetime.utcnow()
        diff = now - dt
        
        seconds = diff.total_seconds()
        if seconds < 60:
            return "à l'instant"
        elif seconds < 3600:
            minutes = int(seconds / 60)
            return f"il y a {minutes} minute{'s' if minutes > 1 else ''}"
        elif seconds < 86400:
            hours = int(seconds / 3600)
            return f"il y a {hours} heure{'s' if hours > 1 else ''}"
        else:
            days = int(seconds / 86400)
            return f"il y a {days} jour{'s' if days > 1 else ''}"
    
    @app.template_filter("status_badge")
    def status_badge(status):
        """Retourne le HTML d'un badge de statut."""
        badges = {
            "open": '<span class="badge badge-danger">[HOURGLASS_WITH_FLOWING_SAND] En attente</span>',
            "in_progress": '<span class="badge badge-warning">[OUTIL] En cours</span>',
            "resolved": '<span class="badge badge-success">[OK] Résolu</span>',
            "rejected": '<span class="badge badge-secondary">[X] Rejeté</span>',
        }
        return badges.get(status, status)
    
    @app.template_filter("nl2br")
    def nl2br(text):
        """Convertit les sauts de ligne en <br> HTML."""
        from markupsafe import Markup, escape
        return Markup(escape(text).replace('\n', '<br>\n'))
```

```html
<!-- Utilisation dans les templates -->
<p>Signalé {{ incident.created_at | relative_time }}</p>
<p>Date exacte : {{ incident.created_at | datetime_fr }}</p>
{{ incident.status | status_badge | safe }}
<div class="description">{{ incident.description | nl2br | safe }}</div>
```

---

## 8.2 Macros — Composants réutilisables

Les macros sont comme des fonctions dans les templates. Parfaites pour éviter la duplication.

```html
<!-- app/templates/macros/incidents.html -->

{% macro incident_card(incident, show_author=True, show_votes=True) %}
{#
    Macro pour afficher une carte d'incident.
    Paramètres :
    - incident : objet Incident
    - show_author : afficher l'auteur (défaut: True)
    - show_votes : afficher les votes (défaut: True)
#}
<article class="incident-card incident-{{ incident.status }}" 
         data-id="{{ incident.id }}">
    
    <!-- Badge de statut -->
    <div class="card-header">
        {{ incident.status | status_badge | safe }}
        <span class="category-tag">{{ incident.category }}</span>
    </div>
    
    <!-- Contenu principal -->
    <div class="card-body">
        <h3 class="card-title">
            <a href="{{ url_for('incidents.detail', incident_id=incident.id) }}">
                {{ incident.title }}
            </a>
        </h3>
        <p class="card-text">{{ incident.description | truncate(200) }}</p>
        
        {% if incident.address %}
            <p class="address">[IMPORTANT] {{ incident.address }}</p>
        {% endif %}
    </div>
    
    <!-- Footer : auteur, date, votes -->
    <div class="card-footer">
        {% if show_author %}
            <span class="author">
                [UTILISATEUR] {{ incident.author.username }}
            </span>
        {% endif %}
        
        <span class="date">
            {{ incident.created_at | relative_time }}
        </span>
        
        {% if show_votes %}
            <button class="vote-btn" 
                    data-incident-id="{{ incident.id }}"
                    data-voted="{{ 'true' if current_user.is_authenticated and incident.id in user_voted_ids else 'false' }}">
                [BLACK_UP-POINTING_TRIANGLE] <span class="vote-count">{{ incident.vote_count }}</span>
            </button>
        {% endif %}
    </div>
</article>
{% endmacro %}


{% macro pagination_nav(pagination, endpoint, **kwargs) %}
{# Macro de pagination réutilisable #}
{% if pagination.pages > 1 %}
<nav class="pagination" aria-label="Navigation des pages">
    {% if pagination.has_prev %}
        <a href="{{ url_for(endpoint, page=pagination.prev_num, **kwargs) }}" 
           class="page-link"><- Précédent</a>
    {% endif %}
    
    {% for page_num in pagination.iter_pages(left_edge=2, right_edge=2, left_current=2, right_current=2) %}
        {% if page_num %}
            <a href="{{ url_for(endpoint, page=page_num, **kwargs) }}"
               class="page-link {% if page_num == pagination.page %}active{% endif %}">
                {{ page_num }}
            </a>
        {% else %}
            <span class="page-ellipsis">…</span>
        {% endif %}
    {% endfor %}
    
    {% if pagination.has_next %}
        <a href="{{ url_for(endpoint, page=pagination.next_num, **kwargs) }}"
           class="page-link">Suivant -></a>
    {% endif %}
    
    <span class="page-info">
        Page {{ pagination.page }} sur {{ pagination.pages }}
        ({{ pagination.total }} résultats)
    </span>
</nav>
{% endif %}
{% endmacro %}
```

```html
<!-- Utilisation des macros dans un template -->
{% from "macros/incidents.html" import incident_card, pagination_nav %}

{% extends "base.html" %}

{% block content %}
<div class="incidents-grid">
    {% for incident in incidents %}
        {{ incident_card(incident, show_votes=True) }}
    {% else %}
        <div class="empty-state">
            <h3>Aucun signalement trouvé</h3>
            <a href="{{ url_for('incidents.create') }}" class="btn btn-primary">
                Être le premier à signaler
            </a>
        </div>
    {% endfor %}
</div>

{{ pagination_nav(pagination, "incidents.list", status=current_status) }}
{% endblock %}
```

---

## 8.3 Context Processors — Variables globales dans les templates

Un **context processor** injecte automatiquement des variables dans **tous** les templates, sans avoir à les passer dans chaque `render_template()`.

```python
# app/__init__.py

def create_app(config_name="development"):
    app = Flask(__name__)
    # ...
    
    # Context processor : disponible dans TOUS les templates
    @app.context_processor
    def inject_globals():
        """Variables disponibles dans tous les templates."""
        from app.models.incident import Incident
        
        return {
            "app_name": "UrbanPulse",
            "app_version": "1.0.0",
            "open_incidents_count": Incident.query.filter_by(status="open").count(),
            "categories": [
                ("voirie", "[MOTORWAY] Voirie"),
                ("eclairage", "[IDEE] Éclairage"),
                ("dechets", "[SUPPRIMER] Déchets"),
                ("espaces_verts", "[ARBRE] Espaces verts"),
                ("autre", "[LISTE] Autre"),
            ]
        }
    
    return app
```

```html
<!-- Dans n'importe quel template, sans avoir besoin de le passer -->
<title>{{ app_name }} — Signalements</title>
<nav>
    <span class="badge">{{ open_incidents_count }} en attente</span>
</nav>
<select name="category">
    {% for value, label in categories %}
        <option value="{{ value }}">{{ label }}</option>
    {% endfor %}
</select>
```

---

## 8.4 Template avancé : Page de détail d'incident

```html
<!-- app/templates/incidents/detail.html -->
{% extends "base.html" %}

{% block title %}{{ incident.title }} — {{ app_name }}{% endblock %}

{% block extra_css %}
<link rel="stylesheet" href="{{ url_for('static', filename='css/incidents.css') }}">
{% endblock %}

{% block content %}
<div class="container incident-detail">
    
    <!-- En-tête de l'incident -->
    <header class="incident-header">
        <div class="incident-meta">
            {{ incident.status | status_badge | safe }}
            <span class="category">{{ incident.category }}</span>
        </div>
        
        <h1>{{ incident.title }}</h1>
        
        <div class="incident-info">
            <span>[UTILISATEUR] Signalé par <strong>{{ incident.author.username }}</strong></span>
            <span>[CALENDRIER] {{ incident.created_at | datetime_fr }}</span>
            {% if incident.address %}
                <span>[IMPORTANT] {{ incident.address }}</span>
            {% endif %}
        </div>
    </header>
    
    <!-- Corps principal -->
    <div class="incident-body">
        
        <!-- Photo si disponible -->
        {% if incident.photo_url %}
            <div class="incident-photo">
                <img src="{{ url_for('static', filename='uploads/' + incident.photo_url) }}"
                     alt="Photo du signalement">
            </div>
        {% endif %}
        
        <!-- Description -->
        <section class="description">
            <h2>Description</h2>
            <div class="description-content">
                {{ incident.description | nl2br | safe }}
            </div>
        </section>
        
        <!-- Carte si coordonnées disponibles -->
        {% if incident.latitude and incident.longitude %}
            <section class="map-section">
                <h2>Localisation</h2>
                <div id="map" 
                     data-lat="{{ incident.latitude }}"
                     data-lng="{{ incident.longitude }}">
                </div>
            </section>
        {% endif %}
        
        <!-- Vote -->
        <section class="vote-section">
            <p>Cet incident vous affecte aussi ?</p>
            <button id="vote-btn" 
                    class="btn btn-primary btn-vote"
                    data-incident-id="{{ incident.id }}">
                [BLACK_UP-POINTING_TRIANGLE] Voter ({{ incident.vote_count }} vote{{ 's' if incident.vote_count > 1 }})
            </button>
        </section>
        
        <!-- Actions (si admin ou auteur) -->
        {% if current_user.is_authenticated %}
            {% if current_user.role in ["admin", "moderator"] or current_user.id == incident.author_id %}
                <section class="actions">
                    <a href="{{ url_for('incidents.edit', incident_id=incident.id) }}" 
                       class="btn btn-secondary">[EDIT] Modifier</a>
                    
                    {% if current_user.role in ["admin", "moderator"] %}
                        <form method="POST" 
                              action="{{ url_for('incidents.change_status', incident_id=incident.id) }}"
                              style="display:inline">
                            <select name="status">
                                <option value="open">En attente</option>
                                <option value="in_progress">En cours</option>
                                <option value="resolved">Résolu</option>
                                <option value="rejected">Rejeté</option>
                            </select>
                            <button type="submit" class="btn btn-warning">
                                Changer le statut
                            </button>
                        </form>
                    {% endif %}
                </section>
            {% endif %}
        {% endif %}
        
        <!-- Commentaires -->
        <section class="comments-section">
            <h2>
                Commentaires 
                <span class="badge">{{ incident.comments.count() }}</span>
            </h2>
            
            {% for comment in incident.comments.order_by("created_at asc").all() %}
                <article class="comment {% if comment.is_official %}official{% endif %}">
                    {% if comment.is_official %}
                        <span class="official-badge">[CLASSICAL_BUILDING] Service municipal</span>
                    {% endif %}
                    
                    <header class="comment-header">
                        <strong>{{ comment.author.username }}</strong>
                        <time>{{ comment.created_at | relative_time }}</time>
                    </header>
                    
                    <div class="comment-body">
                        {{ comment.content | nl2br | safe }}
                    </div>
                </article>
            {% else %}
                <p class="no-comments">Aucun commentaire pour le moment. Soyez le premier !</p>
            {% endfor %}
            
            <!-- Formulaire de commentaire -->
            {% if current_user.is_authenticated %}
                <form method="POST" 
                      action="{{ url_for('incidents.add_comment', incident_id=incident.id) }}"
                      class="comment-form">
                    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
                    <textarea name="content" 
                              placeholder="Votre commentaire..."
                              rows="3"
                              required></textarea>
                    <button type="submit" class="btn btn-primary">
                        Commenter
                    </button>
                </form>
            {% else %}
                <p>
                    <a href="{{ url_for('auth.login') }}">Connectez-vous</a> 
                    pour laisser un commentaire.
                </p>
            {% endif %}
        </section>
        
    </div>
</div>
{% endblock %}

{% block extra_js %}
<script src="{{ url_for('static', filename='js/vote.js') }}"></script>
{% if incident.latitude and incident.longitude %}
    <script src="https://unpkg.com/leaflet/dist/leaflet.js"></script>
    <script src="{{ url_for('static', filename='js/map.js') }}"></script>
{% endif %}
{% endblock %}
```

---

## [EDIT] Exercice 8.1 — Macros et filtres

> **Objectif** : Créer des composants réutilisables.

1. Créez le fichier `app/templates/macros/incidents.html` avec les macros `incident_card` et `pagination_nav`.
2. Créez 3 filtres personnalisés : `relative_time`, `status_badge`, et un filtre `truncate_words(n)` (tronque à n mots).
3. Ajoutez un context processor qui injecte les catégories disponibles.
4. Utilisez les macros dans `incidents/list.html` et `incidents/detail.html`.

## [EDIT] Exercice 8.2 — Template complet

> **Objectif** : Créer la page de détail complète.

1. Créez `app/templates/incidents/detail.html` en héritant de `base.html`.
2. Affichez le titre, la description, le statut, l'auteur, la date.
3. Affichez la liste des commentaires (même fictifs pour l'instant).
4. Ajoutez le formulaire de commentaire (sans le traiter côté serveur pour l'instant).
5. **Bonus** : Ajoutez un CSS minimal pour rendre la page présentable.

---

<a name="chapitre-9"></a>
# [GUIDE] Chapitre 9 — Forms et Validation

## 9.1 Pourquoi Flask-WTF ?

Gérer des formulaires manuellement avec `request.form` est fastidieux et risqué :
- Oubli de validation -> failles de sécurité
- Pas de protection CSRF -> attaques Cross-Site Request Forgery
- Code de validation dupliqué partout

**Flask-WTF** + **WTForms** résout tout ça :

```bash
pip install flask-wtf email-validator
pip freeze > requirements.txt
```

```python
# app/config.py
class Config:
    # Requis pour la protection CSRF
    SECRET_KEY = os.environ.get("SECRET_KEY") or "super-secret-dev-key-change-in-prod"
    WTF_CSRF_ENABLED = True
```

---

## 9.2 Créer des formulaires WTForms

```python
# app/forms/incident.py

from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileAllowed, FileSize
from wtforms import (
    StringField, TextAreaField, SelectField, 
    FloatField, SubmitField, HiddenField
)
from wtforms.validators import (
    DataRequired, Length, Optional, NumberRange, ValidationError
)
from app.models.incident import Incident

class IncidentForm(FlaskForm):
    """
    Formulaire de création/édition d'un incident.
    Hérite de FlaskForm qui gère automatiquement la protection CSRF.
    """
    
    title = StringField(
        "Titre du signalement",
        validators=[
            DataRequired(message="Le titre est obligatoire."),
            Length(
                min=10, 
                max=200,
                message="Le titre doit faire entre 10 et 200 caractères."
            )
        ],
        render_kw={
            "placeholder": "Ex: Nid-de-poule dangereux rue Victor Hugo",
            "class": "form-control"
        }
    )
    
    description = TextAreaField(
        "Description détaillée",
        validators=[
            DataRequired(message="La description est obligatoire."),
            Length(
                min=30,
                message="Décrivez le problème en au moins 30 caractères."
            )
        ],
        render_kw={
            "rows": 5,
            "placeholder": "Décrivez précisément le problème...",
            "class": "form-control"
        }
    )
    
    category = SelectField(
        "Catégorie",
        choices=[
            ("voirie", "[MOTORWAY] Voirie (route, trottoir, signalisation)"),
            ("eclairage", "[IDEE] Éclairage public"),
            ("dechets", "[SUPPRIMER] Déchets et propreté"),
            ("espaces_verts", "[ARBRE] Espaces verts"),
            ("autre", "[LISTE] Autre")
        ],
        validators=[DataRequired()],
        render_kw={"class": "form-select"}
    )
    
    address = StringField(
        "Adresse ou description du lieu",
        validators=[
            Optional(),
            Length(max=300)
        ],
        render_kw={
            "placeholder": "Ex: 45 rue Victor Hugo, angle avec la rue du Commerce",
            "class": "form-control"
        }
    )
    
    latitude = FloatField(
        "Latitude",
        validators=[
            Optional(),
            NumberRange(min=-90, max=90, message="Latitude invalide (-90 à 90).")
        ],
        render_kw={"class": "form-control"}
    )
    
    longitude = FloatField(
        "Longitude",
        validators=[
            Optional(),
            NumberRange(min=-180, max=180, message="Longitude invalide (-180 à 180).")
        ],
        render_kw={"class": "form-control"}
    )
    
    photo = FileField(
        "Photo du problème (optionnel)",
        validators=[
            Optional(),
            FileAllowed(
                ["jpg", "jpeg", "png", "webp"],
                message="Seules les images JPG, PNG et WebP sont acceptées."
            ),
            # FileSize nécessite flask-wtf >= 1.0
            # FileSize(max_size=5 * 1024 * 1024, message="La photo ne doit pas dépasser 5 Mo.")
        ]
    )
    
    submit = SubmitField(
        "Soumettre le signalement",
        render_kw={"class": "btn btn-primary btn-lg"}
    )
    
    # ========== Validation personnalisée ==========
    
    def validate_title(self, field):
        """
        Validation personnalisée pour le titre.
        Nommée validate_<champ> — appelée automatiquement.
        """
        # Vérifier que le titre ne contient pas de caractères suspects
        forbidden = ["<script>", "javascript:", "onclick="]
        for pattern in forbidden:
            if pattern.lower() in field.data.lower():
                raise ValidationError("Le titre contient des caractères non autorisés.")
    
    def validate_photo(self, field):
        """Vérifie la taille de la photo."""
        if field.data:
            # Lire les données pour vérifier la taille
            field.data.seek(0, 2)  # Aller à la fin
            size = field.data.tell()
            field.data.seek(0)     # Revenir au début
            
            max_size = 5 * 1024 * 1024  # 5 Mo
            if size > max_size:
                raise ValidationError("La photo ne doit pas dépasser 5 Mo.")
```

---

## 9.3 Formulaire d'authentification

```python
# app/forms/auth.py

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import (
    DataRequired, Email, Length, EqualTo, ValidationError
)
from app.models.user import User

class RegisterForm(FlaskForm):
    """Formulaire d'inscription."""
    
    username = StringField(
        "Nom d'utilisateur",
        validators=[
            DataRequired(message="Le nom d'utilisateur est obligatoire."),
            Length(
                min=3, max=30,
                message="Le nom d'utilisateur doit faire entre 3 et 30 caractères."
            )
        ],
        render_kw={"placeholder": "alice_martin", "class": "form-control"}
    )
    
    email = StringField(
        "Adresse email",
        validators=[
            DataRequired(message="L'email est obligatoire."),
            Email(message="Adresse email invalide.")
        ],
        render_kw={"placeholder": "alice@example.com", "class": "form-control"}
    )
    
    password = PasswordField(
        "Mot de passe",
        validators=[
            DataRequired(message="Le mot de passe est obligatoire."),
            Length(
                min=8,
                message="Le mot de passe doit faire au moins 8 caractères."
            )
        ],
        render_kw={"class": "form-control"}
    )
    
    password_confirm = PasswordField(
        "Confirmer le mot de passe",
        validators=[
            DataRequired(),
            EqualTo(
                "password",
                message="Les mots de passe ne correspondent pas."
            )
        ],
        render_kw={"class": "form-control"}
    )
    
    accept_terms = BooleanField(
        "J'accepte les conditions d'utilisation",
        validators=[DataRequired(message="Vous devez accepter les conditions.")]
    )
    
    submit = SubmitField(
        "Créer mon compte",
        render_kw={"class": "btn btn-success btn-lg w-100"}
    )
    
    # Validations personnalisées qui vérifient la base de données
    def validate_username(self, field):
        """Vérifie que le username n'est pas déjà utilisé."""
        existing = User.query.filter_by(username=field.data).first()
        if existing:
            raise ValidationError(
                "Ce nom d'utilisateur est déjà pris. Choisissez-en un autre."
            )
    
    def validate_email(self, field):
        """Vérifie que l'email n'est pas déjà utilisé."""
        existing = User.query.filter_by(email=field.data.lower()).first()
        if existing:
            raise ValidationError(
                "Un compte existe déjà avec cette adresse email."
            )


class LoginForm(FlaskForm):
    """Formulaire de connexion."""
    
    email = StringField(
        "Email",
        validators=[
            DataRequired(message="L'email est obligatoire."),
            Email()
        ],
        render_kw={"placeholder": "votre@email.com", "class": "form-control"}
    )
    
    password = PasswordField(
        "Mot de passe",
        validators=[DataRequired(message="Le mot de passe est obligatoire.")],
        render_kw={"class": "form-control"}
    )
    
    remember_me = BooleanField(
        "Rester connecté(e)",
        default=False
    )
    
    submit = SubmitField(
        "Se connecter",
        render_kw={"class": "btn btn-primary btn-lg w-100"}
    )
```

---

## 9.4 Routes utilisant les formulaires

```python
# app/routes/incidents.py

import os
import uuid
from werkzeug.utils import secure_filename
from flask import current_app

@incidents_bp.route("/create", methods=["GET", "POST"])
def create():
    form = IncidentForm()
    
    # form.validate_on_submit() = True si POST ET validation réussie
    if form.validate_on_submit():
        
        # Gestion de l'upload de photo
        photo_filename = None
        if form.photo.data:
            photo = form.photo.data
            # Nom de fichier sécurisé et unique
            ext = photo.filename.rsplit(".", 1)[1].lower()
            photo_filename = f"{uuid.uuid4()}.{ext}"
            
            upload_folder = os.path.join(
                current_app.root_path, "static", "uploads"
            )
            os.makedirs(upload_folder, exist_ok=True)
            photo.save(os.path.join(upload_folder, photo_filename))
        
        # Création de l'incident avec les données validées du formulaire
        incident = Incident(
            title=form.title.data,
            description=form.description.data,
            category=form.category.data,
            address=form.address.data,
            latitude=form.latitude.data,
            longitude=form.longitude.data,
            photo_url=photo_filename,
            author_id=current_user.id  # Sera disponible au chapitre 10
        )
        
        db.session.add(incident)
        db.session.commit()
        
        flash("[OK] Votre signalement a été soumis avec succès !", "success")
        return redirect(url_for("incidents.detail", incident_id=incident.id))
    
    # GET ou erreur de validation : afficher le formulaire
    return render_template("incidents/create.html", form=form)
```

---

## 9.5 Template de formulaire avec gestion des erreurs

```html
<!-- app/templates/incidents/create.html -->
{% extends "base.html" %}

{% block title %}Nouveau signalement — {{ app_name }}{% endblock %}

{% block content %}
<div class="container" style="max-width: 700px;">
    <h1>[IMPORTANT] Signaler un problème</h1>
    <p class="lead">Aidez votre ville à s'améliorer en signalant les problèmes du quotidien.</p>
    
    <form method="POST" enctype="multipart/form-data" novalidate>
        
        {# Token CSRF — OBLIGATOIRE pour chaque formulaire POST #}
        {{ form.hidden_tag() }}
        
        {# Macro pour afficher un champ avec ses erreurs #}
        {% macro render_field(field, help_text=None) %}
            <div class="mb-3 {% if field.errors %}has-error{% endif %}">
                {{ field.label(class="form-label fw-bold") }}
                {{ field(**field.render_kw if field.render_kw else {}) }}
                
                {# Messages d'erreur de validation #}
                {% if field.errors %}
                    {% for error in field.errors %}
                        <div class="invalid-feedback d-block text-danger">
                            [ATTENTION] {{ error }}
                        </div>
                    {% endfor %}
                {% endif %}
                
                {# Texte d'aide optionnel #}
                {% if help_text %}
                    <div class="form-text text-muted">{{ help_text }}</div>
                {% endif %}
            </div>
        {% endmacro %}
        
        {{ render_field(form.title) }}
        {{ render_field(form.category) }}
        {{ render_field(form.description, help_text="Plus votre description est précise, plus vite le problème sera traité.") }}
        {{ render_field(form.address, help_text="Optionnel mais très utile pour localiser le problème.") }}
        
        <div class="row">
            <div class="col">{{ render_field(form.latitude) }}</div>
            <div class="col">{{ render_field(form.longitude) }}</div>
        </div>
        
        {{ render_field(form.photo, help_text="JPG, PNG ou WebP, max 5 Mo.") }}
        
        <div class="d-grid gap-2">
            {{ form.submit() }}
        </div>
    </form>
</div>
{% endblock %}
```

---

## [EDIT] Exercice 9.1 — Formulaires complets

> **Objectif** : Implémenter les formulaires de création d'incident et d'inscription.

1. Créez `app/forms/incident.py` avec `IncidentForm`.
2. Créez `app/forms/auth.py` avec `RegisterForm` et `LoginForm`.
3. Mettez à jour la route `incidents.create` pour utiliser le formulaire.
4. Créez le template `incidents/create.html` avec affichage des erreurs.
5. Testez la validation :
   - Soumettez un formulaire vide -> toutes les erreurs s'affichent ?
   - Soumettez un titre trop court -> l'erreur apparaît ?
   - Soumettez un formulaire valide -> redirection ?

## [EDIT] Exercice 9.2 — Upload de fichiers

> **Objectif** : Gérer l'upload de photos.

1. Ajoutez la gestion d'upload dans la route `create`.
2. Créez le dossier `app/static/uploads/`.
3. Affichez la photo dans `incidents/detail.html`.
4. Testez avec une vraie image, puis avec un fichier trop grand.
5. **Bonus** : Utilisez `Pillow` pour redimensionner l'image à max 800px avant de la sauvegarder.

---

<a name="chapitre-10"></a>
# [GUIDE] Chapitre 10 — Authentification

## 10.1 Flask-Login — Sessions utilisateurs

Flask-Login gère la session des utilisateurs connectés : connexion, déconnexion, protection des routes.

```bash
pip install flask-login
pip freeze > requirements.txt
```

```python
# app/__init__.py

from flask_login import LoginManager

login_manager = LoginManager()

def create_app(config_name="development"):
    app = Flask(__name__)
    # ...
    
    login_manager.init_app(app)
    
    # Route vers laquelle rediriger les non-connectés
    login_manager.login_view = "auth.login"
    login_manager.login_message = "Connectez-vous pour accéder à cette page."
    login_manager.login_message_category = "warning"
    
    return app
```

### Configurer le modèle User pour Flask-Login

```python
# app/models/user.py

from flask_login import UserMixin
from app import db, login_manager

class User(UserMixin, db.Model):
    """
    UserMixin ajoute les méthodes requises par Flask-Login :
    - is_authenticated : True si l'utilisateur est connecté
    - is_active        : True si le compte est actif
    - is_anonymous     : False pour les utilisateurs réels
    - get_id()         : retourne l'ID comme chaîne de caractères
    """
    __tablename__ = "users"
    
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(256), nullable=False)
    role = db.Column(db.String(20), default="citizen", nullable=False)
    is_active = db.Column(db.Boolean, default=True, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    # Override is_active de UserMixin pour utiliser notre colonne
    @property
    def is_active(self):
        return self._is_active
    
    @is_active.setter
    def is_active(self, value):
        self._is_active = value
    
    def set_password(self, password):
        from werkzeug.security import generate_password_hash
        self.password_hash = generate_password_hash(password, method="pbkdf2:sha256:600000")
    
    def check_password(self, password):
        from werkzeug.security import check_password_hash
        return check_password_hash(self.password_hash, password)
    
    def has_role(self, *roles):
        """Vérifie si l'utilisateur a l'un des rôles spécifiés."""
        return self.role in roles


# Callback OBLIGATOIRE : charge un utilisateur depuis l'ID stocké en session
@login_manager.user_loader
def load_user(user_id):
    """
    Flask-Login appelle cette fonction avec l'ID stocké dans le cookie de session.
    Elle doit retourner l'objet User ou None si l'ID est invalide.
    """
    return User.query.get(int(user_id))
```

---

## 10.2 Routes d'authentification

```python
# app/routes/auth.py

from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from app import db
from app.models.user import User
from app.forms.auth import LoginForm, RegisterForm

auth_bp = Blueprint("auth", __name__, url_prefix="/auth")


@auth_bp.route("/register", methods=["GET", "POST"])
def register():
    """Inscription d'un nouvel utilisateur."""
    
    # Rediriger si déjà connecté
    if current_user.is_authenticated:
        return redirect(url_for("main.index"))
    
    form = RegisterForm()
    
    if form.validate_on_submit():
        user = User(
            username=form.username.data,
            email=form.email.data.lower(),
            role="citizen"
        )
        user.set_password(form.password.data)
        
        db.session.add(user)
        db.session.commit()
        
        # Connecter directement après l'inscription
        login_user(user, remember=False)
        
        flash(f"[BRAVO] Bienvenue sur UrbanPulse, {user.username} !", "success")
        return redirect(url_for("main.index"))
    
    return render_template("auth/register.html", form=form)


@auth_bp.route("/login", methods=["GET", "POST"])
def login():
    """Connexion d'un utilisateur existant."""
    
    if current_user.is_authenticated:
        return redirect(url_for("main.index"))
    
    form = LoginForm()
    
    if form.validate_on_submit():
        # Chercher l'utilisateur par email (insensible à la casse)
        user = User.query.filter_by(email=form.email.data.lower()).first()
        
        # Vérification en deux étapes (ne PAS révéler si c'est l'email ou le mdp qui est faux)
        if user is None or not user.check_password(form.password.data):
            flash("Email ou mot de passe incorrect.", "danger")
            return render_template("auth/login.html", form=form)
        
        # Vérifier que le compte est actif
        if not user.is_active:
            flash("Votre compte a été désactivé. Contactez l'administration.", "warning")
            return render_template("auth/login.html", form=form)
        
        # Connexion réussie
        login_user(user, remember=form.remember_me.data)
        
        # Rediriger vers la page demandée (si l'utilisateur a été redirigé vers /login)
        next_page = request.args.get("next")
        
        # Sécurité : vérifier que next_page est une URL relative (éviter les redirections malveillantes)
        if next_page and not next_page.startswith("/"):
            next_page = None
        
        flash(f"Bon retour, {user.username} !", "success")
        return redirect(next_page or url_for("main.index"))
    
    return render_template("auth/login.html", form=form)


@auth_bp.route("/logout")
@login_required  # Protège la route — seuls les connectés peuvent se déconnecter
def logout():
    """Déconnexion."""
    username = current_user.username
    logout_user()
    flash(f"À bientôt, {username} ! Vous avez été déconnecté(e).", "info")
    return redirect(url_for("main.index"))


@auth_bp.route("/profile")
@login_required
def profile():
    """Profil de l'utilisateur connecté."""
    # current_user est disponible grâce à Flask-Login
    incidents_count = current_user.incidents.count()
    votes_count = current_user.votes.count()
    
    return render_template(
        "auth/profile.html",
        user=current_user,
        incidents_count=incidents_count,
        votes_count=votes_count
    )
```

---

## 10.3 Protection des routes

```python
from flask_login import login_required, current_user

# Protection simple — seulement les utilisateurs connectés
@incidents_bp.route("/create", methods=["GET", "POST"])
@login_required
def create():
    # current_user est l'utilisateur connecté
    # Si non connecté, redirigé vers login_view configuré
    pass

# Protection avec vérification de rôle
def admin_required(f):
    """Décorateur personnalisé pour les routes admin uniquement."""
    from functools import wraps
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not current_user.is_authenticated:
            return redirect(url_for("auth.login"))
        if not current_user.has_role("admin"):
            abort(403)  # Forbidden
        return f(*args, **kwargs)
    return decorated_function

def moderator_required(f):
    """Décorateur pour modérateurs et admins."""
    from functools import wraps
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not current_user.is_authenticated:
            return redirect(url_for("auth.login"))
        if not current_user.has_role("admin", "moderator"):
            abort(403)
        return f(*args, **kwargs)
    return decorated_function

# Utilisation des décorateurs
@incidents_bp.route("/<int:incident_id>/status", methods=["POST"])
@login_required
@moderator_required
def change_status(incident_id):
    """Seuls les modérateurs et admins peuvent changer le statut."""
    pass

@admin_bp.route("/users")
@login_required
@admin_required
def user_management():
    """Seuls les admins voient la gestion des utilisateurs."""
    pass
```

---

## 10.4 JWT pour les APIs (Flask-JWT-Extended)

Pour les APIs REST consommées par un frontend JavaScript, on utilise des JWT (JSON Web Tokens) plutôt que des cookies de session.

```bash
pip install flask-jwt-extended
```

```python
# app/__init__.py
from flask_jwt_extended import JWTManager

jwt = JWTManager()

def create_app(config_name="development"):
    app = Flask(__name__)
    app.config["JWT_SECRET_KEY"] = os.environ.get("JWT_SECRET_KEY", "jwt-dev-secret")
    app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(hours=1)
    app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=30)
    
    jwt.init_app(app)
    return app
```

```python
# app/routes/api/auth.py — Routes d'auth pour l'API

from flask import Blueprint, request, jsonify
from flask_jwt_extended import (
    create_access_token, create_refresh_token,
    jwt_required, get_jwt_identity, get_jwt
)
from app.models.user import User

api_auth_bp = Blueprint("api_auth", __name__, url_prefix="/api/auth")

@api_auth_bp.route("/login", methods=["POST"])
def login():
    """Connexion via API — retourne un JWT."""
    data = request.get_json()
    
    if not data:
        return jsonify({"error": "Pas de données"}), 400
    
    user = User.query.filter_by(email=data.get("email", "").lower()).first()
    
    if not user or not user.check_password(data.get("password", "")):
        return jsonify({"error": "Identifiants incorrects"}), 401
    
    # Créer les tokens
    access_token = create_access_token(
        identity=user.id,
        additional_claims={"role": user.role}
    )
    refresh_token = create_refresh_token(identity=user.id)
    
    return jsonify({
        "access_token": access_token,
        "refresh_token": refresh_token,
        "user": user.to_dict()
    }), 200


@api_auth_bp.route("/refresh", methods=["POST"])
@jwt_required(refresh=True)
def refresh():
    """Renouveler l'access token avec le refresh token."""
    current_user_id = get_jwt_identity()
    access_token = create_access_token(identity=current_user_id)
    return jsonify({"access_token": access_token})


# Utilisation dans les routes API
@incidents_api.route("/incidents", methods=["POST"])
@jwt_required()  # Vérifie le Bearer token dans Authorization header
def create_incident():
    current_user_id = get_jwt_identity()  # ID de l'utilisateur connecté
    claims = get_jwt()                     # additional_claims {"role": "citizen"}
    
    user = User.query.get(current_user_id)
    # ...
```

---

## [EDIT] Exercice 10.1 — Authentification complète

> **Objectif** : Implémenter l'inscription et la connexion.

1. Installez Flask-Login et configurez-le.
2. Ajoutez `UserMixin` au modèle User et configurez `user_loader`.
3. Créez les routes `register`, `login`, `logout`, `profile`.
4. Créez les templates correspondants.
5. Protégez la route `incidents.create` avec `@login_required`.
6. Mettez à jour la route `create` pour utiliser `current_user.id`.
7. Testez le flux complet : inscription -> connexion -> créer un incident -> déconnexion.

---

<a name="chapitre-11"></a>
# [GUIDE] Chapitre 11 — Sécurité

## 11.1 Hashing des mots de passe

**[ATTENTION] Règle absolue : Ne jamais stocker un mot de passe en clair.**

```python
from werkzeug.security import generate_password_hash, check_password_hash

# Générer un hash bcrypt/pbkdf2 (irréversible)
password = "MonMotDePasse123!"
hashed = generate_password_hash(
    password,
    method="pbkdf2:sha256:600000"  # 600 000 itérations PBKDF2-SHA256
    # Ou "scrypt" pour encore plus de sécurité (Python 3.9+)
)
print(hashed)
# pbkdf2:sha256:600000$abc123...xyz (toujours différent même pour le même mdp)

# Vérification
is_valid = check_password_hash(hashed, "MonMotDePasse123!")  # True
is_valid = check_password_hash(hashed, "AutreMotDePasse")    # False

# Dans le modèle User :
class User(db.Model):
    def set_password(self, password):
        self.password_hash = generate_password_hash(
            password,
            method="pbkdf2:sha256:600000"
        )
    
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)
```

---

## 11.2 Protection CSRF

Le **Cross-Site Request Forgery** est une attaque où un site malveillant fait effectuer des actions à un utilisateur connecté sur votre site à son insu.

Flask-WTF gère automatiquement la protection CSRF avec `form.hidden_tag()`.

```python
# app/__init__.py
from flask_wtf.csrf import CSRFProtect

csrf = CSRFProtect()

def create_app(config_name="development"):
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "clé-secrète-très-longue-et-aléatoire"
    app.config["WTF_CSRF_ENABLED"] = True
    
    csrf.init_app(app)
    return app
```

```html
<!-- Dans chaque formulaire POST -->
<form method="POST">
    {{ form.hidden_tag() }}   <!-- Génère le champ CSRF token -->
    <!-- Ou manuellement : -->
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
    ...
</form>
```

```python
# Pour les routes AJAX qui n'utilisent pas WTForms
# Le token doit être dans le header X-CSRFToken

# Dans le JavaScript :
const csrfToken = document.querySelector('meta[name="csrf-token"]').content;

fetch("/incidents/5/vote", {
    method: "POST",
    headers: {
        "X-CSRFToken": csrfToken,
        "Content-Type": "application/json"
    }
})

# Dans le template base.html :
# <meta name="csrf-token" content="{{ csrf_token() }}">

# Pour exclure une route de la protection CSRF (par exemple une API avec JWT)
@csrf.exempt
@api_bp.route("/incidents", methods=["POST"])
@jwt_required()
def create_incident():
    pass
```

---

## 11.3 Protection XSS (Cross-Site Scripting)

Le XSS consiste à injecter du JavaScript malveillant dans une page pour voler des données.

```html
<!-- Jinja2 échappe automatiquement les variables — sécurisé par défaut -->
{{ user_input }}
<!-- Si user_input = "<script>alert('XSS')</script>",
     Jinja2 rend : &lt;script&gt;alert('XSS')&lt;/script&gt;
     -> Affiché comme texte, pas exécuté comme code -->

<!-- DANGEREUX — n'utiliser que sur du contenu de confiance -->
{{ user_input | safe }}

<!-- Pour du HTML de confiance (ex: contenu admin) : utiliser Markup -->
from markupsafe import Markup
safe_html = Markup("<strong>Texte bold de confiance</strong>")
```

```python
# Configuration des headers de sécurité HTTP
@app.after_request
def add_security_headers(response):
    """Ajoute des headers de sécurité à chaque réponse."""
    
    # Empêche le navigateur d'exécuter des scripts dans des contextes non prévus
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "script-src 'self' 'unsafe-inline' https://cdnjs.cloudflare.com; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: blob:;"
    )
    
    # Empêche le framing de la page (protection clickjacking)
    response.headers["X-Frame-Options"] = "SAMEORIGIN"
    
    # Empêche le sniffing MIME
    response.headers["X-Content-Type-Options"] = "nosniff"
    
    # Force HTTPS (à activer seulement en production)
    # response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    
    return response
```

---

## 11.4 Protection contre l'injection SQL

SQLAlchemy protège **automatiquement** contre l'injection SQL via ses paramètres bindés.

```python
# DANGEREUX — injection SQL possible
username = "' OR '1'='1"  # Attaque classique
db.engine.execute(f"SELECT * FROM users WHERE username = '{username}'")
# Génère : SELECT * FROM users WHERE username = '' OR '1'='1'
# -> Retourne TOUS les utilisateurs !

# SÉCURISÉ — SQLAlchemy échappe automatiquement
User.query.filter_by(username=username).first()
# Génère : SELECT * FROM users WHERE username = ?  avec ['...'] comme paramètre

# Également sécurisé
User.query.filter(User.username == username).first()

# Si vous devez écrire du SQL brut, utilisez des paramètres bindés
from sqlalchemy import text
result = db.session.execute(
    text("SELECT * FROM users WHERE username = :name"),
    {"name": username}  # Jamais f-string !
)
```

---

## 11.5 Rate Limiting avec Flask-Limiter

Empêche les attaques par force brute et l'abus d'API.

```bash
pip install flask-limiter
```

```python
# app/__init__.py

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,  # Limite par adresse IP
    default_limits=["200 per day", "50 per hour"]  # Limites globales
)

def create_app(config_name="development"):
    app = Flask(__name__)
    limiter.init_app(app)
    return app
```

```python
# Limites spécifiques par route
from app import limiter

@auth_bp.route("/login", methods=["GET", "POST"])
@limiter.limit("10 per minute")   # Max 10 tentatives/minute/IP
@limiter.limit("50 per hour")     # Max 50 tentatives/heure/IP
def login():
    pass

@auth_bp.route("/register", methods=["GET", "POST"])
@limiter.limit("5 per hour")      # Évite la création massive de comptes
def register():
    pass

@incidents_bp.route("/<int:id>/vote", methods=["POST"])
@limiter.limit("30 per minute")   # Évite le vote spam
def vote(id):
    pass

# Exempter une route des limites (ex: assets statiques)
@limiter.exempt
@app.route("/health")
def health_check():
    return "OK"
```

---

## 11.6 Variables d'environnement avec python-dotenv

Les secrets ne doivent **jamais** être dans le code source.

```bash
pip install python-dotenv
```

```bash
# .env (jamais committé dans Git !)
FLASK_ENV=development
SECRET_KEY=ma-cle-super-secrete-de-512-bits-generee-aleatoirement
DATABASE_URL=sqlite:///urbanpulse.db
JWT_SECRET_KEY=une-autre-cle-pour-jwt
MAIL_PASSWORD=password-email-secret
```

```python
# run.py — Chargement automatique du .env
from dotenv import load_dotenv
load_dotenv()  # Charge les variables du .env dans os.environ

from app import create_app
app = create_app()

if __name__ == "__main__":
    app.run()
```

```
# .gitignore — Fichiers à ne JAMAIS committer
.env
*.env
venv/
__pycache__/
*.pyc
instance/
*.db
static/uploads/
```

---

## [EDIT] Exercice 11.1 — Hardening sécurité

> **Objectif** : Sécuriser l'application UrbanPulse.

1. Vérifiez que tous les formulaires POST ont `{{ form.hidden_tag() }}`.
2. Ajoutez le middleware `add_security_headers`.
3. Installez Flask-Limiter et appliquez des limites aux routes d'auth.
4. Créez un fichier `.env` et déplacez `SECRET_KEY` dedans.
5. Ajoutez `.env` et `venv/` dans `.gitignore`.
6. **Test de sécurité** : Essayez d'injecter `<script>alert('XSS')</script>` dans le titre d'un incident. Que se passe-t-il ?
7. **Test CSRF** : Essayez de soumettre un formulaire sans le token CSRF. Que se passe-t-il ?

---

## [TROPHEE] Récapitulatif Module 2

Vous maîtrisez maintenant :

[OK] Les Function-Based Views et les Class-Based Views
[OK] L'organisation en Blueprints avec prefixes d'URL
[OK] Les hooks before/after_request
[OK] Les templates Jinja2 avancés (macros, filtres, context processors)
[OK] Les formulaires WTForms avec validation intégrée
[OK] L'upload de fichiers sécurisé
[OK] L'authentification complète avec Flask-Login
[OK] JWT pour les APIs
[OK] La sécurité : CSRF, XSS, injection SQL, rate limiting
[OK] La gestion des secrets avec .env

### [PACKAGE] État du projet UrbanPulse à la fin du Module 2

```
urbanpulse/
├── app/
│   ├── __init__.py          [OK] Flask-Login, Flask-WTF, Flask-Limiter, JWT
│   ├── config.py            [OK] Variables d'environnement
│   ├── models/              [OK] User (avec UserMixin), Incident, Comment, Vote
│   ├── routes/
│   │   ├── main.py          [OK] Pages statiques
│   │   ├── incidents.py     [OK] CRUD complet avec auth
│   │   ├── auth.py          [OK] Register, Login, Logout, Profile
│   │   └── api/
│   │       ├── incidents.py [OK] API REST avec JWT
│   │       └── auth.py      [OK] Auth API avec JWT
│   ├── forms/
│   │   ├── incident.py      [OK] IncidentForm avec upload
│   │   └── auth.py          [OK] RegisterForm, LoginForm
│   ├── templates/
│   │   ├── base.html        [OK] Layout avec nav conditionnelle
│   │   ├── macros/          [OK] incident_card, pagination_nav
│   │   ├── incidents/       [OK] list, detail, create, edit
│   │   └── auth/            [OK] login, register, profile
│   └── static/
│       ├── css/             [OK] Styles de base
│       ├── js/              [OK] vote.js, map.js
│       └── uploads/         [OK] Photos des incidents
├── .env                     [OK] Secrets (hors Git)
├── .gitignore               [OK]
└── requirements.txt         [OK]
```

**-> Module 3 : API REST, Tests, Qualité de code**

# [PYTHON] Formation Flask — Module 3
## API REST Professionnelle, Tâches Asynchrones & Qualité
### Parties V & VI — Chapitres 12 à 16

---

> [OBJECTIF] **UrbanPulse à ce stade**
> L'application a une base solide : modèles, auth, routes, templates.
> Ce module ajoute une **API REST documentée**, des **tâches en arrière-plan**
> (emails de notification, traitement d'images), une **gestion d'erreurs professionnelle**,
> et une **suite de tests** complète.

---

## [WORLD_MAP] Table des matières

- [Chapitre 12 — Création d'API REST](#chapitre-12)
- [Chapitre 13 — Async et Background Tasks](#chapitre-13)
- [Chapitre 14 — Gestion d'erreurs et Logging](#chapitre-14)
- [Chapitre 15 — Tests unitaires et d'intégration](#chapitre-15)
- [Chapitre 16 — Linting et Typage](#chapitre-16)

---

<a name="chapitre-12"></a>
# [GUIDE] Chapitre 12 — Création d'une API REST Professionnelle

## 12.1 Principes REST

**REST (Representational State Transfer)** est un style d'architecture pour les APIs web.

**Les 6 principes REST :**
1. **Client-Serveur** : séparation claire entre frontend et backend
2. **Stateless** : chaque requête est indépendante (pas d'état côté serveur)
3. **Cacheable** : les réponses peuvent être mises en cache
4. **Interface uniforme** : conventions d'URL et méthodes HTTP cohérentes
5. **Système en couches** : le client ne sait pas s'il parle directement au serveur
6. **Code à la demande** (optionnel) : le serveur peut envoyer du code exécutable

**Convention d'URL RESTful pour UrbanPulse :**

```
GET    /api/v1/incidents          -> Liste paginée des incidents
POST   /api/v1/incidents          -> Créer un incident
GET    /api/v1/incidents/{id}     -> Détail d'un incident
PUT    /api/v1/incidents/{id}     -> Remplacer un incident
PATCH  /api/v1/incidents/{id}     -> Modifier partiellement
DELETE /api/v1/incidents/{id}     -> Supprimer un incident

GET    /api/v1/incidents/{id}/comments  -> Commentaires d'un incident
POST   /api/v1/incidents/{id}/comments -> Ajouter un commentaire
POST   /api/v1/incidents/{id}/vote     -> Voter pour un incident

GET    /api/v1/users/{id}         -> Profil public d'un utilisateur
GET    /api/v1/users/me           -> Profil de l'utilisateur connecté

POST   /api/v1/auth/login         -> Connexion (retourne JWT)
POST   /api/v1/auth/register      -> Inscription
POST   /api/v1/auth/refresh       -> Renouveler le token
```

---

## 12.2 Sérialisation avec Marshmallow

**Marshmallow** transforme les objets Python complexes (modèles SQLAlchemy) en JSON, et valide les données entrantes.

```bash
pip install marshmallow flask-marshmallow marshmallow-sqlalchemy
```

```python
# app/schemas/incident.py

from marshmallow import Schema, fields, validate, validates, ValidationError, post_load
from app.models.incident import Incident

class AuthorSchema(Schema):
    """Schema simplifié pour l'auteur (évite la récursion)."""
    id = fields.Int(dump_only=True)
    username = fields.Str(dump_only=True)

class CommentSchema(Schema):
    """Schema pour les commentaires."""
    id = fields.Int(dump_only=True)
    content = fields.Str(
        required=True,
        validate=validate.Length(min=3, max=2000, error="Le commentaire doit faire entre 3 et 2000 caractères.")
    )
    author = fields.Nested(AuthorSchema, dump_only=True)
    is_official = fields.Bool(dump_only=True)
    created_at = fields.DateTime(dump_only=True, format="iso")

class IncidentSchema(Schema):
    """
    Schema principal pour les incidents.
    
    dump_only=True  -> Champ retourné dans les réponses, ignoré dans les requêtes
    load_only=True  -> Champ accepté dans les requêtes, jamais retourné
    required=True   -> Champ obligatoire (pour la création/mise à jour)
    """
    
    # Champs en lecture seule
    id = fields.Int(dump_only=True)
    vote_count = fields.Int(dump_only=True)
    created_at = fields.DateTime(dump_only=True, format="iso")
    updated_at = fields.DateTime(dump_only=True, format="iso")
    resolved_at = fields.DateTime(dump_only=True, format="iso")
    author = fields.Nested(AuthorSchema, dump_only=True)
    
    # Champs modifiables
    title = fields.Str(
        required=True,
        validate=validate.Length(
            min=10, max=200,
            error="Le titre doit faire entre 10 et 200 caractères."
        )
    )
    
    description = fields.Str(
        required=True,
        validate=validate.Length(
            min=30,
            error="La description doit faire au moins 30 caractères."
        )
    )
    
    category = fields.Str(
        validate=validate.OneOf(
            ["voirie", "eclairage", "dechets", "espaces_verts", "autre"],
            error="Catégorie invalide."
        ),
        missing="autre"  # Valeur par défaut si absent
    )
    
    status = fields.Str(
        validate=validate.OneOf(
            ["open", "in_progress", "resolved", "rejected"],
            error="Statut invalide."
        ),
        dump_only=True  # Le statut ne peut être changé que via l'endpoint dédié
    )
    
    address = fields.Str(validate=validate.Length(max=300), allow_none=True)
    latitude = fields.Float(validate=validate.Range(min=-90, max=90), allow_none=True)
    longitude = fields.Float(validate=validate.Range(min=-180, max=180), allow_none=True)
    photo_url = fields.Str(dump_only=True)
    
    # Inclusion des commentaires (optionnelle, via param ?include_comments=true)
    comments = fields.List(fields.Nested(CommentSchema), dump_only=True)
    
    @validates("title")
    def validate_title_no_html(self, value):
        """Validation personnalisée : pas de balises HTML."""
        import re
        if re.search(r'<[^>]+>', value):
            raise ValidationError("Le titre ne peut pas contenir de balises HTML.")


class IncidentUpdateSchema(IncidentSchema):
    """Schema pour la mise à jour partielle (PATCH) — aucun champ obligatoire."""
    title = fields.Str(validate=validate.Length(min=10, max=200))
    description = fields.Str(validate=validate.Length(min=30))
    status = fields.Str(
        validate=validate.OneOf(["open", "in_progress", "resolved", "rejected"])
    )


# Instances réutilisables
incident_schema = IncidentSchema()
incidents_schema = IncidentSchema(many=True)
incident_update_schema = IncidentUpdateSchema()
```

---

## 12.3 Flask-RESTX — API avec documentation automatique

Flask-RESTX génère automatiquement une documentation Swagger/OpenAPI interactve.

```bash
pip install flask-restx
```

```python
# app/api/__init__.py

from flask import Blueprint
from flask_restx import Api

api_blueprint = Blueprint("api", __name__, url_prefix="/api/v1")

api = Api(
    api_blueprint,
    version="1.0",
    title="UrbanPulse API",
    description="API REST de la plateforme citoyenne UrbanPulse",
    doc="/docs",               # URL de la documentation Swagger
    authorizations={
        "Bearer": {
            "type": "apiKey",
            "in": "header",
            "name": "Authorization",
            "description": "Entrez: Bearer <votre_jwt_token>"
        }
    },
    security="Bearer"          # Sécurité par défaut pour tous les endpoints
)
```

```python
# app/api/incidents.py

from flask_restx import Namespace, Resource, fields
from flask_jwt_extended import jwt_required, get_jwt_identity
from flask import request

from app import db
from app.models.incident import Incident
from app.models.user import User
from app.schemas.incident import incident_schema, incidents_schema, incident_update_schema

# Namespace = groupe d'endpoints liés
ns = Namespace("incidents", description="Gestion des signalements citoyens")

# Modèles Swagger pour la documentation (séparés des schemas Marshmallow)
incident_model = ns.model("Incident", {
    "id": fields.Integer(readonly=True, description="Identifiant unique"),
    "title": fields.String(required=True, description="Titre du signalement"),
    "description": fields.String(required=True, description="Description détaillée"),
    "category": fields.String(
        description="Catégorie",
        enum=["voirie", "eclairage", "dechets", "espaces_verts", "autre"]
    ),
    "status": fields.String(
        readonly=True,
        description="Statut actuel",
        enum=["open", "in_progress", "resolved", "rejected"]
    ),
    "vote_count": fields.Integer(readonly=True),
    "address": fields.String(description="Adresse ou description du lieu"),
    "latitude": fields.Float(),
    "longitude": fields.Float(),
    "created_at": fields.DateTime(readonly=True),
})

incident_create_model = ns.model("IncidentCreate", {
    "title": fields.String(required=True),
    "description": fields.String(required=True),
    "category": fields.String(enum=["voirie", "eclairage", "dechets", "espaces_verts", "autre"]),
    "address": fields.String(),
    "latitude": fields.Float(),
    "longitude": fields.Float(),
})

paginated_model = ns.model("PaginatedIncidents", {
    "incidents": fields.List(fields.Nested(incident_model)),
    "total": fields.Integer(description="Nombre total d'incidents"),
    "pages": fields.Integer(description="Nombre total de pages"),
    "current_page": fields.Integer(),
    "per_page": fields.Integer(),
})


@ns.route("/")
class IncidentList(Resource):
    """Endpoint pour la liste et la création d'incidents."""
    
    @ns.doc("list_incidents", params={
        "page": "Numéro de page (défaut: 1)",
        "per_page": "Éléments par page (défaut: 10, max: 50)",
        "status": "Filtrer par statut",
        "category": "Filtrer par catégorie",
        "q": "Recherche textuelle",
        "sort": "Tri : vote_count|created_at (défaut: created_at)",
        "order": "Ordre : asc|desc (défaut: desc)"
    })
    @ns.marshal_with(paginated_model)
    def get(self):
        """Récupérer la liste paginée des incidents."""
        page = request.args.get("page", 1, type=int)
        per_page = min(request.args.get("per_page", 10, type=int), 50)
        status = request.args.get("status")
        category = request.args.get("category")
        search = request.args.get("q")
        sort_by = request.args.get("sort", "created_at")
        order = request.args.get("order", "desc")
        
        query = Incident.query
        
        # Filtres
        if status:
            query = query.filter(Incident.status == status)
        if category:
            query = query.filter(Incident.category == category)
        if search:
            query = query.filter(
                db.or_(
                    Incident.title.ilike(f"%{search}%"),
                    Incident.description.ilike(f"%{search}%"),
                    Incident.address.ilike(f"%{search}%")
                )
            )
        
        # Tri
        sort_column = getattr(Incident, sort_by, Incident.created_at)
        if order == "asc":
            query = query.order_by(sort_column.asc())
        else:
            query = query.order_by(sort_column.desc())
        
        pagination = query.paginate(page=page, per_page=per_page, error_out=False)
        
        return {
            "incidents": incidents_schema.dump(pagination.items),
            "total": pagination.total,
            "pages": pagination.pages,
            "current_page": pagination.page,
            "per_page": per_page
        }
    
    @ns.doc("create_incident")
    @ns.expect(incident_create_model, validate=True)
    @ns.marshal_with(incident_model, code=201)
    @jwt_required()
    def post(self):
        """Créer un nouveau signalement (authentification requise)."""
        current_user_id = get_jwt_identity()
        
        # Validation avec Marshmallow
        errors = incident_schema.validate(request.json)
        if errors:
            ns.abort(422, "Données invalides", errors=errors)
        
        data = incident_schema.load(request.json)
        
        incident = Incident(
            title=data["title"],
            description=data["description"],
            category=data.get("category", "autre"),
            address=data.get("address"),
            latitude=data.get("latitude"),
            longitude=data.get("longitude"),
            author_id=current_user_id
        )
        
        db.session.add(incident)
        db.session.commit()
        
        return incident_schema.dump(incident), 201


@ns.route("/<int:incident_id>")
@ns.param("incident_id", "Identifiant de l'incident")
@ns.response(404, "Incident non trouvé")
class IncidentDetail(Resource):
    """Endpoint pour un incident spécifique."""
    
    @ns.marshal_with(incident_model)
    def get(self, incident_id):
        """Récupérer les détails d'un incident."""
        include_comments = request.args.get("include_comments", "false").lower() == "true"
        
        incident = Incident.query.get_or_404(incident_id)
        result = incident_schema.dump(incident)
        
        if include_comments:
            from app.schemas.incident import CommentSchema
            result["comments"] = CommentSchema(many=True).dump(
                incident.comments.order_by("created_at").all()
            )
        
        return result
    
    @ns.expect(incident_create_model)
    @ns.marshal_with(incident_model)
    @jwt_required()
    def patch(self, incident_id):
        """Modifier partiellement un incident."""
        current_user_id = get_jwt_identity()
        incident = Incident.query.get_or_404(incident_id)
        
        # Vérifier les permissions
        user = User.query.get(current_user_id)
        if incident.author_id != current_user_id and not user.has_role("admin", "moderator"):
            ns.abort(403, "Vous n'êtes pas autorisé à modifier cet incident.")
        
        # Validation partielle
        errors = incident_update_schema.validate(request.json)
        if errors:
            ns.abort(422, "Données invalides", errors=errors)
        
        data = incident_update_schema.load(request.json, partial=True)
        
        # Mise à jour uniquement des champs fournis
        for key, value in data.items():
            setattr(incident, key, value)
        
        db.session.commit()
        return incident_schema.dump(incident)
    
    @jwt_required()
    @ns.response(204, "Incident supprimé")
    def delete(self, incident_id):
        """Supprimer un incident (auteur ou admin uniquement)."""
        current_user_id = get_jwt_identity()
        incident = Incident.query.get_or_404(incident_id)
        
        user = User.query.get(current_user_id)
        if incident.author_id != current_user_id and not user.has_role("admin"):
            ns.abort(403, "Non autorisé.")
        
        db.session.delete(incident)
        db.session.commit()
        return "", 204


@ns.route("/<int:incident_id>/vote")
class IncidentVote(Resource):
    """Endpoint de vote pour un incident."""
    
    @jwt_required()
    @ns.response(200, "Vote enregistré")
    @ns.response(409, "Vote déjà enregistré")
    def post(self, incident_id):
        """Voter pour un incident (un vote par utilisateur)."""
        current_user_id = get_jwt_identity()
        incident = Incident.query.get_or_404(incident_id)
        
        from app.models.vote import Vote
        existing_vote = Vote.query.filter_by(
            user_id=current_user_id,
            incident_id=incident_id
        ).first()
        
        if existing_vote:
            ns.abort(409, "Vous avez déjà voté pour cet incident.")
        
        vote = Vote(user_id=current_user_id, incident_id=incident_id)
        incident.vote_count += 1
        
        db.session.add(vote)
        db.session.commit()
        
        return {"message": "Vote enregistré", "vote_count": incident.vote_count}
```

---

## 12.4 Pagination avancée et filtres

```python
# Utilitaire de pagination réutilisable
def paginate_query(query, schema, default_per_page=10, max_per_page=100):
    """
    Utilitaire générique de pagination.
    Retourne un dict prêt à être retourné en JSON.
    """
    page = request.args.get("page", 1, type=int)
    per_page = min(
        request.args.get("per_page", default_per_page, type=int),
        max_per_page
    )
    
    pagination = query.paginate(page=page, per_page=per_page, error_out=False)
    
    return {
        "data": schema.dump(pagination.items),
        "meta": {
            "total": pagination.total,
            "pages": pagination.pages,
            "page": pagination.page,
            "per_page": per_page,
            "has_next": pagination.has_next,
            "has_prev": pagination.has_prev,
            "next_page": pagination.next_num if pagination.has_next else None,
            "prev_page": pagination.prev_num if pagination.has_prev else None,
        },
        "links": {
            "self": request.url,
            "first": _page_url(1, per_page),
            "last": _page_url(pagination.pages, per_page),
            "next": _page_url(pagination.next_num, per_page) if pagination.has_next else None,
            "prev": _page_url(pagination.prev_num, per_page) if pagination.has_prev else None,
        }
    }

def _page_url(page, per_page):
    """Génère une URL de pagination."""
    from flask import request
    args = request.args.copy()
    args["page"] = page
    args["per_page"] = per_page
    return f"{request.base_url}?{'&'.join(f'{k}={v}' for k, v in args.items())}"
```

---

## [EDIT] Exercice 12.1 — API REST complète

> **Objectif** : Construire l'API REST d'UrbanPulse avec documentation.

1. Créez le namespace `incidents` avec tous les endpoints CRUD.
2. Créez le namespace `auth` avec `login`, `register`, `refresh`.
3. Créez les schemas Marshmallow pour `Incident`, `User`, `Comment`.
4. Ajoutez la pagination à l'endpoint `GET /incidents`.
5. Ajoutez les filtres `status`, `category`, et la recherche `q`.
6. Visitez `http://localhost:5000/api/v1/docs` et testez via Swagger UI.

## [EDIT] Exercice 12.2 — Recherche et filtres avancés

> **Objectif** : Implémenter une recherche multi-critères.

1. Ajoutez le filtre de tri `sort` (vote_count, created_at) avec ordre `asc`/`desc`.
2. Ajoutez un filtre géographique : `/incidents?near_lat=48.8&near_lng=2.3&radius=1000` (radius en mètres).
3. Implémentez la recherche full-text sur titre, description et adresse.
4. Documentez tous les paramètres dans la doc Swagger.

---

<a name="chapitre-13"></a>
# [GUIDE] Chapitre 13 — Async et Background Tasks

## 13.1 Async natif dans Flask ≥ 2.0

Depuis Flask 2.0, vous pouvez utiliser `async`/`await` dans vos vues.

```bash
pip install flask[async]  # Installe aussi aiohttp et asyncio
```

```python
# Vues asynchrones Flask 2.0+
import asyncio
import aiohttp

@app.route("/external-data")
async def external_data():
    """Exemple de vue asynchrone qui appelle une API externe."""
    async with aiohttp.ClientSession() as session:
        # Les deux requêtes s'exécutent en parallèle
        results = await asyncio.gather(
            fetch_weather(session, "Paris"),
            fetch_air_quality(session, "Paris")
        )
    
    return jsonify({
        "weather": results[0],
        "air_quality": results[1]
    })

async def fetch_weather(session, city):
    """Récupère la météo depuis une API."""
    async with session.get(f"https://api.weather.example.com/{city}") as resp:
        return await resp.json()
```

**Note importante :** L'async Flask est utile pour les I/O (appels API, requêtes HTTP), mais SQLAlchemy standard n'est pas async-natif. Pour de l'async avec SQLAlchemy, utilisez SQLAlchemy Async + AsyncSession.

---

## 13.2 Celery — Tâches en arrière-plan

Certaines opérations sont trop longues pour être traitées pendant la requête HTTP (envoi d'email, traitement d'image, génération de rapport...). **Celery** les exécute en arrière-plan.

```bash
pip install celery redis flower
# flower = interface web pour monitorer les tâches Celery
```

### Architecture Celery

```
┌──────────────┐     tâche      ┌─────────┐      déqueue     ┌─────────────┐
│  Flask App   │ ─────────────[BLACK_RIGHT-POINTING_POINTER] │  Redis  │ ──────────────[BLACK_RIGHT-POINTING_POINTER] │   Worker    │
│  (Producer)  │                │ (Broker)│                  │  (Consumer) │
└──────────────┘                └─────────┘                  └─────────────┘
       │                              │                              │
       │          résultat            │           stockage           │
       └──────────────────────────────┴──────────────────────────────┘
                                  (optionnel)
```

### Configuration Celery avec Flask

```python
# app/celery_app.py

from celery import Celery
from flask import Flask

def make_celery(app: Flask) -> Celery:
    """
    Crée une instance Celery configurée pour Flask.
    Chaque tâche a accès au contexte applicatif Flask.
    """
    celery = Celery(
        app.import_name,
        broker=app.config["CELERY_BROKER_URL"],
        backend=app.config["CELERY_RESULT_BACKEND"]
    )
    celery.conf.update(app.config)
    
    class ContextTask(celery.Task):
        """Tâche qui s'exécute dans le contexte applicatif Flask."""
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    
    celery.Task = ContextTask
    return celery
```

```python
# app/config.py

class Config:
    # Redis comme broker de messages et stockage des résultats
    CELERY_BROKER_URL = os.environ.get("REDIS_URL") or "redis://localhost:6379/0"
    CELERY_RESULT_BACKEND = os.environ.get("REDIS_URL") or "redis://localhost:6379/0"
    
    # Sérialisation JSON (plus sûr que pickle)
    CELERY_TASK_SERIALIZER = "json"
    CELERY_RESULT_SERIALIZER = "json"
    CELERY_ACCEPT_CONTENT = ["json"]
    
    # Timezone
    CELERY_TIMEZONE = "Europe/Paris"
    
    # Retry automatique en cas d'échec
    CELERY_TASK_MAX_RETRIES = 3
    CELERY_TASK_DEFAULT_RETRY_DELAY = 60  # secondes
```

---

## 13.3 Définition et appel des tâches

```python
# app/tasks/email_tasks.py

from app.celery_app import celery
from app.models.incident import Incident
from app.models.user import User
import smtplib
from email.mime.text import MIMEText

@celery.task(
    bind=True,           # Accès à self (l'objet Task)
    max_retries=3,       # 3 tentatives max
    default_retry_delay=60  # 60s entre les tentatives
)
def send_incident_notification(self, incident_id: int, recipient_email: str):
    """
    Envoie un email de confirmation quand un incident est créé.
    S'exécute en arrière-plan, sans bloquer la requête HTTP.
    """
    try:
        # Accès à la base de données (possible grâce à ContextTask)
        incident = Incident.query.get(incident_id)
        if not incident:
            # L'incident n'existe pas, pas besoin de retry
            return {"status": "skipped", "reason": "incident not found"}
        
        # Construction de l'email
        subject = f"[UrbanPulse] Signalement #{incident.id} reçu"
        body = f"""
        Bonjour,
        
        Votre signalement "{incident.title}" a bien été enregistré.
        Numéro de référence : #{incident.id}
        
        Nous vous informerons dès qu'un agent prendra en charge votre demande.
        
        Cordialement,
        L'équipe UrbanPulse
        """
        
        # Envoi réel de l'email
        _send_email(
            to=recipient_email,
            subject=subject,
            body=body
        )
        
        return {"status": "sent", "incident_id": incident_id}
        
    except Exception as exc:
        # Retry automatique avec backoff exponentiel
        raise self.retry(
            exc=exc,
            countdown=60 * (2 ** self.request.retries)  # 60s, 120s, 240s
        )


@celery.task
def process_incident_photo(incident_id: int, photo_path: str):
    """
    Traitement d'image en arrière-plan :
    - Redimensionnement
    - Compression
    - Génération de miniature
    """
    from PIL import Image
    import os
    
    try:
        with Image.open(photo_path) as img:
            # Redimensionner si trop grand (max 1920px)
            max_size = (1920, 1080)
            img.thumbnail(max_size, Image.LANCZOS)
            img.save(photo_path, optimize=True, quality=85)
            
            # Créer une miniature
            thumb_path = photo_path.replace(".", "_thumb.")
            img.thumbnail((400, 300), Image.LANCZOS)
            img.save(thumb_path)
        
        # Mettre à jour le modèle avec le chemin de la miniature
        incident = Incident.query.get(incident_id)
        if incident:
            thumb_filename = os.path.basename(thumb_path)
            incident.photo_thumbnail = thumb_filename
            from app import db
            db.session.commit()
        
        return {"status": "processed", "photo": photo_path}
        
    except Exception as e:
        return {"status": "error", "error": str(e)}


@celery.task
def generate_weekly_report():
    """
    Tâche planifiée : génère un rapport hebdomadaire.
    Configurée avec Celery Beat pour s'exécuter chaque lundi à 8h.
    """
    from datetime import datetime, timedelta
    from app.models.incident import Incident
    
    last_week = datetime.utcnow() - timedelta(days=7)
    
    stats = {
        "new_incidents": Incident.query.filter(
            Incident.created_at >= last_week
        ).count(),
        "resolved_incidents": Incident.query.filter(
            Incident.status == "resolved",
            Incident.resolved_at >= last_week
        ).count(),
        "top_categories": {}
    }
    
    # Envoyer aux admins
    admins = User.query.filter_by(role="admin").all()
    for admin in admins:
        send_report_email.delay(admin.email, stats)
    
    return stats
```

### Planification de tâches (Celery Beat)

```python
# app/config.py

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    "weekly-report": {
        "task": "app.tasks.email_tasks.generate_weekly_report",
        "schedule": crontab(hour=8, minute=0, day_of_week=1),  # Lundi 8h
    },
    "daily-cleanup": {
        "task": "app.tasks.maintenance.cleanup_old_uploads",
        "schedule": crontab(hour=2, minute=30),  # Chaque nuit à 2h30
    }
}
```

### Appel des tâches depuis Flask

```python
# Dans une route Flask

@incidents_bp.route("/create", methods=["POST"])
@login_required
def create():
    form = IncidentForm()
    
    if form.validate_on_submit():
        incident = Incident(
            title=form.title.data,
            # ...
            author_id=current_user.id
        )
        db.session.add(incident)
        db.session.commit()
        
        # Lancer les tâches en arrière-plan — la requête HTTP répond IMMÉDIATEMENT
        # sans attendre que les tâches se terminent
        
        # .delay() = appel asynchrone (le plus courant)
        send_incident_notification.delay(
            incident_id=incident.id,
            recipient_email=current_user.email
        )
        
        # Traitement de photo si uploadée
        if incident.photo_url:
            photo_path = os.path.join(app.root_path, "static/uploads", incident.photo_url)
            process_incident_photo.delay(incident.id, photo_path)
        
        flash("[OK] Signalement créé ! Un email de confirmation vous a été envoyé.", "success")
        return redirect(url_for("incidents.detail", incident_id=incident.id))
    
    return render_template("incidents/create.html", form=form)
```

### Lancement du worker Celery

```bash
# Terminal 1 : Démarrer Redis (pré-requis)
redis-server

# Terminal 2 : Démarrer le worker Celery
celery -A app.celery_app worker --loglevel=info

# Terminal 3 : Démarrer Celery Beat (pour les tâches planifiées)
celery -A app.celery_app beat --loglevel=info

# Terminal 4 : Flower (interface de monitoring)
celery -A app.celery_app flower --port=5555
# Visitez http://localhost:5555
```

---

## [EDIT] Exercice 13.1 — Notifications par email

> **Objectif** : Implémenter les emails de confirmation asynchrones.

1. Installez Redis, Celery, et configurez-les.
2. Créez la tâche `send_incident_notification`.
3. Appelez la tâche dans la route de création d'incident.
4. Démarrez le worker et créez un incident.
5. Vérifiez dans les logs du worker que la tâche s'est exécutée.
6. **Bonus** : Configurez Flask-Mail pour un envoi d'email réel.

---

<a name="chapitre-14"></a>
# [GUIDE] Chapitre 14 — Gestion d'erreurs et Logging

## 14.1 Gestionnaires d'erreurs Flask

```python
# app/errors.py

from flask import jsonify, render_template, request

def register_error_handlers(app):
    """Enregistre tous les gestionnaires d'erreurs."""
    
    def wants_json():
        """Détermine si le client veut du JSON ou du HTML."""
        return (
            request.path.startswith("/api/") or
            request.accept_mimetypes.accept_json and
            not request.accept_mimetypes.accept_html
        )
    
    @app.errorhandler(400)
    def bad_request(error):
        if wants_json():
            return jsonify({
                "error": "Bad Request",
                "message": str(error),
                "status_code": 400
            }), 400
        return render_template("errors/400.html", error=error), 400
    
    @app.errorhandler(401)
    def unauthorized(error):
        if wants_json():
            return jsonify({
                "error": "Unauthorized",
                "message": "Authentification requise.",
                "status_code": 401
            }), 401
        return render_template("errors/401.html"), 401
    
    @app.errorhandler(403)
    def forbidden(error):
        if wants_json():
            return jsonify({
                "error": "Forbidden",
                "message": "Vous n'avez pas les permissions nécessaires.",
                "status_code": 403
            }), 403
        return render_template("errors/403.html"), 403
    
    @app.errorhandler(404)
    def not_found(error):
        if wants_json():
            return jsonify({
                "error": "Not Found",
                "message": "La ressource demandée n'existe pas.",
                "status_code": 404
            }), 404
        return render_template("errors/404.html"), 404
    
    @app.errorhandler(422)
    def unprocessable(error):
        """Erreur de validation (données invalides)."""
        if wants_json():
            return jsonify({
                "error": "Unprocessable Entity",
                "message": "Les données envoyées sont invalides.",
                "status_code": 422
            }), 422
        return render_template("errors/422.html"), 422
    
    @app.errorhandler(429)
    def too_many_requests(error):
        """Rate limit dépassé."""
        return jsonify({
            "error": "Too Many Requests",
            "message": "Trop de requêtes. Réessayez dans quelques minutes.",
            "retry_after": error.retry_after,
            "status_code": 429
        }), 429
    
    @app.errorhandler(500)
    def internal_error(error):
        """Erreur serveur — toujours logger."""
        from app import db
        db.session.rollback()  # Rollback la session en cas d'erreur
        
        app.logger.error(f"Erreur 500 : {error}", exc_info=True)
        
        if wants_json():
            return jsonify({
                "error": "Internal Server Error",
                "message": "Une erreur inattendue s'est produite. Notre équipe a été notifiée.",
                "status_code": 500
            }), 500
        return render_template("errors/500.html"), 500
```

---

## 14.2 Exceptions personnalisées

```python
# app/exceptions.py

class UrbanPulseError(Exception):
    """Exception de base pour UrbanPulse."""
    status_code = 500
    error_code = "INTERNAL_ERROR"
    
    def __init__(self, message=None, status_code=None, payload=None):
        super().__init__()
        self.message = message or "Une erreur s'est produite."
        if status_code:
            self.status_code = status_code
        self.payload = payload
    
    def to_dict(self):
        result = {
            "error": self.error_code,
            "message": self.message,
            "status_code": self.status_code
        }
        if self.payload:
            result["details"] = self.payload
        return result


class ValidationError(UrbanPulseError):
    """Erreur de validation des données."""
    status_code = 422
    error_code = "VALIDATION_ERROR"


class NotFoundError(UrbanPulseError):
    """Ressource non trouvée."""
    status_code = 404
    error_code = "NOT_FOUND"


class PermissionError(UrbanPulseError):
    """Permission insuffisante."""
    status_code = 403
    error_code = "FORBIDDEN"


class ConflictError(UrbanPulseError):
    """Conflit (ex: vote en double)."""
    status_code = 409
    error_code = "CONFLICT"


# Enregistrement dans Flask
def register_custom_exceptions(app):
    @app.errorhandler(UrbanPulseError)
    def handle_custom_error(error):
        response = jsonify(error.to_dict())
        response.status_code = error.status_code
        return response
```

---

## 14.3 Logging structuré

```python
# app/logging_config.py

import logging
import logging.handlers
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    """
    Formateur JSON pour les logs structurés.
    Chaque log est un objet JSON sur une ligne -> facile à parser avec ELK/Grafana.
    """
    
    def format(self, record):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
        
        # Informations de requête HTTP (si disponibles)
        try:
            from flask import request, g
            log_entry["request"] = {
                "method": request.method,
                "path": request.path,
                "remote_addr": request.remote_addr,
                "user_agent": request.user_agent.string[:100]
            }
            if hasattr(g, "current_user_id"):
                log_entry["user_id"] = g.current_user_id
        except RuntimeError:
            pass  # Hors contexte de requête
        
        # Exception si présente
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        
        return json.dumps(log_entry, ensure_ascii=False)


def setup_logging(app):
    """Configure le système de logging de l'application."""
    
    # Niveau de log selon l'environnement
    log_level = logging.DEBUG if app.debug else logging.INFO
    
    # =================== Handler Console ===================
    console_handler = logging.StreamHandler()
    console_handler.setLevel(log_level)
    
    if app.debug:
        # Format lisible en développement
        console_handler.setFormatter(logging.Formatter(
            "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
            datefmt="%H:%M:%S"
        ))
    else:
        # Format JSON en production
        console_handler.setFormatter(JsonFormatter())
    
    # =================== Handler Fichier ===================
    file_handler = logging.handlers.RotatingFileHandler(
        "logs/urbanpulse.log",
        maxBytes=10 * 1024 * 1024,  # 10 Mo max
        backupCount=5,               # Garde 5 fichiers de log
        encoding="utf-8"
    )
    file_handler.setLevel(logging.WARNING)  # Seulement WARNING+ dans les fichiers
    file_handler.setFormatter(JsonFormatter())
    
    # =================== Handler Erreurs ===================
    error_handler = logging.handlers.RotatingFileHandler(
        "logs/errors.log",
        maxBytes=5 * 1024 * 1024,
        backupCount=10
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(JsonFormatter())
    
    # Configuration du logger Flask
    app.logger.setLevel(log_level)
    app.logger.addHandler(console_handler)
    app.logger.addHandler(file_handler)
    app.logger.addHandler(error_handler)
    
    # Logger SQLAlchemy (requêtes SQL)
    if app.debug:
        sql_logger = logging.getLogger("sqlalchemy.engine")
        sql_logger.setLevel(logging.INFO)
        sql_logger.addHandler(console_handler)
    
    app.logger.info("[CITYSCAPE] UrbanPulse démarré", extra={"version": "1.0.0"})
```

### Utilisation du logging dans le code

```python
from flask import current_app

# Dans une route
@incidents_bp.route("/create", methods=["POST"])
@login_required
def create():
    current_app.logger.info(
        "Création d'incident initiée",
        extra={
            "user_id": current_user.id,
            "action": "incident_create"
        }
    )
    
    # ... logique ...
    
    if form.validate_on_submit():
        # ...
        current_app.logger.info(
            f"Incident #{incident.id} créé",
            extra={
                "incident_id": incident.id,
                "category": incident.category,
                "action": "incident_created"
            }
        )
    else:
        current_app.logger.warning(
            "Tentative de création d'incident invalide",
            extra={
                "user_id": current_user.id,
                "errors": form.errors
            }
        )
```

---

## 14.4 Intégration Sentry (monitoring production)

```bash
pip install sentry-sdk[flask]
```

```python
# app/__init__.py

import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
from sentry_sdk.integrations.celery import CeleryIntegration

def create_app(config_name="development"):
    app = Flask(__name__)
    
    # Sentry en production uniquement
    if config_name == "production":
        sentry_sdk.init(
            dsn=os.environ.get("SENTRY_DSN"),
            integrations=[
                FlaskIntegration(),
                SqlalchemyIntegration(),
                CeleryIntegration()
            ],
            traces_sample_rate=0.1,    # 10% des transactions tracées
            environment="production",
            release=os.environ.get("APP_VERSION", "1.0.0"),
            
            # Scrubbing des données sensibles
            before_send=lambda event, hint: scrub_sensitive_data(event)
        )
    
    return app

def scrub_sensitive_data(event):
    """Retire les données sensibles avant l'envoi à Sentry."""
    if "request" in event and "data" in event["request"]:
        sensitive_fields = ["password", "password_hash", "token", "credit_card"]
        for field in sensitive_fields:
            if field in event["request"]["data"]:
                event["request"]["data"][field] = "[Filtered]"
    return event
```

---

## [EDIT] Exercice 14.1 — Gestion d'erreurs professionnelle

> **Objectif** : Ajouter une gestion d'erreurs robuste.

1. Créez `app/errors.py` avec tous les gestionnaires (400, 401, 403, 404, 500).
2. Créez les templates `errors/404.html` et `errors/500.html` (pages d'erreur personnalisées).
3. Ajoutez les exceptions personnalisées et leur handler.
4. Configurez le logging avec rotation de fichiers.
5. Testez : visitez une URL inexistante, une URL d'admin sans permission.
6. **Bonus** : Créez un script qui génère intentionnellement une erreur 500 et vérifiez les logs.

---

<a name="chapitre-15"></a>
# [GUIDE] Chapitre 15 — Tests Unitaires et d'Intégration

## 15.1 Philosophie des tests

**Pourquoi tester ?**
- Confiance lors des refactorings
- Documentation vivante du comportement attendu
- Détection des régressions avant la production
- Meilleure conception (le code testable est souvent mieux conçu)

**Types de tests :**
```
Tests unitaires   -> Tester une fonction/classe isolément (rapides, nombreux)
Tests d'intégration -> Tester l'interaction entre composants (DB, API...)
Tests end-to-end  -> Tester le parcours utilisateur complet (lents, rares)
```

---

## 15.2 Configuration de pytest

```bash
pip install pytest pytest-flask pytest-cov factory-boy faker
```

```python
# tests/conftest.py
# conftest.py est chargé automatiquement par pytest — parfait pour les fixtures partagées

import pytest
from app import create_app, db as _db
from app.models.user import User
from app.models.incident import Incident

@pytest.fixture(scope="session")
def app():
    """
    Crée l'application Flask en mode test pour toute la session.
    scope="session" = créée une seule fois pour tous les tests.
    """
    app = create_app("testing")
    
    # Vérification que la config de test est bien chargée
    assert app.config["TESTING"] == True
    assert "memory" in app.config["SQLALCHEMY_DATABASE_URI"]
    
    with app.app_context():
        yield app


@pytest.fixture(scope="function")
def db(app):
    """
    Base de données fraîche pour chaque test.
    scope="function" = recréée pour CHAQUE test (isolation totale).
    """
    with app.app_context():
        _db.create_all()
        yield _db
        _db.session.remove()
        _db.drop_all()


@pytest.fixture
def client(app):
    """Client HTTP de test — simule un navigateur."""
    return app.test_client()


@pytest.fixture
def runner(app):
    """Runner pour tester les commandes CLI Flask."""
    return app.test_cli_runner()


# ==================== FACTORIES ====================

@pytest.fixture
def make_user(db):
    """Factory pour créer des utilisateurs de test."""
    def _make_user(
        username="testuser",
        email="test@example.com",
        role="citizen",
        password="TestPass123!"
    ):
        user = User(
            username=username,
            email=email,
            role=role
        )
        user.set_password(password)
        db.session.add(user)
        db.session.commit()
        return user
    return _make_user


@pytest.fixture
def citizen(make_user):
    """Utilisateur citoyen de test."""
    return make_user(username="citizen_test", email="citizen@test.com", role="citizen")


@pytest.fixture
def moderator(make_user):
    """Utilisateur modérateur de test."""
    return make_user(username="mod_test", email="mod@test.com", role="moderator")


@pytest.fixture
def admin(make_user):
    """Utilisateur admin de test."""
    return make_user(username="admin_test", email="admin@test.com", role="admin")


@pytest.fixture
def make_incident(db, citizen):
    """Factory pour créer des incidents de test."""
    def _make_incident(
        title="Incident de test - titre suffisamment long",
        description="Description détaillée de l'incident de test pour dépasser le minimum requis.",
        category="voirie",
        status="open",
        author=None
    ):
        incident = Incident(
            title=title,
            description=description,
            category=category,
            status=status,
            author_id=(author or citizen).id
        )
        db.session.add(incident)
        db.session.commit()
        return incident
    return _make_incident


@pytest.fixture
def auth_client(client, citizen):
    """Client HTTP avec un utilisateur connecté."""
    with client.session_transaction() as session:
        # Simuler une session de connexion Flask-Login
        session["_user_id"] = str(citizen.id)
        session["_fresh"] = True
    return client


@pytest.fixture
def auth_headers(citizen, app):
    """Headers JWT pour les tests de l'API."""
    from flask_jwt_extended import create_access_token
    with app.app_context():
        token = create_access_token(identity=citizen.id)
    return {"Authorization": f"Bearer {token}"}
```

---

## 15.3 Tests des modèles (unitaires)

```python
# tests/unit/test_models.py

import pytest
from datetime import datetime
from app.models.user import User
from app.models.incident import Incident

class TestUserModel:
    """Tests unitaires du modèle User."""
    
    def test_create_user(self, db):
        """Test de création d'un utilisateur."""
        user = User(
            username="alice",
            email="alice@test.com",
            role="citizen"
        )
        user.set_password("SecurePass123!")
        db.session.add(user)
        db.session.commit()
        
        assert user.id is not None
        assert user.username == "alice"
        assert user.email == "alice@test.com"
        assert user.role == "citizen"
        assert user.is_active == True
        assert user.created_at is not None
    
    def test_password_hashing(self, db, make_user):
        """Le mot de passe doit être hashé, jamais en clair."""
        user = make_user(password="MonMotDePasse123!")
        
        # Le hash ne doit pas être le mot de passe en clair
        assert user.password_hash != "MonMotDePasse123!"
        
        # La vérification doit fonctionner
        assert user.check_password("MonMotDePasse123!") == True
        assert user.check_password("MauvaisMotDePasse") == False
    
    def test_username_unique(self, db, make_user):
        """Deux utilisateurs ne peuvent pas avoir le même username."""
        make_user(username="alice", email="alice1@test.com")
        
        from sqlalchemy.exc import IntegrityError
        with pytest.raises(IntegrityError):
            make_user(username="alice", email="alice2@test.com")
            db.session.flush()
    
    def test_has_role(self, db, make_user):
        """Test de la vérification de rôle."""
        admin = make_user(username="admin", email="admin@test.com", role="admin")
        citizen = make_user(username="citizen", email="citizen@test.com", role="citizen")
        
        assert admin.has_role("admin") == True
        assert admin.has_role("citizen") == False
        assert admin.has_role("admin", "moderator") == True
        assert citizen.has_role("admin", "moderator") == False
    
    def test_to_dict_excludes_password(self, db, make_user):
        """to_dict ne doit jamais contenir le hash du mot de passe."""
        user = make_user()
        user_dict = user.to_dict()
        
        assert "password_hash" not in user_dict
        assert "password" not in user_dict
        assert "id" in user_dict
        assert "username" in user_dict
        assert "email" in user_dict


class TestIncidentModel:
    """Tests unitaires du modèle Incident."""
    
    def test_create_incident(self, db, citizen):
        """Test de création d'un incident."""
        incident = Incident(
            title="Nid-de-poule dangereux rue Victor Hugo",
            description="Large nid-de-poule d'environ 30cm de diamètre, très dangereux pour les cyclistes.",
            category="voirie",
            author_id=citizen.id
        )
        db.session.add(incident)
        db.session.commit()
        
        assert incident.id is not None
        assert incident.status == "open"
        assert incident.vote_count == 0
        assert incident.created_at is not None
    
    def test_resolve_incident(self, db, make_incident):
        """Test de résolution d'un incident."""
        incident = make_incident()
        assert incident.status == "open"
        assert incident.resolved_at is None
        
        incident.resolve()
        db.session.commit()
        
        assert incident.status == "resolved"
        assert incident.resolved_at is not None
    
    def test_upvote(self, db, make_incident):
        """Test du système de vote."""
        incident = make_incident()
        initial_count = incident.vote_count
        
        incident.upvote()
        db.session.commit()
        
        assert incident.vote_count == initial_count + 1
    
    def test_author_relationship(self, db, make_incident, citizen):
        """Test de la relation avec l'auteur."""
        incident = make_incident(author=citizen)
        
        # Rechargement depuis la base
        fresh_incident = Incident.query.get(incident.id)
        
        assert fresh_incident.author is not None
        assert fresh_incident.author.id == citizen.id
        assert fresh_incident.author.username == citizen.username
```

---

## 15.4 Tests des routes (intégration)

```python
# tests/integration/test_incidents.py

import json
import pytest


class TestIncidentRoutes:
    """Tests d'intégration pour les routes d'incidents."""
    
    def test_list_incidents_empty(self, client, db):
        """La liste vide retourne 200."""
        response = client.get("/incidents")
        assert response.status_code == 200
        assert b"Aucun signalement" in response.data
    
    def test_list_incidents_with_data(self, client, make_incident):
        """La liste affiche les incidents existants."""
        make_incident(title="Incident visible dans la liste - titre complet")
        
        response = client.get("/incidents")
        assert response.status_code == 200
        assert b"Incident visible dans la liste" in response.data
    
    def test_create_incident_requires_login(self, client):
        """La création d'incident nécessite d'être connecté."""
        response = client.get("/incidents/create")
        # Redirigé vers la page de login
        assert response.status_code == 302
        assert "/auth/login" in response.location
    
    def test_create_incident_authenticated(self, auth_client, db, citizen):
        """Un utilisateur connecté peut créer un incident."""
        response = auth_client.post("/incidents/create", data={
            "title": "Titre du signalement suffisamment long",
            "description": "Description très détaillée du problème signalé qui dépasse bien le minimum requis.",
            "category": "voirie",
            "csrf_token": "test"  # En mode test, CSRF est désactivé
        }, follow_redirects=True)
        
        assert response.status_code == 200
        assert b"soumis avec succ" in response.data  # "soumis avec succès"
        
        # Vérifier en base
        from app.models.incident import Incident
        incident = Incident.query.filter_by(author_id=citizen.id).first()
        assert incident is not None
        assert incident.title == "Titre du signalement suffisamment long"
    
    def test_create_incident_validation_error(self, auth_client):
        """La validation rejette les données invalides."""
        response = auth_client.post("/incidents/create", data={
            "title": "Court",  # Trop court
            "description": "Trop court",
            "category": "voirie"
        })
        
        assert response.status_code in [200, 400]
        assert b"au moins 10" in response.data  # Message d'erreur
    
    def test_incident_detail(self, client, make_incident):
        """La page de détail affiche les informations de l'incident."""
        incident = make_incident(
            title="Incident de test pour détail - titre suffisamment long"
        )
        
        response = client.get(f"/incidents/{incident.id}")
        assert response.status_code == 200
        assert b"Incident de test pour d" in response.data
    
    def test_incident_detail_404(self, client):
        """Une ID inexistante retourne 404."""
        response = client.get("/incidents/99999")
        assert response.status_code == 404
    
    def test_vote_requires_auth(self, client, make_incident):
        """Le vote nécessite d'être connecté."""
        incident = make_incident()
        response = client.post(f"/incidents/{incident.id}/vote")
        assert response.status_code == 302


class TestIncidentAPI:
    """Tests d'intégration pour l'API REST."""
    
    def test_get_incidents_empty(self, client, db):
        """L'API retourne une liste vide en JSON."""
        response = client.get("/api/v1/incidents/")
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["total"] == 0
        assert data["incidents"] == []
    
    def test_get_incidents_pagination(self, client, make_incident):
        """L'API pagine correctement les résultats."""
        # Créer 15 incidents
        for i in range(15):
            make_incident(title=f"Incident numéro {i+1:02d} - titre complet pour test")
        
        response = client.get("/api/v1/incidents/?per_page=5&page=1")
        data = json.loads(response.data)
        
        assert data["total"] == 15
        assert len(data["incidents"]) == 5
        assert data["pages"] == 3
    
    def test_create_incident_api(self, client, auth_headers, db):
        """L'API crée un incident avec auth JWT."""
        payload = {
            "title": "Nid-de-poule rue test suffisamment long",
            "description": "Description longue et détaillée du problème signalé via l'API REST.",
            "category": "voirie"
        }
        
        response = client.post(
            "/api/v1/incidents/",
            data=json.dumps(payload),
            content_type="application/json",
            headers=auth_headers
        )
        
        assert response.status_code == 201
        data = json.loads(response.data)
        assert data["title"] == payload["title"]
        assert data["status"] == "open"
        assert data["id"] is not None
    
    def test_create_incident_without_auth(self, client):
        """L'API rejette la création sans token JWT."""
        payload = {"title": "test", "description": "test"}
        response = client.post(
            "/api/v1/incidents/",
            data=json.dumps(payload),
            content_type="application/json"
        )
        assert response.status_code == 401
    
    def test_vote_api(self, client, make_incident, auth_headers):
        """L'API enregistre un vote."""
        incident = make_incident()
        initial_votes = incident.vote_count
        
        response = client.post(
            f"/api/v1/incidents/{incident.id}/vote",
            headers=auth_headers
        )
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["vote_count"] == initial_votes + 1
    
    def test_double_vote_rejected(self, client, make_incident, auth_headers):
        """L'API rejette un double vote."""
        incident = make_incident()
        
        # Premier vote
        client.post(f"/api/v1/incidents/{incident.id}/vote", headers=auth_headers)
        
        # Deuxième vote — doit échouer
        response = client.post(f"/api/v1/incidents/{incident.id}/vote", headers=auth_headers)
        assert response.status_code == 409


class TestAuthAPI:
    """Tests pour les routes d'authentification."""
    
    def test_register(self, client, db):
        """Inscription d'un nouvel utilisateur."""
        response = client.post("/auth/register", data={
            "username": "newuser",
            "email": "new@test.com",
            "password": "SecurePass123!",
            "password_confirm": "SecurePass123!",
            "accept_terms": True
        }, follow_redirects=True)
        
        assert response.status_code == 200
        
        from app.models.user import User
        user = User.query.filter_by(email="new@test.com").first()
        assert user is not None
        assert user.username == "newuser"
    
    def test_login_success(self, client, make_user):
        """Connexion avec des identifiants valides."""
        make_user(username="logintest", email="login@test.com", password="TestPass123!")
        
        response = client.post("/auth/login", data={
            "email": "login@test.com",
            "password": "TestPass123!"
        }, follow_redirects=True)
        
        assert response.status_code == 200
        assert b"Bon retour" in response.data
    
    def test_login_wrong_password(self, client, make_user):
        """Connexion avec mauvais mot de passe."""
        make_user(username="logintest2", email="login2@test.com", password="CorrectPass!")
        
        response = client.post("/auth/login", data={
            "email": "login2@test.com",
            "password": "WrongPassword!"
        })
        
        # Pas de redirection, on reste sur la page de login
        assert response.status_code == 200
        assert b"incorrect" in response.data
```

---

## 15.5 Coverage et rapport de tests

```bash
# Lancer les tests avec couverture de code
pytest --cov=app --cov-report=html --cov-report=term-missing tests/

# Le rapport HTML est généré dans htmlcov/index.html
# Ouvrez-le dans votre navigateur pour visualiser quelles lignes ne sont pas testées

# Exiger un taux minimum de couverture
pytest --cov=app --cov-fail-under=80 tests/

# Structure recommandée des tests
tests/
├── conftest.py            # Fixtures partagées
├── unit/
│   ├── test_models.py     # Tests des modèles
│   ├── test_forms.py      # Tests des formulaires
│   └── test_schemas.py    # Tests des schemas Marshmallow
├── integration/
│   ├── test_incidents.py  # Tests des routes incidents
│   ├── test_auth.py       # Tests des routes auth
│   └── test_api.py        # Tests de l'API REST
└── e2e/
    └── test_user_journey.py  # Tests de parcours complet
```

```ini
# pytest.ini — Configuration pytest
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# Marqueurs personnalisés
markers =
    slow: Tests lents (désactivés avec -m "not slow")
    integration: Tests d'intégration
    unit: Tests unitaires

# Options par défaut
addopts = -v --tb=short
```

---

## [EDIT] Exercice 15.1 — Suite de tests UrbanPulse

> **Objectif** : Atteindre 80% de couverture de code.

1. Créez `tests/conftest.py` avec toutes les fixtures.
2. Écrivez les tests unitaires pour `User` et `Incident`.
3. Écrivez les tests d'intégration pour les routes auth (register, login, logout).
4. Écrivez les tests pour l'API REST (CRUD incidents avec JWT).
5. Lancez `pytest --cov=app --cov-report=html` et examinez le rapport.
6. Identifiez les parties non testées et ajoutez des tests.
7. **Objectif** : 80%+ de couverture.

---

<a name="chapitre-16"></a>
# [GUIDE] Chapitre 16 — Linting, Typage et Qualité de Code

## 16.1 Black — Formateur automatique

```bash
pip install black
# Formatter le code automatiquement
black app/ tests/
# Vérifier sans modifier (pour la CI)
black --check app/ tests/
```

```toml
# pyproject.toml
[tool.black]
line-length = 88
target-version = ["py311"]
include = '\.pyi?$'
exclude = '''
/(
    migrations
  | venv
  | __pycache__
)/
'''
```

---

## 16.2 Flake8 — Linter

```bash
pip install flake8 flake8-bugbear flake8-docstrings
# Vérifier le code
flake8 app/ tests/
```

```ini
# setup.cfg
[flake8]
max-line-length = 88
extend-ignore = E203, W503, D100, D101, D102
exclude =
    .git,
    __pycache__,
    migrations,
    venv
per-file-ignores =
    tests/*: D103, D101
```

---

## 16.3 MyPy — Typage statique

```bash
pip install mypy
```

```python
# Exemple de code typé pour UrbanPulse

from typing import Optional, List, Dict, Any
from datetime import datetime

def get_incidents_by_status(
    status: str,
    page: int = 1,
    per_page: int = 10
) -> Dict[str, Any]:
    """
    Récupère les incidents filtrés par statut.
    
    Args:
        status: Le statut à filtrer ("open", "resolved", etc.)
        page: Numéro de page (1-indexed)
        per_page: Nombre d'éléments par page
    
    Returns:
        Dictionnaire avec les incidents et les métadonnées de pagination.
    """
    pagination = Incident.query.filter_by(
        status=status
    ).paginate(page=page, per_page=per_page)
    
    return {
        "incidents": [i.to_dict() for i in pagination.items],
        "total": pagination.total,
        "page": pagination.page,
        "pages": pagination.pages
    }


def find_user_by_email(email: str) -> Optional[User]:
    """Trouve un utilisateur par email ou retourne None."""
    return User.query.filter_by(email=email.lower()).first()


class IncidentService:
    """Service métier pour la gestion des incidents."""
    
    @staticmethod
    def create(
        title: str,
        description: str,
        author_id: int,
        category: str = "autre",
        address: Optional[str] = None,
        latitude: Optional[float] = None,
        longitude: Optional[float] = None
    ) -> Incident:
        """Crée un nouvel incident avec validation."""
        if len(title) < 10:
            raise ValueError("Le titre doit faire au moins 10 caractères.")
        
        incident = Incident(
            title=title,
            description=description,
            category=category,
            address=address,
            latitude=latitude,
            longitude=longitude,
            author_id=author_id
        )
        db.session.add(incident)
        db.session.commit()
        return incident
    
    @staticmethod
    def get_top_voted(limit: int = 10) -> List[Incident]:
        """Retourne les incidents les plus votés."""
        return (
            Incident.query
            .filter(Incident.status != "rejected")
            .order_by(Incident.vote_count.desc())
            .limit(limit)
            .all()
        )
```

```ini
# mypy.ini
[mypy]
python_version = 3.11
warn_return_any = True
warn_unused_configs = True
ignore_missing_imports = True

[mypy-flask.*]
ignore_missing_imports = True

[mypy-sqlalchemy.*]
ignore_missing_imports = True
```

---

## 16.4 Pre-commit hooks — Automatiser la qualité

```bash
pip install pre-commit
```

```yaml
# .pre-commit-config.yaml
repos:
  - repo: https://github.com/psf/black
    rev: 24.1.1
    hooks:
      - id: black
        language_version: python3.11

  - repo: https://github.com/pycqa/flake8
    rev: 7.0.0
    hooks:
      - id: flake8
        args: [--max-line-length=88]

  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.5.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-added-large-files
        args: [--maxkb=1000]
      - id: check-merge-conflict
      - id: detect-private-key

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.8.0
    hooks:
      - id: mypy
        args: [--ignore-missing-imports]
```

```bash
# Installer les hooks (une seule fois par projet)
pre-commit install

# Les hooks s'exécutent maintenant à chaque git commit
# Pour lancer manuellement sur tous les fichiers :
pre-commit run --all-files
```

---

## [EDIT] Exercice 16.1 — Pipeline de qualité

> **Objectif** : Mettre en place la chaîne qualité complète.

1. Installez Black, Flake8, MyPy, et pre-commit.
2. Formatez tout le code avec `black app/ tests/`.
3. Corrigez les avertissements Flake8 un par un.
4. Ajoutez les annotations de type aux 5 fonctions les plus utilisées.
5. Configurez les pre-commit hooks.
6. Faites un commit — les hooks s'exécutent-ils ?
7. Introduisez intentionnellement une erreur de style (ligne trop longue) — le hook la bloque-t-il ?

---

## [TROPHEE] Récapitulatif Module 3

Vous maîtrisez maintenant :

[OK] L'architecture REST avec conventions d'URL
[OK] Flask-RESTX avec documentation Swagger automatique
[OK] La sérialisation avec Marshmallow (schemas, validation, nested)
[OK] La pagination avancée avec filtres et recherche
[OK] Celery pour les tâches en arrière-plan (emails, traitements)
[OK] La planification de tâches avec Celery Beat
[OK] La gestion d'erreurs multicouche (HTTP, custom exceptions, logging)
[OK] Le logging structuré JSON et l'intégration Sentry
[OK] Les tests avec pytest (unitaires + intégration)
[OK] La couverture de code avec coverage
[OK] Le formatage, linting, typage et pre-commit hooks

### [PACKAGE] État du projet UrbanPulse à la fin du Module 3

```
urbanpulse/
├── app/
│   ├── api/
│   │   ├── __init__.py      [OK] Flask-RESTX config
│   │   ├── incidents.py     [OK] API REST avec Swagger
│   │   └── auth.py          [OK] Auth API JWT
│   ├── schemas/
│   │   ├── incident.py      [OK] Marshmallow schemas
│   │   └── user.py          [OK]
│   ├── tasks/
│   │   ├── email_tasks.py   [OK] Celery tasks
│   │   └── maintenance.py   [OK] Tâches planifiées
│   ├── errors.py            [OK] Gestionnaires d'erreurs
│   └── logging_config.py    [OK] Logging structuré
├── tests/
│   ├── conftest.py          [OK] Fixtures pytest
│   ├── unit/
│   │   └── test_models.py   [OK]
│   └── integration/
│       ├── test_incidents.py [OK]
│       └── test_auth.py      [OK]
├── logs/                    [OK] Dossier logs (gitignore)
├── .pre-commit-config.yaml  [OK]
├── pyproject.toml           [OK] Config Black
├── setup.cfg                [OK] Config Flake8
└── mypy.ini                 [OK]
```

**-> Module 4 : Déploiement, Performance, Architecture et Projets**

# [PYTHON] Formation Flask — Module 4
## Déploiement, Performance & Architecture Professionnelle
### Parties VII & VIII — Chapitres 17 à 22

---

> [OBJECTIF] **UrbanPulse : De l'app locale à la production**
> Votre application fonctionne parfaitement en local. Ce module vous apprend à
> la déployer de façon robuste, à l'optimiser pour la performance,
> et à l'architecturer pour qu'elle puisse évoluer.

---

## [WORLD_MAP] Table des matières

- [Chapitre 17 — Serveurs et WSGI](#chapitre-17)
- [Chapitre 18 — Dockerisation](#chapitre-18)
- [Chapitre 19 — CI/CD](#chapitre-19)
- [Chapitre 20 — Performance](#chapitre-20)
- [Chapitre 21 — Observabilité](#chapitre-21)
- [Chapitre 22 — Architecture et Modularité](#chapitre-22)

---

<a name="chapitre-17"></a>
# [GUIDE] Chapitre 17 — Serveurs et WSGI

## 17.1 Pourquoi pas le serveur de développement Flask en production ?

Le serveur de développement Flask (Werkzeug) est :
- **Single-threaded** : gère une seule requête à la fois
- **Non sécurisé** : pas conçu pour être exposé au public
- **Instable** : peut planter sans récupération automatique
- **Lent** : pas optimisé pour les charges importantes

En production, on utilise des serveurs **WSGI** (Web Server Gateway Interface) — un standard Python qui définit comment un serveur web communique avec une application Python.

## 17.2 Gunicorn — Le serveur de production standard

```bash
pip install gunicorn
```

### Lancement minimal

```bash
# Format : gunicorn module:variable
# module = run (le fichier run.py)
# variable = app (l'objet Flask créé dans run.py)
gunicorn run:app

# Sur une adresse et port spécifiques
gunicorn run:app --bind 0.0.0.0:8000

# Avec plusieurs workers
gunicorn run:app --bind 0.0.0.0:8000 --workers 4
```

### Configuration avancée de Gunicorn

```python
# gunicorn.conf.py — Configuration complète

import multiprocessing
import os

# ==================== Adresse de liaison ====================
bind = "0.0.0.0:8000"

# ==================== Workers ====================
# Règle empirique : 2 * nb_CPUs + 1
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "sync"    # "sync" pour SQLAlchemy classique
                          # "gevent" ou "uvicorn.workers.UvicornWorker" pour async

# Threads par worker (utile avec sync pour les I/O)
threads = 2

# ==================== Timeouts ====================
timeout = 30          # Worker tué s'il ne répond pas en 30s
graceful_timeout = 10  # Temps pour terminer les requêtes en cours
keepalive = 5          # Connexions persistantes (secondes)

# ==================== Redémarrage automatique ====================
max_requests = 1000          # Worker redémarré après 1000 requêtes (évite les fuites mémoire)
max_requests_jitter = 100    # Aléatoire pour éviter que tous redémarrent en même temps
preload_app = True           # Charge l'app avant de forker les workers (économise la RAM)

# ==================== Logging ====================
accesslog = "/var/log/urbanpulse/access.log"
errorlog = "/var/log/urbanpulse/error.log"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# ==================== Process ====================
pidfile = "/var/run/urbanpulse/gunicorn.pid"
user = "urbanpulse"
group = "urbanpulse"

# ==================== Hooks ====================
def on_starting(server):
    """Appelé quand Gunicorn démarre."""
    print(f"[RAPIDE] UrbanPulse démarre avec {workers} workers")

def worker_exit(server, worker):
    """Appelé quand un worker se termine."""
    print(f"Worker {worker.pid} terminé")
    from app import db
    db.engine.dispose()  # Fermer les connexions DB proprement
```

```bash
# Lancer avec le fichier de config
gunicorn -c gunicorn.conf.py run:app

# Arrêt gracieux (attend la fin des requêtes en cours)
kill -TERM $(cat /var/run/urbanpulse/gunicorn.pid)

# Rechargement à chaud (sans coupure de service)
kill -HUP $(cat /var/run/urbanpulse/gunicorn.pid)
```

---

## 17.3 Nginx — Reverse Proxy

Nginx se place **devant** Gunicorn et gère :
- Le SSL/HTTPS
- Les fichiers statiques (bien plus efficace que Python)
- Le load balancing si plusieurs serveurs
- La limitation de débit
- La compression gzip

```nginx
# /etc/nginx/sites-available/urbanpulse

# Redirection HTTP -> HTTPS
server {
    listen 80;
    server_name urbanpulse.example.com www.urbanpulse.example.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name urbanpulse.example.com www.urbanpulse.example.com;
    
    # ==================== SSL ====================
    ssl_certificate /etc/letsencrypt/live/urbanpulse.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/urbanpulse.example.com/privkey.pem;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers ECDHE-RSA-AES256-GCM-SHA512:DHE-RSA-AES256-GCM-SHA512;
    ssl_prefer_server_ciphers off;
    ssl_session_cache shared:SSL:10m;
    
    # ==================== Headers de sécurité ====================
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header X-Content-Type-Options nosniff;
    add_header X-Frame-Options SAMEORIGIN;
    add_header X-XSS-Protection "1; mode=block";
    
    # ==================== Fichiers statiques ====================
    # Servis directement par Nginx — BEAUCOUP plus rapide que Flask
    location /static {
        alias /var/www/urbanpulse/app/static;
        expires 1y;                    # Cache navigateur 1 an
        add_header Cache-Control "public, immutable";
        gzip_static on;               # Servir .gz si disponible
    }
    
    # Uploads (photos des incidents)
    location /static/uploads {
        alias /var/www/urbanpulse/app/static/uploads;
        expires 30d;
        
        # Sécurité : empêcher l'exécution de scripts dans le dossier uploads
        location ~* \.(php|py|sh|cgi)$ {
            deny all;
        }
    }
    
    # ==================== Proxy vers Gunicorn ====================
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # Timeouts
        proxy_connect_timeout 10s;
        proxy_send_timeout 30s;
        proxy_read_timeout 30s;
        
        # Taille des uploads (photos d'incidents)
        client_max_body_size 10M;
    }
    
    # ==================== Compression ====================
    gzip on;
    gzip_vary on;
    gzip_types text/plain text/css application/json application/javascript;
    gzip_min_length 1000;
    
    # ==================== Logs ====================
    access_log /var/log/nginx/urbanpulse_access.log;
    error_log /var/log/nginx/urbanpulse_error.log;
}
```

```bash
# Activer le site
sudo ln -s /etc/nginx/sites-available/urbanpulse /etc/nginx/sites-enabled/
sudo nginx -t        # Vérifier la syntaxe
sudo nginx -s reload # Recharger la configuration
```

---

## 17.4 Systemd — Gérer Gunicorn comme un service

```ini
# /etc/systemd/system/urbanpulse.service

[Unit]
Description=UrbanPulse Flask Application
After=network.target postgresql.service redis.service

[Service]
User=urbanpulse
Group=urbanpulse
WorkingDirectory=/var/www/urbanpulse

# Variables d'environnement
EnvironmentFile=/var/www/urbanpulse/.env

# Lancement de Gunicorn
ExecStart=/var/www/urbanpulse/venv/bin/gunicorn \
    -c gunicorn.conf.py \
    run:app

# Redémarrage automatique
Restart=on-failure
RestartSec=5

# Limits
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target
```

```bash
# Activer et démarrer le service
sudo systemctl daemon-reload
sudo systemctl enable urbanpulse
sudo systemctl start urbanpulse

# Commandes utiles
sudo systemctl status urbanpulse
sudo journalctl -u urbanpulse -f  # Voir les logs en temps réel
```

---

## [EDIT] Exercice 17.1 — Déploiement local simulé

> **Objectif** : Déployer UrbanPulse avec Gunicorn.

1. Créez `gunicorn.conf.py` avec 2 workers (pour votre machine locale).
2. Lancez Gunicorn : `gunicorn -c gunicorn.conf.py run:app`
3. Visitez `http://localhost:8000` — l'application répond ?
4. Ouvrez 2 onglets et faites des requêtes simultanées — observez les workers dans les logs.
5. **Bonus** : Installez Nginx sur votre machine et configurez-le comme reverse proxy.

---

<a name="chapitre-18"></a>
# [GUIDE] Chapitre 18 — Dockerisation

## 18.1 Pourquoi Docker ?

Docker crée des **conteneurs** — des environnements isolés et reproductibles qui contiennent l'application et toutes ses dépendances. "Ça marche sur ma machine" disparaît.

**Avantages :**
- Environnement identique en développement, test et production
- Déploiement simplifié et reproductible
- Isolation des composants (app, DB, Redis...)
- Scalabilité facilitée

---

## 18.2 Dockerfile pour UrbanPulse

```dockerfile
# Dockerfile

# ==================== STAGE 1 : Builder ====================
# On utilise un stage séparé pour installer les dépendances
# Cela évite d'avoir pip et les outils de build dans l'image finale
FROM python:3.12-slim AS builder

WORKDIR /build

# Optimisation : copier requirements en premier
# Si requirements.txt ne change pas, Docker utilise le cache
COPY requirements.txt .

# Installer dans un dossier dédié (pas dans le système)
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt


# ==================== STAGE 2 : Production ====================
FROM python:3.12-slim AS production

# Métadonnées de l'image
LABEL maintainer="urbanpulse-team@example.com"
LABEL version="1.0.0"
LABEL description="UrbanPulse - Plateforme citoyenne"

# Variables d'environnement
ENV PYTHONUNBUFFERED=1          # Logs non bufferisés (apparaissent immédiatement)
ENV PYTHONDONTWRITEBYTECODE=1   # Pas de fichiers .pyc
ENV FLASK_ENV=production
ENV PORT=8000

# Créer un utilisateur non-root pour la sécurité
# [ATTENTION] Ne jamais lancer des conteneurs en root !
RUN groupadd -r urbanpulse && \
    useradd -r -g urbanpulse -m -d /home/urbanpulse urbanpulse

WORKDIR /app

# Copier les dépendances installées depuis le stage builder
COPY --from=builder /install /usr/local

# Copier le code de l'application
COPY --chown=urbanpulse:urbanpulse . .

# Créer les dossiers nécessaires avec les bonnes permissions
RUN mkdir -p logs app/static/uploads && \
    chown -R urbanpulse:urbanpulse /app

# Basculer vers l'utilisateur non-root
USER urbanpulse

# Exposer le port (documentation — ne publie pas réellement)
EXPOSE 8000

# Health check — Docker peut vérifier que l'app répond
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

# Commande de démarrage
CMD ["gunicorn", "run:app", \
     "--bind", "0.0.0.0:8000", \
     "--workers", "4", \
     "--timeout", "30", \
     "--access-logfile", "-", \
     "--error-logfile", "-"]
```

```
# .dockerignore — Fichiers à ne PAS copier dans l'image Docker
venv/
__pycache__/
*.pyc
*.pyo
.git/
.env
*.md
tests/
docs/
.pytest_cache/
htmlcov/
*.db
logs/
app/static/uploads/
```

---

## 18.3 Docker Compose — Orchestrer tous les services

```yaml
# docker-compose.yml — Développement

version: "3.9"

services:
  
  # ==================== Application Flask ====================
  web:
    build:
      context: .
      target: production    # Utilise le stage production du Dockerfile multi-stage
    ports:
      - "8000:8000"
    environment:
      - FLASK_ENV=development
      - DATABASE_URL=postgresql://urbanpulse:password@db:5432/urbanpulse
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=${SECRET_KEY:-dev-secret-key}
    volumes:
      - ./app/static/uploads:/app/app/static/uploads  # Uploads persistants
      - ./logs:/app/logs                               # Logs persistants
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    restart: unless-stopped
    networks:
      - urbanpulse-network
  
  # ==================== Base de données PostgreSQL ====================
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: urbanpulse
      POSTGRES_USER: urbanpulse
      POSTGRES_PASSWORD: password
    volumes:

# [PYTHON] Formation Flask — Module 4 (suite)
## Docker Compose, CI/CD, Performance & Architecture

---

## 18.3 Docker Compose (suite)

```yaml
# docker-compose.yml (suite)

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: urbanpulse
      POSTGRES_USER: urbanpulse
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U urbanpulse"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - urbanpulse-network

  # ==================== Redis ====================
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - urbanpulse-network

  # ==================== Celery Worker ====================
  celery_worker:
    build:
      context: .
      target: production
    command: celery -A app.celery_app worker --loglevel=info --concurrency=2
    environment:
      - DATABASE_URL=postgresql://urbanpulse:password@db:5432/urbanpulse
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis
    volumes:
      - ./app/static/uploads:/app/app/static/uploads
    restart: unless-stopped
    networks:
      - urbanpulse-network

  # ==================== Celery Beat (planificateur) ====================
  celery_beat:
    build:
      context: .
      target: production
    command: celery -A app.celery_app beat --loglevel=info --schedule=/tmp/celerybeat-schedule
    environment:
      - DATABASE_URL=postgresql://urbanpulse:password@db:5432/urbanpulse
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    restart: unless-stopped
    networks:
      - urbanpulse-network

  # ==================== Flower (monitoring Celery) ====================
  flower:
    build:
      context: .
      target: production
    command: celery -A app.celery_app flower --port=5555
    ports:
      - "5555:5555"
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    networks:
      - urbanpulse-network

  # ==================== Nginx ====================
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/conf.d/default.conf:ro
      - ./app/static:/var/www/static:ro
      - ./nginx/ssl:/etc/nginx/ssl:ro
    depends_on:
      - web
    restart: unless-stopped
    networks:
      - urbanpulse-network

# ==================== Volumes persistants ====================
volumes:
  postgres_data:
  redis_data:

# ==================== Réseau isolé ====================
networks:
  urbanpulse-network:
    driver: bridge
```

```yaml
# docker-compose.override.yml — Overrides pour le développement local

version: "3.9"

services:
  web:
    build:
      target: production
    command: flask run --host=0.0.0.0 --port=8000 --debugger --reload
    volumes:
      - .:/app          # Monte le code local pour le rechargement automatique
    environment:
      - FLASK_ENV=development
      - FLASK_DEBUG=1
      - SQLALCHEMY_ECHO=true
    ports:
      - "8000:8000"
```

### Commandes Docker Compose essentielles

```bash
# Construire les images
docker-compose build

# Démarrer tous les services
docker-compose up -d

# Voir les logs en temps réel
docker-compose logs -f web
docker-compose logs -f celery_worker

# Exécuter les migrations
docker-compose exec web flask db upgrade

# Ouvrir un shell dans le conteneur web
docker-compose exec web bash

# Arrêter tous les services
docker-compose down

# Arrêter ET supprimer les volumes (réinitialise la DB !)
docker-compose down -v

# Voir l'état des services
docker-compose ps

# Redémarrer un seul service
docker-compose restart web
```

---

## 18.4 Variables d'environnement avec Docker

```bash
# .env.docker — Variables pour Docker Compose
SECRET_KEY=super-longue-cle-secrete-generee-avec-openssl-rand-hex-32
DATABASE_URL=postgresql://urbanpulse:strongpassword@db:5432/urbanpulse
REDIS_URL=redis://redis:6379/0
JWT_SECRET_KEY=une-autre-cle-jwt-tres-secrete
FLASK_ENV=production
SENTRY_DSN=https://key@sentry.io/project-id
MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USERNAME=noreply@urbanpulse.example.com
MAIL_PASSWORD=app-password-gmail
```

```bash
# Générer une clé secrète robuste
python -c "import secrets; print(secrets.token_hex(32))"
# ou
openssl rand -hex 32
```

---

## [EDIT] Exercice 18.1 — Dockerisation complète

> **Objectif** : Faire tourner UrbanPulse dans Docker.

1. Créez le `Dockerfile` multi-stage.
2. Créez `.dockerignore`.
3. Créez `docker-compose.yml` avec Flask + PostgreSQL + Redis.
4. Lancez `docker-compose up -d`.
5. Exécutez les migrations : `docker-compose exec web flask db upgrade`.
6. Visitez `http://localhost:8000` — l'app tourne dans Docker ?
7. **Bonus** : Ajoutez le service Celery worker et testez un envoi d'email.

---

<a name="chapitre-19"></a>
# [GUIDE] Chapitre 19 — CI/CD avec GitHub Actions

## 19.1 Qu'est-ce que le CI/CD ?

**CI (Continuous Integration) :** À chaque push, des tests automatiques vérifient que le code ne casse rien.

**CD (Continuous Deployment) :** Si les tests passent, le code est déployé automatiquement en production.

```
Developer pushes code
        │
        [BLACK_DOWN-POINTING_TRIANGLE]
   GitHub Actions
        │
   ┌────┴────┐
   │  CI     │
   │ ─────── │
   │ Lint    │  <- black --check, flake8
   │ Tests   │  <- pytest --cov
   │ Build   │  <- docker build
   └────┬────┘
        │ (si tout passe)
        [BLACK_DOWN-POINTING_TRIANGLE]
   ┌────┴────┐
   │  CD     │
   │ ─────── │
   │ Push    │  <- docker push
   │ Deploy  │  <- ssh + docker-compose pull + up
   └─────────┘
```

---

## 19.2 Pipeline GitHub Actions complet

```yaml
# .github/workflows/ci-cd.yml

name: UrbanPulse CI/CD

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}/urbanpulse

jobs:

  # ==================== JOB 1 : Qualité du code ====================
  lint:
    name: [RECHERCHE] Linting & Formatting
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dev dependencies
        run: |
          pip install black flake8 mypy

      - name: Check formatting with Black
        run: black --check --diff app/ tests/

      - name: Lint with Flake8
        run: flake8 app/ tests/ --max-line-length=88

      - name: Type checking with MyPy
        run: mypy app/ --ignore-missing-imports
        continue-on-error: true   # MyPy non bloquant pour l'instant


  # ==================== JOB 2 : Tests ====================
  test:
    name: [TEST] Tests
    runs-on: ubuntu-latest
    needs: lint    # Lance seulement si lint passe

    services:
      # Base de données PostgreSQL pour les tests
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_DB: urbanpulse_test
          POSTGRES_USER: test_user
          POSTGRES_PASSWORD: test_pass
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    env:
      DATABASE_URL: postgresql://test_user:test_pass@localhost:5432/urbanpulse_test
      REDIS_URL: redis://localhost:6379/0
      SECRET_KEY: ci-test-secret-key
      FLASK_ENV: testing

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-flask pytest-cov

      - name: Run migrations
        run: flask db upgrade

      - name: Run tests with coverage
        run: |
          pytest tests/ \
            --cov=app \
            --cov-report=xml \
            --cov-report=term-missing \
            --cov-fail-under=75 \
            -v

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v4
        with:
          file: ./coverage.xml
          fail_ci_if_error: false


  # ==================== JOB 3 : Build Docker ====================
  build:
    name: [DOCKER] Docker Build
    runs-on: ubuntu-latest
    needs: test
    if: github.event_name == 'push'

    outputs:
      image_tag: ${{ steps.meta.outputs.tags }}
      image_digest: ${{ steps.build.outputs.digest }}

    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Login to GitHub Container Registry
        uses: docker/login-action@v3
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Extract metadata (tags, labels)
        id: meta
        uses: docker/metadata-action@v5
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
          tags: |
            type=ref,event=branch
            type=sha,prefix={{branch}}-
            type=semver,pattern={{version}}
            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}

      - name: Build and push Docker image
        id: build
        uses: docker/build-push-action@v5
        with:
          context: .
          target: production
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha        # Cache GitHub Actions
          cache-to: type=gha,mode=max


  # ==================== JOB 4 : Déploiement en production ====================
  deploy:
    name: [RAPIDE] Deploy to Production
    runs-on: ubuntu-latest
    needs: build
    if: github.ref == 'refs/heads/main'
    environment: production    # Protège avec des approbations manuelles

    steps:
      - name: Deploy via SSH
        uses: appleboy/ssh-action@v1.0.0
        with:
          host: ${{ secrets.PROD_SERVER_HOST }}
          username: ${{ secrets.PROD_SERVER_USER }}
          key: ${{ secrets.PROD_SSH_KEY }}
          script: |
            cd /var/www/urbanpulse
            
            # Tirer la nouvelle image
            docker-compose pull web celery_worker celery_beat
            
            # Appliquer les migrations sans downtime
            docker-compose run --rm web flask db upgrade
            
            # Redémarrer les services
            docker-compose up -d --no-deps web celery_worker celery_beat
            
            # Vérifier que l'app répond
            sleep 10
            curl -f http://localhost:8000/health || exit 1
            
            echo "[OK] Déploiement réussi !"

      - name: Notify Slack on success
        if: success()
        uses: slackapi/slack-github-action@v1.26.0
        with:
          payload: |
            {
              "text": "[OK] UrbanPulse déployé avec succès en production !",
              "blocks": [{
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "[RAPIDE] *Déploiement réussi*\nCommit: `${{ github.sha }}`\nBranch: `${{ github.ref_name }}`"
                }
              }]
            }
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

      - name: Notify on failure
        if: failure()
        uses: slackapi/slack-github-action@v1.26.0
        with:
          payload: |
            {"text": "[X] Échec du déploiement UrbanPulse ! Vérifiez les logs GitHub Actions."}
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
```

---

## 19.3 Secrets GitHub Actions

Dans votre repo GitHub -> Settings -> Secrets and variables -> Actions :

```
PROD_SERVER_HOST     = 185.xxx.xxx.xxx (IP de votre serveur)
PROD_SERVER_USER     = urbanpulse
PROD_SSH_KEY         = (contenu de votre clé SSH privée)
SLACK_WEBHOOK_URL    = https://hooks.slack.com/...
SENTRY_DSN           = https://key@sentry.io/xxx
```

---

## [EDIT] Exercice 19.1 — Pipeline CI/CD

> **Objectif** : Automatiser les tests et le déploiement.

1. Créez `.github/workflows/ci-cd.yml` avec les jobs lint et test.
2. Poussez sur GitHub et observez l'exécution dans l'onglet "Actions".
3. Introduisez une erreur volontaire -> le pipeline échoue ?
4. Corrigez et pushez -> le pipeline repasse au vert ?
5. **Bonus** : Ajoutez le job `build` qui construit l'image Docker.

---

<a name="chapitre-20"></a>
# [GUIDE] Chapitre 20 — Performance

## 20.1 Optimisation des requêtes SQLAlchemy

Le principal ennemi des performances est le **problème N+1** — déclencher N requêtes SQL supplémentaires pour N objets.

```python
# app/routes/incidents.py — Requêtes optimisées

from sqlalchemy.orm import joinedload, subqueryload, contains_eager
from sqlalchemy import func

@incidents_bp.route("/")
def list():
    """Liste optimisée — 1 seule requête SQL."""
    
    # [X] MAUVAIS : N+1 queries
    incidents = Incident.query.all()
    for inc in incidents:
        print(inc.author.username)    # 1 requête par incident !
    
    # [OK] BON : 1 seule requête avec JOIN
    incidents = (
        Incident.query
        .options(joinedload(Incident.author))           # JOIN vers users
        .options(subqueryload(Incident.comments)        # Subquery pour les commentaires
                 .joinedload(Comment.author))           # JOIN commentaire -> auteur
        .order_by(Incident.vote_count.desc())
        .paginate(page=page, per_page=10)
    )
    
    return render_template("incidents/list.html", incidents=incidents.items)


def get_incidents_with_stats():
    """Requête avancée avec agrégation."""
    return (
        db.session.query(
            Incident,
            func.count(Comment.id).label("comment_count"),
            func.count(Vote.user_id).label("vote_count_real")
        )
        .outerjoin(Comment, Comment.incident_id == Incident.id)
        .outerjoin(Vote, Vote.incident_id == Incident.id)
        .group_by(Incident.id)
        .order_by(func.count(Vote.user_id).desc())
        .all()
    )


def get_top_categories():
    """Statistiques groupées par catégorie."""
    return (
        db.session.query(
            Incident.category,
            func.count(Incident.id).label("total"),
            func.sum(Incident.vote_count).label("total_votes")
        )
        .filter(Incident.status != "rejected")
        .group_by(Incident.category)
        .order_by(func.count(Incident.id).desc())
        .all()
    )
```

---

## 20.2 Caching avec Flask-Caching et Redis

```bash
pip install flask-caching
```

```python
# app/__init__.py

from flask_caching import Cache

cache = Cache()

def create_app(config_name="development"):
    app = Flask(__name__)
    
    app.config["CACHE_TYPE"] = "RedisCache"
    app.config["CACHE_REDIS_URL"] = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
    app.config["CACHE_DEFAULT_TIMEOUT"] = 300   # 5 minutes par défaut
    
    cache.init_app(app)
    return app
```

```python
# app/routes/incidents.py — Routes avec cache

@incidents_bp.route("/")
@cache.cached(timeout=60, query_string=True)  # Cache 60s, inclut les query params
def list():
    """Liste cachée — la même URL retourne les données en cache pendant 60s."""
    # Cette fonction n'est appelée qu'une fois toutes les 60 secondes
    # pour la même combinaison de paramètres
    incidents = Incident.query.options(joinedload(Incident.author)).all()
    return render_template("incidents/list.html", incidents=incidents)


@incidents_bp.route("/<int:incident_id>")
@cache.cached(timeout=120, key_prefix="incident_%s")
def detail(incident_id):
    incident = Incident.query.get_or_404(incident_id)
    return render_template("incidents/detail.html", incident=incident)


# Invalider le cache quand les données changent
@incidents_bp.route("/create", methods=["POST"])
@login_required
def create():
    # ... créer l'incident ...
    
    # Invalider le cache de la liste
    cache.delete("incidents_list")
    cache.delete_many("incident_*")  # Tous les caches d'incidents
    
    return redirect(url_for("incidents.detail", incident_id=incident.id))


# Cache au niveau fonction (pas de route)
@cache.memoize(timeout=300)
def get_city_stats(city_name: str) -> dict:
    """Stats pour une ville — recalculées toutes les 5 minutes."""
    return {
        "open_incidents": Incident.query.filter_by(status="open").count(),
        "resolved_this_month": Incident.query.filter(
            Incident.status == "resolved",
            Incident.resolved_at >= datetime.utcnow().replace(day=1)
        ).count(),
        "top_category": db.session.query(
            Incident.category, func.count(Incident.id)
        ).group_by(Incident.category).order_by(
            func.count(Incident.id).desc()
        ).first()
    }


# Invalider un cache mémoïsé
@cache.memoize()
def expensive_function(arg):
    pass

cache.delete_memoized(expensive_function, "specific_arg")
```

---

## 20.3 Caching HTTP avec ETag et Last-Modified

```python
from flask import make_response, request
import hashlib
from datetime import datetime

@incidents_bp.route("/<int:incident_id>")
def detail(incident_id):
    incident = Incident.query.get_or_404(incident_id)
    
    # Générer un ETag basé sur le contenu
    content = f"{incident.id}:{incident.updated_at}:{incident.vote_count}"
    etag = hashlib.md5(content.encode()).hexdigest()
    
    # Vérifier si le client a déjà la version à jour
    if request.headers.get("If-None-Match") == etag:
        return "", 304   # Not Modified — le navigateur utilise son cache
    
    response = make_response(
        render_template("incidents/detail.html", incident=incident)
    )
    response.set_etag(etag)
    response.last_modified = incident.updated_at
    response.cache_control.max_age = 60       # Cache navigateur 60s
    response.cache_control.public = True
    
    return response
```

---

## 20.4 Pagination efficace avec curseur

Pour les grandes tables, la pagination par OFFSET devient lente. La pagination par curseur est bien plus rapide.

```python
# Pagination classique (lente sur grands volumes)
# SELECT * FROM incidents ORDER BY id LIMIT 10 OFFSET 10000
# -> PostgreSQL scan 10010 lignes pour en retourner 10

# Pagination par curseur (rapide)
# SELECT * FROM incidents WHERE id > 10000 ORDER BY id LIMIT 10
# -> Index lookup direct

@incidents_bp.route("/api/incidents/cursor")
def list_cursor():
    """
    Pagination par curseur — efficace pour les grands volumes.
    Le client passe le dernier ID reçu comme cursor.
    """
    cursor = request.args.get("cursor", type=int)
    limit = min(request.args.get("limit", 10, type=int), 50)
    
    query = Incident.query.order_by(Incident.id.asc())
    
    if cursor:
        query = query.filter(Incident.id > cursor)
    
    incidents = query.limit(limit + 1).all()   # +1 pour savoir s'il y a une page suivante
    
    has_next = len(incidents) > limit
    items = incidents[:limit]
    
    return jsonify({
        "incidents": [i.to_dict() for i in items],
        "next_cursor": items[-1].id if has_next else None,
        "has_next": has_next
    })
```

---

## 20.5 Profiling et mesure des performances

```python
# app/__init__.py — Profiler en développement

def create_app(config_name="development"):
    app = Flask(__name__)
    
    if config_name == "development" and os.environ.get("PROFILE"):
        from werkzeug.middleware.profiler import ProfilerMiddleware
        app.wsgi_app = ProfilerMiddleware(
            app.wsgi_app,
            restrictions=[30],     # Top 30 fonctions les plus lentes
            profile_dir="./profiles"
        )
    
    return app
```

```bash
# Lancer avec profiling activé
PROFILE=1 flask run

# Analyser les résultats
python -m pstats profiles/xxx.prof
```

---

## [EDIT] Exercice 20.1 — Audit de performance

> **Objectif** : Identifier et corriger les goulets d'étranglement.

1. Activez `SQLALCHEMY_ECHO = True` et chargez la liste des incidents.
2. Comptez les requêtes SQL générées.
3. Ajoutez `joinedload` et `subqueryload` pour réduire ce nombre.
4. Installez Flask-Caching et cachez la liste des incidents pendant 60s.
5. Mesurez le temps de réponse avant et après avec `time curl http://localhost:5000/incidents`.
6. **Bonus** : Utilisez `flask-debugtoolbar` pour visualiser les requêtes SQL dans le navigateur.

---

<a name="chapitre-21"></a>
# [GUIDE] Chapitre 21 — Observabilité

## 21.1 Métriques avec Prometheus

```bash
pip install prometheus-flask-exporter
```

```python
# app/__init__.py

from prometheus_flask_exporter import PrometheusMetrics

metrics = PrometheusMetrics.for_app_factory()

def create_app(config_name="development"):
    app = Flask(__name__)
    
    # Expose /metrics pour que Prometheus puisse scraper
    metrics.init_app(app)
    
    # Métriques métier personnalisées
    metrics.info("urbanpulse_info", "UrbanPulse application info", version="1.0.0")
    
    return app
```

```python
# Métriques personnalisées dans les routes

from prometheus_client import Counter, Histogram, Gauge

# Compteurs
incidents_created = Counter(
    "urbanpulse_incidents_created_total",
    "Nombre total d'incidents créés",
    ["category"]   # Label pour distinguer par catégorie
)

votes_cast = Counter(
    "urbanpulse_votes_total",
    "Nombre total de votes"
)

# Histogramme (distribution des valeurs)
incident_resolution_time = Histogram(
    "urbanpulse_incident_resolution_seconds",
    "Temps de résolution des incidents en secondes",
    buckets=[3600, 86400, 604800, 2592000]  # 1h, 1j, 1sem, 1mois
)

# Jauge (valeur actuelle)
open_incidents_gauge = Gauge(
    "urbanpulse_open_incidents",
    "Nombre d'incidents actuellement ouverts"
)


# Dans les routes :
@incidents_bp.route("/create", methods=["POST"])
@login_required
def create():
    if form.validate_on_submit():
        incident = Incident(...)
        db.session.add(incident)
        db.session.commit()
        
        # Incrémenter le compteur avec le label catégorie
        incidents_created.labels(category=incident.category).inc()
        
        # Mettre à jour la jauge
        open_incidents_gauge.set(
            Incident.query.filter_by(status="open").count()
        )
```

---

## 21.2 Stack de monitoring : Prometheus + Grafana

```yaml
# docker-compose.monitoring.yml — Stack de monitoring

version: "3.9"

services:
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'
    ports:
      - "9090:9090"

  grafana:
    image: grafana/grafana:latest
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin123
      GF_INSTALL_PLUGINS: grafana-piechart-panel
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/grafana/dashboards:/etc/grafana/provisioning/dashboards
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

volumes:
  prometheus_data:
  grafana_data:
```

```yaml
# monitoring/prometheus.yml

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "urbanpulse"
    static_configs:
      - targets: ["web:8000"]   # L'endpoint /metrics de Flask
    metrics_path: "/metrics"
```

---

## 21.3 Alerting

```yaml
# monitoring/alerts.yml — Règles d'alerte Prometheus

groups:
  - name: urbanpulse
    rules:
      - alert: HighErrorRate
        expr: rate(flask_http_request_total{status=~"5.."}[5m]) > 0.1
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Taux d'erreur élevé"
          description: "Plus de 10% de requêtes en erreur 5xx depuis 2 minutes"

      - alert: SlowResponses
        expr: histogram_quantile(0.95, rate(flask_http_request_duration_seconds_bucket[5m])) > 2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Réponses lentes"
          description: "95e percentile des temps de réponse > 2 secondes"

      - alert: TooManyOpenIncidents
        expr: urbanpulse_open_incidents > 500
        labels:
          severity: info
        annotations:
          summary: "Beaucoup d'incidents ouverts"
```

---

## [EDIT] Exercice 21.1 — Monitoring UrbanPulse

> **Objectif** : Visualiser les métriques de l'application.

1. Installez `prometheus-flask-exporter` et visitez `/metrics`.
2. Ajoutez des compteurs pour la création d'incidents et les votes.
3. Lancez Prometheus avec `docker-compose -f docker-compose.monitoring.yml up -d`.
4. Configurez Prometheus pour scraper votre app.
5. Ouvrez Grafana sur `http://localhost:3000` et créez un dashboard avec :
   - Requêtes par seconde
   - Temps de réponse moyen
   - Taux d'erreur

---

<a name="chapitre-22"></a>
# [GUIDE] Chapitre 22 — Architecture et Modularité

## 22.1 Service Layer — Séparer la logique métier

Le **service layer** isole la logique métier des routes Flask. Cela rend le code testable, réutilisable, et maintenable.

```
Routes (HTTP)    ->  Services (logique métier)  ->  Repositories (DB)
app/routes/         app/services/                 app/models/
```

```python
# app/services/incident_service.py

from datetime import datetime
from typing import Optional, List, Dict, Any
from app import db
from app.models.incident import Incident
from app.models.vote import Vote
from app.models.user import User
from app.exceptions import ConflictError, PermissionError, NotFoundError
from app.tasks.email_tasks import send_incident_notification, send_status_update


class IncidentService:
    """
    Service gérant toute la logique métier des incidents.
    
    Avantages :
    - Logique isolée des routes -> testable indépendamment
    - Réutilisable par l'API web ET l'API REST
    - Transactions DB gérées en un seul endroit
    """

    @staticmethod
    def create(
        title: str,
        description: str,
        author_id: int,
        category: str = "autre",
        address: Optional[str] = None,
        latitude: Optional[float] = None,
        longitude: Optional[float] = None,
        photo_url: Optional[str] = None
    ) -> Incident:
        """
        Crée un incident avec toute la logique associée :
        - Validation métier
        - Création en base
        - Envoi de notification
        """
        # Validation métier (au-delà de la validation de formulaire)
        author = User.query.get(author_id)
        if not author:
            raise NotFoundError("Utilisateur introuvable.")

        if not author.is_active:
            raise PermissionError("Votre compte est désactivé.")

        # Vérifier les doublons récents (même auteur, même titre dans les 5 dernières minutes)
        recent_duplicate = Incident.query.filter(
            Incident.author_id == author_id,
            Incident.title == title,
            Incident.created_at >= datetime.utcnow() - timedelta(minutes=5)
        ).first()

        if recent_duplicate:
            raise ConflictError(
                "Un signalement identique a déjà été créé récemment.",
                payload={"existing_id": recent_duplicate.id}
            )

        incident = Incident(
            title=title,
            description=description,
            category=category,
            address=address,
            latitude=latitude,
            longitude=longitude,
            photo_url=photo_url,
            author_id=author_id
        )

        db.session.add(incident)
        db.session.commit()

        # Tâche asynchrone — n'attend pas la réponse
        send_incident_notification.delay(incident.id, author.email)

        return incident

    @staticmethod
    def vote(incident_id: int, user_id: int) -> Dict[str, Any]:
        """Enregistre un vote avec vérification des doublons."""
        incident = Incident.query.get(incident_id)
        if not incident:
            raise NotFoundError(f"Incident #{incident_id} introuvable.")

        if incident.status == "rejected":
            raise PermissionError("Impossible de voter pour un incident rejeté.")

        existing = Vote.query.filter_by(
            user_id=user_id,
            incident_id=incident_id
        ).first()

        if existing:
            raise ConflictError("Vous avez déjà voté pour cet incident.")

        vote = Vote(user_id=user_id, incident_id=incident_id)
        incident.vote_count += 1
        db.session.add(vote)
        db.session.commit()

        return {
            "vote_count": incident.vote_count,
            "message": "Vote enregistré avec succès."
        }

    @staticmethod
    def change_status(
        incident_id: int,
        new_status: str,
        changed_by_id: int,
        comment: Optional[str] = None
    ) -> Incident:
        """Change le statut d'un incident avec vérification des permissions."""
        incident = Incident.query.get(incident_id)
        if not incident:
            raise NotFoundError(f"Incident #{incident_id} introuvable.")

        moderator = User.query.get(changed_by_id)
        if not moderator or not moderator.has_role("admin", "moderator"):
            raise PermissionError("Seuls les modérateurs peuvent changer le statut.")

        if new_status not in Incident.VALID_STATUSES:
            raise ValueError(f"Statut invalide : {new_status}")

        old_status = incident.status
        incident.status = new_status

        if new_status == "resolved":
            incident.resolved_at = datetime.utcnow()

        # Ajouter un commentaire officiel si fourni
        if comment:
            from app.models.comment import Comment
            official_comment = Comment(
                content=comment,
                author_id=changed_by_id,
                incident_id=incident_id,
                is_official=True
            )
            db.session.add(official_comment)

        db.session.commit()

        # Notifier l'auteur du changement de statut
        send_status_update.delay(
            incident_id=incident.id,
            old_status=old_status,
            new_status=new_status,
            author_email=incident.author.email
        )

        return incident

    @staticmethod
    def get_statistics() -> Dict[str, Any]:
        """Calcule les statistiques globales d'UrbanPulse."""
        from sqlalchemy import func

        stats = db.session.query(
            func.count(Incident.id).label("total"),
            func.sum(
                db.case((Incident.status == "open", 1), else_=0)
            ).label("open"),
            func.sum(
                db.case((Incident.status == "resolved", 1), else_=0)
            ).label("resolved"),
            func.avg(Incident.vote_count).label("avg_votes")
        ).first()

        return {
            "total_incidents": stats.total or 0,
            "open_incidents": stats.open or 0,
            "resolved_incidents": stats.resolved or 0,
            "avg_votes_per_incident": round(float(stats.avg_votes or 0), 1),
            "resolution_rate": (
                round(stats.resolved / stats.total * 100, 1)
                if stats.total else 0
            )
        }
```

---

## 22.2 Configuration modulaire (Environnements)

```python
# app/config.py — Configuration complète et modulaire

import os
from datetime import timedelta

class Config:
    """Configuration de base — héritée par tous les environnements."""

    # ==================== Sécurité ====================
    SECRET_KEY = os.environ.get("SECRET_KEY") or "dev-key-change-in-production"
    WTF_CSRF_ENABLED = True

    # ==================== Base de données ====================
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_ENGINE_OPTIONS = {
        "pool_size": 10,           # Taille du pool de connexions
        "pool_timeout": 30,        # Timeout pour obtenir une connexion
        "pool_recycle": 1800,      # Recycler les connexions après 30min
        "pool_pre_ping": True,     # Vérifier que la connexion est vivante
        "max_overflow": 20         # Connexions supplémentaires si le pool est plein
    }

    # ==================== JWT ====================
    JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY") or "jwt-dev-key"
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)

    # ==================== Celery ====================
    CELERY_BROKER_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
    CELERY_RESULT_BACKEND = os.environ.get("REDIS_URL", "redis://localhost:6379/0")

    # ==================== Cache ====================
    CACHE_TYPE = "SimpleCache"    # Remplacé par Redis en production
    CACHE_DEFAULT_TIMEOUT = 300

    # ==================== Email ====================
    MAIL_SERVER = os.environ.get("MAIL_SERVER", "localhost")
    MAIL_PORT = int(os.environ.get("MAIL_PORT", 587))
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.environ.get("MAIL_USERNAME")
    MAIL_PASSWORD = os.environ.get("MAIL_PASSWORD")
    MAIL_DEFAULT_SENDER = os.environ.get("MAIL_DEFAULT_SENDER", "noreply@urbanpulse.com")

    # ==================== Uploads ====================
    MAX_CONTENT_LENGTH = 10 * 1024 * 1024   # 10 Mo max
    UPLOAD_FOLDER = os.path.join(os.path.dirname(__file__), "static", "uploads")
    ALLOWED_EXTENSIONS = {"jpg", "jpeg", "png", "webp"}

    # ==================== Pagination ====================
    INCIDENTS_PER_PAGE = 10
    MAX_PER_PAGE = 50

    @staticmethod
    def init_app(app):
        """Hook d'initialisation pour chaque environnement."""
        os.makedirs(app.config["UPLOAD_FOLDER"], exist_ok=True)


class DevelopmentConfig(Config):
    """Développement local."""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL") or \
        "sqlite:///urbanpulse_dev.db"
    SQLALCHEMY_ECHO = True          # Affiche les requêtes SQL
    WTF_CSRF_ENABLED = False        # Désactivé pour faciliter les tests manuels

    @staticmethod
    def init_app(app):
        Config.init_app(app)
        app.logger.setLevel("DEBUG")


class TestingConfig(Config):
    """Tests automatisés."""
    TESTING = True
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
    WTF_CSRF_ENABLED = False
    CELERY_TASK_ALWAYS_EAGER = True     # Tâches Celery exécutées synchronement
    CELERY_TASK_EAGER_PROPAGATES = True
    CACHE_TYPE = "SimpleCache"
    MAIL_SUPPRESS_SEND = True           # Pas d'emails réels pendant les tests


class ProductionConfig(Config):
    """Production."""
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL")
    CACHE_TYPE = "RedisCache"
    CACHE_REDIS_URL = os.environ.get("REDIS_URL")

    @staticmethod
    def init_app(app):
        Config.init_app(app)

        # Vérifier que les variables critiques sont définies
        required_vars = [
            "SECRET_KEY", "DATABASE_URL", "JWT_SECRET_KEY", "REDIS_URL"
        ]
        missing = [var for var in required_vars if not os.environ.get(var)]
        if missing:
            raise RuntimeError(
                f"Variables d'environnement manquantes en production : {missing}"
            )

        # Logging vers syslog en production
        import logging
        from logging.handlers import SysLogHandler
        syslog = SysLogHandler()
        syslog.setLevel(logging.WARNING)
        app.logger.addHandler(syslog)


# Registre de configurations
config = {
    "development": DevelopmentConfig,
    "testing": TestingConfig,
    "production": ProductionConfig,
    "default": DevelopmentConfig
}
```

---

## 22.3 Architecture hexagonale simplifiée

Pour les applications complexes, l'architecture hexagonale (ports & adapters) sépare clairement le domaine métier des détails d'implémentation.

```
app/
├── domain/              <- Logique métier pure (pas de Flask, pas de SQLAlchemy)
│   ├── entities/
│   │   ├── incident.py  <- Classe Incident Python pure
│   │   └── user.py
│   └── services/
│       └── incident_service.py  <- Règles métier
│
├── infrastructure/      <- Détails techniques (DB, email, etc.)
│   ├── database/
│   │   ├── models.py    <- Modèles SQLAlchemy
│   │   └── repositories.py
│   └── email/
│       └── email_service.py
│
├── interfaces/          <- Points d'entrée (web, API, CLI)
│   ├── web/
│   │   └── routes/
│   └── api/
│       └── endpoints/
│
└── app.py               <- Assembly / Composition root
```

---

## 22.4 Scalabilité

### Scalabilité horizontale

```yaml
# docker-compose.scale.yml — Plusieurs instances de l'app

services:
  web:
    image: urbanpulse:latest
    deploy:
      replicas: 3          # 3 instances de l'application
    # Pas de ports exposés directement — passent par le load balancer

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx/load-balancer.conf:/etc/nginx/conf.d/default.conf
```

```nginx
# nginx/load-balancer.conf — Load balancing

upstream urbanpulse_backend {
    least_conn;              # Algorithme : envoyer vers le serveur le moins chargé
    server web_1:8000;
    server web_2:8000;
    server web_3:8000;

    keepalive 32;            # Connexions persistantes vers les backends
}

server {
    listen 80;

    location / {
        proxy_pass http://urbanpulse_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}
```

### Connection Pooling

```python
# Pour PostgreSQL avec pgBouncer (connection pooler)
# pgBouncer maintient un pool de connexions PostgreSQL
# et multiplex les connexions de l'application

# app/config.py
class ProductionConfig(Config):
    # Connexion via pgBouncer (port 6432 au lieu de 5432)
    SQLALCHEMY_DATABASE_URI = "postgresql://user:pass@pgbouncer:6432/urbanpulse"

    # En mode transaction pooling avec pgBouncer, désactiver les prepared statements
    SQLALCHEMY_ENGINE_OPTIONS = {
        "pool_size": 5,       # Réduire : pgBouncer gère le pooling
        "max_overflow": 10,
        "connect_args": {
            "options": "-c default_transaction_isolation=read committed"
        }
    }
```

---

## [EDIT] Exercice 22.1 — Service Layer

> **Objectif** : Refactoriser les routes pour utiliser le service layer.

1. Créez `app/services/incident_service.py` avec les méthodes `create`, `vote`, `change_status`.
2. Refactorisez la route `incidents.create` pour utiliser `IncidentService.create()`.
3. Refactorisez la route `incidents.vote` pour utiliser `IncidentService.vote()`.
4. Testez les services **indépendamment des routes** dans `tests/unit/test_incident_service.py`.
5. Vérifiez que les erreurs personnalisées sont bien propagées jusqu'aux routes.

## [EDIT] Exercice 22.2 — Configuration multi-environnements

> **Objectif** : Gérer proprement les configurations.

1. Créez `app/config.py` avec `DevelopmentConfig`, `TestingConfig`, `ProductionConfig`.
2. Mettez à jour `create_app()` pour sélectionner la config selon l'environnement.
3. Ajoutez la validation des variables d'environnement obligatoires en production.
4. Créez un fichier `.env.example` (sans valeurs sensibles) à committer dans Git.
5. **Test** : Lancez avec `FLASK_ENV=production` sans DATABASE_URL -> l'erreur est-elle claire ?

---

## [TROPHEE] Récapitulatif Module 4

Vous maîtrisez maintenant :

[OK] Gunicorn avec configuration production (workers, timeouts, logging)
[OK] Nginx comme reverse proxy avec SSL, compression, et fichiers statiques
[OK] Systemd pour gérer l'application comme un service
[OK] Docker multi-stage pour des images légères et sécurisées
[OK] Docker Compose avec tous les services (DB, Redis, Celery, Nginx)
[OK] GitHub Actions pour le CI/CD complet (lint -> test -> build -> deploy)
[OK] Optimisation SQLAlchemy (eager loading, agrégations)
[OK] Caching avec Redis (routes, fonctions, HTTP ETag)
[OK] Pagination par curseur pour les grands volumes
[OK] Prometheus + Grafana pour le monitoring
[OK] Service Layer pour la séparation des responsabilités
[OK] Configuration modulaire pour les environnements

### [PACKAGE] État final du projet UrbanPulse

```
urbanpulse/
├── .github/
│   └── workflows/
│       └── ci-cd.yml              [OK] CI/CD complet
├── app/
│   ├── __init__.py                [OK] Factory avec toutes les extensions
│   ├── config.py                  [OK] 3 environnements + validation
│   ├── models/                    [OK] User, Incident, Comment, Vote
│   ├── routes/                    [OK] Blueprints modulaires
│   ├── api/                       [OK] REST API avec Swagger
│   ├── services/
│   │   ├── incident_service.py    [OK] Logique métier
│   │   └── user_service.py        [OK]
│   ├── schemas/                   [OK] Marshmallow
│   ├── forms/                     [OK] WTForms
│   ├── tasks/                     [OK] Celery
│   ├── templates/                 [OK] Jinja2 avec macros
│   └── static/                    [OK]
├── monitoring/
│   ├── prometheus.yml             [OK]
│   └── alerts.yml                 [OK]
├── nginx/
│   └── nginx.conf                 [OK]
├── tests/                         [OK] 80%+ couverture
├── Dockerfile                     [OK] Multi-stage
├── docker-compose.yml             [OK] Tous services
├── docker-compose.monitoring.yml  [OK] Prometheus + Grafana
├── gunicorn.conf.py               [OK] Config production
├── .github/workflows/ci-cd.yml   [OK]
├── .env.example                   [OK]
└── requirements.txt               [OK]

**-> Module 5 : Projets pratiques complets**
```

# [PYTHON] Formation Flask — Module 5
## Projets Pratiques, Expertise & Aller Plus Loin
### Parties IX & X — Chapitres 23 à 30

---

> [OBJECTIF] **La dernière ligne droite**
> Vous avez maintenant toutes les compétences. Ce module vous fait construire
> 4 mini-projets complémentaires à UrbanPulse, puis vous guide vers l'expertise.

---

## [WORLD_MAP] Table des matières

- [Chapitre 23 — Projet 1 : Blog CRUD Complet](#chapitre-23)
- [Chapitre 24 — Projet 2 : API REST E-commerce](#chapitre-24)
- [Chapitre 25 — Projet 3 : Microservices avec Celery](#chapitre-25)
- [Chapitre 26 — Projet 4 : Fullstack React + Flask](#chapitre-26)
- [Chapitre 27 — Extensions Essentielles](#chapitre-27)
- [Chapitre 28 — Scalabilité Avancée](#chapitre-28)
- [Chapitre 29 — Documentation et OpenAPI](#chapitre-29)
- [Chapitre 30 — Aller Plus Loin](#chapitre-30)

---

<a name="chapitre-23"></a>
# [GUIDE] Chapitre 23 — Projet 1 : Blog CRUD Complet

## 23.1 Présentation du projet

Un blog complet avec gestion des articles, tags, commentaires et authentification. Ce projet consolide tout ce que vous avez appris dans un contexte différent d'UrbanPulse.

**Fonctionnalités :**
- Articles avec titre, contenu Markdown, image de couverture, statut (brouillon/publié)
- Tags et catégories
- Commentaires imbriqués
- Auteurs avec profils
- Recherche full-text
- Fil RSS
- Sitemap XML

---

## 23.2 Modèles du blog

```python
# blog/models/post.py

from datetime import datetime
from slugify import slugify       # pip install python-slugify
from app import db

# Table de liaison Post <-> Tag (Many-to-Many)
post_tags = db.Table(
    "post_tags",
    db.Column("post_id", db.Integer, db.ForeignKey("posts.id"), primary_key=True),
    db.Column("tag_id", db.Integer, db.ForeignKey("tags.id"), primary_key=True)
)

class Post(db.Model):
    __tablename__ = "posts"

    id = db.Column(db.Integer, primary_key=True)

    # Slug auto-généré à partir du titre (URL-friendly)
    # Ex: "Mon Article Sympa" -> "mon-article-sympa"
    slug = db.Column(db.String(250), unique=True, nullable=False, index=True)

    title = db.Column(db.String(300), nullable=False)
    summary = db.Column(db.String(500))        # Résumé court pour les listes
    content = db.Column(db.Text, nullable=False)  # Contenu Markdown
    cover_image = db.Column(db.String(500))

    # Statut de publication
    STATUS_DRAFT = "draft"
    STATUS_PUBLISHED = "published"
    STATUS_ARCHIVED = "archived"
    status = db.Column(db.String(20), default=STATUS_DRAFT, nullable=False)

    # Compteurs
    view_count = db.Column(db.Integer, default=0)
    comment_count = db.Column(db.Integer, default=0)  # Dénormalisé pour performance

    # SEO
    meta_description = db.Column(db.String(160))
    meta_keywords = db.Column(db.String(200))

    # Timestamps
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    published_at = db.Column(db.DateTime)   # Quand l'article a été publié

    # Relations
    author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    category_id = db.Column(db.Integer, db.ForeignKey("categories.id"))

    tags = db.relationship("Tag", secondary=post_tags, backref="posts", lazy="subquery")
    comments = db.relationship(
        "PostComment", backref="post", lazy="dynamic",
        cascade="all, delete-orphan"
    )

    def generate_slug(self):
        """Génère un slug unique à partir du titre."""
        base_slug = slugify(self.title)
        slug = base_slug
        counter = 1

        # Garantir l'unicité du slug
        while Post.query.filter_by(slug=slug).filter(Post.id != self.id).first():
            slug = f"{base_slug}-{counter}"
            counter += 1

        self.slug = slug

    def publish(self):
        """Publie l'article."""
        self.status = self.STATUS_PUBLISHED
        self.published_at = datetime.utcnow()

    @property
    def reading_time(self) -> int:
        """Temps de lecture estimé en minutes (250 mots/minute)."""
        word_count = len(self.content.split())
        return max(1, round(word_count / 250))

    def increment_views(self):
        """Incrémente le compteur de vues de manière atomique."""
        Post.query.filter_by(id=self.id).update({
            "view_count": Post.view_count + 1
        })
        db.session.commit()

    def __repr__(self):
        return f"<Post '{self.slug}' [{self.status}]>"


class Tag(db.Model):
    __tablename__ = "tags"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False)
    slug = db.Column(db.String(60), unique=True, nullable=False)
    color = db.Column(db.String(7), default="#6c757d")   # Couleur hex du tag

    def __repr__(self):
        return f"<Tag #{self.name}>"


class Category(db.Model):
    __tablename__ = "categories"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)
    slug = db.Column(db.String(120), unique=True, nullable=False)
    description = db.Column(db.Text)
    posts = db.relationship("Post", backref="category", lazy="dynamic")
```

---

## 23.3 Routes du blog avec Markdown

```python
# blog/routes/posts.py

import bleach
import markdown
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort
from flask_login import login_required, current_user

posts_bp = Blueprint("posts", __name__, url_prefix="/blog")

# Balises HTML autorisées dans le contenu Markdown rendu
ALLOWED_TAGS = ["p", "h1", "h2", "h3", "h4", "ul", "ol", "li",
                "strong", "em", "code", "pre", "blockquote", "a",
                "img", "table", "thead", "tbody", "tr", "th", "td"]
ALLOWED_ATTRIBUTES = {"a": ["href", "title"], "img": ["src", "alt"]}


def render_markdown(content: str) -> str:
    """Convertit le Markdown en HTML sécurisé."""
    html = markdown.markdown(
        content,
        extensions=["fenced_code", "tables", "toc", "nl2br"]
    )
    # Nettoyage XSS : autoriser seulement les balises sûres
    return bleach.clean(html, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRIBUTES)


@posts_bp.route("/")
def index():
    """Liste paginée des articles publiés."""
    page = request.args.get("page", 1, type=int)
    tag_slug = request.args.get("tag")
    category_slug = request.args.get("category")

    query = Post.query.filter_by(status=Post.STATUS_PUBLISHED)\
                      .order_by(Post.published_at.desc())

    if tag_slug:
        tag = Tag.query.filter_by(slug=tag_slug).first_or_404()
        query = query.filter(Post.tags.contains(tag))

    if category_slug:
        category = Category.query.filter_by(slug=category_slug).first_or_404()
        query = query.filter_by(category_id=category.id)

    pagination = query.paginate(page=page, per_page=10)

    return render_template(
        "blog/index.html",
        posts=pagination.items,
        pagination=pagination
    )


@posts_bp.route("/<string:slug>")
def detail(slug):
    """Affiche un article complet."""
    post = Post.query.filter_by(
        slug=slug,
        status=Post.STATUS_PUBLISHED
    ).first_or_404()

    # Incrémenter les vues (en arrière-plan pour ne pas ralentir la page)
    from blog.tasks import increment_post_views
    increment_post_views.delay(post.id)

    # Rendu Markdown sécurisé
    rendered_content = render_markdown(post.content)

    # Articles liés (même catégorie, excluant l'article actuel)
    related_posts = []
    if post.category:
        related_posts = Post.query.filter(
            Post.category_id == post.category_id,
            Post.id != post.id,
            Post.status == Post.STATUS_PUBLISHED
        ).order_by(Post.published_at.desc()).limit(3).all()

    return render_template(
        "blog/detail.html",
        post=post,
        rendered_content=rendered_content,
        related_posts=related_posts
    )


@posts_bp.route("/new", methods=["GET", "POST"])
@login_required
def create():
    """Formulaire de création d'article."""
    form = PostForm()

    if form.validate_on_submit():
        post = Post(
            title=form.title.data,
            content=form.content.data,
            summary=form.summary.data,
            author_id=current_user.id
        )
        post.generate_slug()

        if form.publish_now.data:
            post.publish()

        # Gestion des tags (créés à la volée si inexistants)
        for tag_name in form.tags.data.split(","):
            tag_name = tag_name.strip().lower()
            if tag_name:
                tag = Tag.query.filter_by(name=tag_name).first()
                if not tag:
                    tag = Tag(name=tag_name, slug=slugify(tag_name))
                    db.session.add(tag)
                post.tags.append(tag)

        db.session.add(post)
        db.session.commit()

        flash("Article créé avec succès !", "success")
        return redirect(url_for("posts.detail", slug=post.slug))

    return render_template("blog/create.html", form=form)


@posts_bp.route("/feed.rss")
def rss_feed():
    """Fil RSS des derniers articles."""
    from flask import Response
    posts = Post.query.filter_by(status=Post.STATUS_PUBLISHED)\
                      .order_by(Post.published_at.desc())\
                      .limit(20).all()

    rss = render_template("blog/feed.xml", posts=posts)
    return Response(rss, mimetype="application/rss+xml")
```

---

## [EDIT] Exercice 23.1 — Blog complet

> **Objectif** : Construire le blog de A à Z.

1. Créez le projet `blog/` avec la structure Blueprint.
2. Implémentez les modèles `Post`, `Tag`, `Category`, `PostComment`.
3. Créez les routes : liste, détail, création, édition, suppression.
4. Ajoutez le rendu Markdown avec `python-markdown` et la sanitisation avec `bleach`.
5. Implémentez la génération automatique de slugs.
6. Créez le template de liste avec les tags et la pagination.
7. **Bonus** : Implémentez la génération du fil RSS.
8. **Bonus** : Ajoutez une barre de recherche full-text.

---

<a name="chapitre-24"></a>
# [GUIDE] Chapitre 24 — Projet 2 : API REST E-commerce

## 24.1 Présentation du projet

Une API REST complète pour un système de commandes simplifié. Ce projet met en pratique les concepts d'API avancés : transactions complexes, états de machine, webhooks.

**Fonctionnalités :**
- Catalogue produits avec variantes (taille, couleur)
- Panier d'achat (session ou DB)
- Commandes avec workflow d'état (pending -> paid -> shipped -> delivered)
- Calcul de stock en temps réel
- Historique des transactions

---

## 24.2 Modèles E-commerce

```python
# shop/models/product.py

class Product(db.Model):
    __tablename__ = "products"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text)
    price = db.Column(db.Numeric(10, 2), nullable=False)   # Précis pour les montants
    sku = db.Column(db.String(100), unique=True)            # Stock Keeping Unit
    stock = db.Column(db.Integer, default=0, nullable=False)
    is_active = db.Column(db.Boolean, default=True)

    variants = db.relationship("ProductVariant", backref="product", lazy="dynamic")
    category_id = db.Column(db.Integer, db.ForeignKey("product_categories.id"))

    def is_in_stock(self, quantity: int = 1) -> bool:
        return self.stock >= quantity

    def reserve_stock(self, quantity: int):
        """Décrémente le stock de manière atomique (évite les race conditions)."""
        result = Product.query.filter(
            Product.id == self.id,
            Product.stock >= quantity
        ).update({"stock": Product.stock - quantity})

        if result == 0:
            raise ValueError(f"Stock insuffisant pour le produit '{self.name}'")

        db.session.flush()


class Order(db.Model):
    __tablename__ = "orders"

    # États de la commande
    STATUS_PENDING = "pending"         # En attente de paiement
    STATUS_PAID = "paid"               # Payée
    STATUS_PROCESSING = "processing"   # En cours de préparation
    STATUS_SHIPPED = "shipped"         # Expédiée
    STATUS_DELIVERED = "delivered"     # Livrée
    STATUS_CANCELLED = "cancelled"     # Annulée
    STATUS_REFUNDED = "refunded"       # Remboursée

    VALID_TRANSITIONS = {
        STATUS_PENDING:    [STATUS_PAID, STATUS_CANCELLED],
        STATUS_PAID:       [STATUS_PROCESSING, STATUS_REFUNDED],
        STATUS_PROCESSING: [STATUS_SHIPPED, STATUS_CANCELLED],
        STATUS_SHIPPED:    [STATUS_DELIVERED],
        STATUS_DELIVERED:  [STATUS_REFUNDED],
        STATUS_CANCELLED:  [],
        STATUS_REFUNDED:   [],
    }

    id = db.Column(db.Integer, primary_key=True)
    order_number = db.Column(db.String(20), unique=True, nullable=False)
    status = db.Column(db.String(20), default=STATUS_PENDING, nullable=False)

    # Montants (stockés en centimes pour éviter les erreurs virgule flottante)
    subtotal_cents = db.Column(db.Integer, nullable=False)
    tax_cents = db.Column(db.Integer, default=0)
    shipping_cents = db.Column(db.Integer, default=0)
    total_cents = db.Column(db.Integer, nullable=False)

    customer_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    items = db.relationship("OrderItem", backref="order", lazy="subquery")

    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    paid_at = db.Column(db.DateTime)
    shipped_at = db.Column(db.DateTime)

    @property
    def total(self):
        """Retourne le total en euros."""
        return self.total_cents / 100

    def transition_to(self, new_status: str):
        """Machine d'état : valide et applique la transition."""
        allowed = self.VALID_TRANSITIONS.get(self.status, [])
        if new_status not in allowed:
            raise ValueError(
                f"Transition invalide : {self.status} -> {new_status}. "
                f"Transitions autorisées : {allowed}"
            )

        self.status = new_status

        if new_status == self.STATUS_PAID:
            self.paid_at = datetime.utcnow()
        elif new_status == self.STATUS_SHIPPED:
            self.shipped_at = datetime.utcnow()

    @staticmethod
    def generate_order_number() -> str:
        """Génère un numéro de commande unique."""
        import random
        import string
        prefix = datetime.utcnow().strftime("%Y%m")
        suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
        return f"UP-{prefix}-{suffix}"
```

---

## 24.3 Service de commande avec transaction

```python
# shop/services/order_service.py

from decimal import Decimal
from sqlalchemy.exc import IntegrityError

class OrderService:
    """Gestion des commandes avec transactions atomiques."""

    @staticmethod
    def create_from_cart(customer_id: int, cart_items: list) -> Order:
        """
        Crée une commande depuis un panier.
        Utilise une transaction pour garantir la cohérence :
        - Vérification du stock
        - Réservation du stock
        - Création de la commande
        Le tout ou rien.
        """
        if not cart_items:
            raise ValueError("Le panier est vide.")

        try:
            subtotal_cents = 0
            order_items = []

            for item in cart_items:
                product = Product.query.with_for_update().get(item["product_id"])
                # with_for_update() = verrou pessimiste -> empêche les achats simultanés du même stock

                if not product or not product.is_active:
                    raise ValueError(f"Produit introuvable : {item['product_id']}")

                quantity = item["quantity"]
                product.reserve_stock(quantity)   # Lève une exception si stock insuffisant

                item_total = int(product.price * 100) * quantity
                subtotal_cents += item_total

                order_items.append(OrderItem(
                    product_id=product.id,
                    product_name=product.name,   # Snapshot du nom (le produit peut changer)
                    quantity=quantity,
                    unit_price_cents=int(product.price * 100),
                    total_cents=item_total
                ))

            # Calcul des taxes (TVA 20%)
            tax_cents = int(subtotal_cents * 0.20)
            shipping_cents = 500 if subtotal_cents < 5000 else 0   # Gratuit > 50€
            total_cents = subtotal_cents + tax_cents + shipping_cents

            order = Order(
                order_number=Order.generate_order_number(),
                customer_id=customer_id,
                subtotal_cents=subtotal_cents,
                tax_cents=tax_cents,
                shipping_cents=shipping_cents,
                total_cents=total_cents
            )
            db.session.add(order)
            db.session.flush()   # Obtenir l'ID de la commande

            for item in order_items:
                item.order_id = order.id
                db.session.add(item)

            db.session.commit()

            # Notification asynchrone
            from shop.tasks import send_order_confirmation
            send_order_confirmation.delay(order.id)

            return order

        except Exception:
            db.session.rollback()   # Annuler TOUTES les modifications si quelque chose échoue
            raise
```

---

## [EDIT] Exercice 24.1 — API E-commerce

> **Objectif** : Construire l'API du système de commandes.

1. Créez les modèles `Product`, `Order`, `OrderItem`.
2. Implémentez `OrderService.create_from_cart()` avec la transaction atomique.
3. Créez les endpoints API :
   - `GET /api/products` -> Catalogue avec filtres
   - `POST /api/cart/add` -> Ajouter au panier (session Flask)
   - `GET /api/cart` -> Voir le panier
   - `POST /api/orders` -> Commander depuis le panier
   - `GET /api/orders/{id}` -> Statut de la commande
   - `PATCH /api/orders/{id}/status` -> Changer le statut (admin)
4. Testez le scénario complet : créer produit -> ajouter au panier -> commander -> changer statut.
5. **Bonus** : Testez la race condition en créant 2 commandes simultanées pour 1 seul stock.

---

<a name="chapitre-25"></a>
# [GUIDE] Chapitre 25 — Projet 3 : Microservices avec Celery

## 25.1 Architecture microservices

Ce projet transforme UrbanPulse en architecture microservices :

```
┌──────────────────────────────────────────────────────────┐
│                    API Gateway (Nginx)                     │
└──────────┬───────────────┬──────────────────┬────────────┘
           │               │                  │
           [BLACK_DOWN-POINTING_TRIANGLE]               [BLACK_DOWN-POINTING_TRIANGLE]                  [BLACK_DOWN-POINTING_TRIANGLE]
    ┌──────────────┐  ┌──────────┐  ┌─────────────────┐
    │  Incidents   │  │  Users   │  │  Notifications   │
    │  Service     │  │  Service │  │  Service         │
    │  :5001       │  │  :5002   │  │  :5003           │
    └──────┬───────┘  └────┬─────┘  └────────┬────────┘
           │               │                  │
           └───────────────┴──────────────────┘
                           │
                    ┌──────[BLACK_DOWN-POINTING_TRIANGLE]──────┐
                    │    Redis    │
                    │  (Message   │
                    │    Bus)     │
                    └─────────────┘
```

---

## 25.2 Communication entre microservices

```python
# shared/events.py — Events partagés entre services

from celery import Celery
from dataclasses import dataclass
from datetime import datetime
import json

# Instance Celery partagée (même broker Redis)
celery_app = Celery(
    "urbanpulse_events",
    broker="redis://redis:6379/0"
)

@dataclass
class IncidentCreatedEvent:
    """Publié quand un incident est créé."""
    incident_id: int
    title: str
    category: str
    author_id: int
    author_email: str
    created_at: str = datetime.utcnow().isoformat()

    def publish(self):
        """Publie l'événement sur le bus de messages."""
        celery_app.send_task(
            "notifications.handle_incident_created",
            kwargs={"event": self.__dict__},
            queue="notifications"
        )
        celery_app.send_task(
            "analytics.track_incident_created",
            kwargs={"event": self.__dict__},
            queue="analytics"
        )


@dataclass
class IncidentStatusChangedEvent:
    """Publié quand le statut d'un incident change."""
    incident_id: int
    old_status: str
    new_status: str
    author_id: int
    author_email: str
    changed_by_id: int
```

```python
# notifications_service/tasks.py — Service de notifications

from shared.events import celery_app

@celery_app.task(name="notifications.handle_incident_created", queue="notifications")
def handle_incident_created(event: dict):
    """
    Réagit à la création d'un incident.
    Envoie un email de confirmation.
    """
    print(f"[EMAIL] Envoi email pour incident #{event['incident_id']}")

    send_email(
        to=event["author_email"],
        subject=f"Signalement #{event['incident_id']} reçu",
        body=f"Votre signalement '{event['title']}' a été enregistré."
    )

    # Notifier les modérateurs de la catégorie
    notify_category_moderators(event["category"], event["incident_id"])


@celery_app.task(name="notifications.handle_status_change", queue="notifications")
def handle_status_change(event: dict):
    """Notifie l'auteur du changement de statut."""
    status_messages = {
        "in_progress": "Votre signalement est en cours de traitement.",
        "resolved": "[BRAVO] Votre signalement a été résolu !",
        "rejected": "Votre signalement a été rejeté."
    }

    message = status_messages.get(event["new_status"], "Statut mis à jour.")
    send_email(to=event["author_email"], subject="Mise à jour de votre signalement", body=message)
```

---

## [EDIT] Exercice 25.1 — Architecture événementielle

> **Objectif** : Découpler les services avec des événements Celery.

1. Créez 3 dossiers : `incidents_service/`, `notifications_service/`, `analytics_service/`.
2. Chacun a son propre `app.py` Flask et ses propres workers Celery.
3. Quand un incident est créé dans `incidents_service`, publiez un événement sur Redis.
4. `notifications_service` réagit en envoyant un email (log en dev).
5. `analytics_service` réagit en incrémentant un compteur.
6. Créez un `docker-compose.yml` qui lance les 3 services + Redis.
7. **Test** : Créez un incident -> vérifiez que les 2 autres services réagissent dans leurs logs.

---

<a name="chapitre-26"></a>
# [GUIDE] Chapitre 26 — Projet 4 : Fullstack React + Flask

## 26.1 Architecture fullstack

Flask devient une **API pure** (pas de templates) et React le frontend.

```
React (Vite)          Flask API
:3000          ->      :5000/api/v1/
               <-      JSON responses
```

---

## 26.2 Configuration CORS

```bash
pip install flask-cors
```

```python
# app/__init__.py

from flask_cors import CORS

def create_app(config_name="development"):
    app = Flask(__name__)

    # Autoriser les requêtes depuis le frontend React
    CORS(app, resources={
        r"/api/*": {
            "origins": [
                "http://localhost:3000",       # Dev React
                "https://urbanpulse.example.com"  # Production
            ],
            "methods": ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
            "allow_headers": ["Content-Type", "Authorization"],
            "supports_credentials": True       # Pour les cookies de session
        }
    })

    return app
```

---

## 26.3 API adaptée au frontend React

```python
# app/api/endpoints.py — API optimisée pour React

@ns.route("/incidents")
class IncidentList(Resource):

    def get(self):
        """API utilisée par le frontend React."""
        # Retourner tout le nécessaire en une seule requête
        incidents = Incident.query.options(
            joinedload(Incident.author),
            subqueryload(Incident.tags) if hasattr(Incident, "tags") else None
        ).filter_by(status="open").order_by(Incident.vote_count.desc()).limit(20).all()

        return {
            "incidents": [i.to_dict() for i in incidents],
            "categories": [
                {"value": "voirie", "label": "Voirie", "count": 42},
                {"value": "eclairage", "label": "Éclairage", "count": 18},
            ],
            "total_open": Incident.query.filter_by(status="open").count()
        }
```

---

## 26.4 Frontend React minimal

```jsx
// frontend/src/App.jsx — Application React qui consomme l'API Flask

import { useState, useEffect } from "react";

const API_BASE = "http://localhost:5000/api/v1";

// Hook personnalisé pour les appels API
function useAPI(endpoint) {
  const [data, setData] = useState(null);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState(null);

  useEffect(() => {
    const token = localStorage.getItem("access_token");
    
    fetch(`${API_BASE}${endpoint}`, {
      headers: {
        "Content-Type": "application/json",
        ...(token ? { Authorization: `Bearer ${token}` } : {})
      }
    })
      .then(res => {
        if (!res.ok) throw new Error(`HTTP ${res.status}`);
        return res.json();
      })
      .then(setData)
      .catch(setError)
      .finally(() => setLoading(false));
  }, [endpoint]);

  return { data, loading, error };
}

// Composant de connexion
function LoginForm({ onLogin }) {
  const [email, setEmail] = useState("");
  const [password, setPassword] = useState("");
  const [error, setError] = useState("");

  const handleSubmit = async (e) => {
    e.preventDefault();
    try {
      const res = await fetch(`${API_BASE}/auth/login`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ email, password })
      });

      if (!res.ok) {
        const data = await res.json();
        throw new Error(data.message || "Identifiants incorrects");
      }

      const data = await res.json();
      localStorage.setItem("access_token", data.access_token);
      onLogin(data.user);
    } catch (err) {
      setError(err.message);
    }
  };

  return (
    <form onSubmit={handleSubmit} className="login-form">
      <h2>Connexion</h2>
      {error && <div className="error">{error}</div>}
      <input
        type="email"
        placeholder="Email"
        value={email}
        onChange={e => setEmail(e.target.value)}
        required
      />
      <input
        type="password"
        placeholder="Mot de passe"
        value={password}
        onChange={e => setPassword(e.target.value)}
        required
      />
      <button type="submit">Se connecter</button>
    </form>
  );
}

// Composant liste des incidents
function IncidentList({ currentUser }) {
  const { data, loading, error } = useAPI("/incidents/");
  const [votedIds, setVotedIds] = useState(new Set());

  const handleVote = async (incidentId) => {
    const token = localStorage.getItem("access_token");
    if (!token) {
      alert("Connectez-vous pour voter !");
      return;
    }

    try {
      const res = await fetch(`${API_BASE}/incidents/${incidentId}/vote`, {
        method: "POST",
        headers: { Authorization: `Bearer ${token}` }
      });

      if (res.ok) {
        setVotedIds(prev => new Set([...prev, incidentId]));
      }
    } catch (err) {
      console.error("Erreur vote:", err);
    }
  };

  if (loading) return <div className="loading">Chargement...</div>;
  if (error) return <div className="error">Erreur : {error.message}</div>;
  if (!data) return null;

  return (
    <div className="incidents-container">
      <header>
        <h1>[CITYSCAPE] UrbanPulse</h1>
        <span className="badge">{data.total_open} incidents ouverts</span>
      </header>

      <div className="incidents-grid">
        {data.incidents.map(incident => (
          <article key={incident.id} className={`card card-${incident.status}`}>
            <div className="card-header">
              <span className="category">{incident.category}</span>
              <span className="status">{incident.status}</span>
            </div>

            <h3>{incident.title}</h3>
            <p>{incident.description.slice(0, 150)}...</p>

            <footer>
              <span>[UTILISATEUR] {incident.author?.username}</span>
              <button
                onClick={() => handleVote(incident.id)}
                disabled={votedIds.has(incident.id)}
                className={`vote-btn ${votedIds.has(incident.id) ? "voted" : ""}`}
              >
                [BLACK_UP-POINTING_TRIANGLE] {incident.vote_count}
              </button>
            </footer>
          </article>
        ))}
      </div>
    </div>
  );
}

// Application principale
export default function App() {
  const [user, setUser] = useState(null);

  return (
    <div className="app">
      {!user ? (
        <LoginForm onLogin={setUser} />
      ) : (
        <IncidentList currentUser={user} />
      )}
    </div>
  );
}
```

---

## [EDIT] Exercice 26.1 — Fullstack complet

> **Objectif** : Connecter React à Flask.

1. Configurez CORS dans Flask.
2. Créez une app React simple (Vite : `npm create vite@latest frontend -- --template react`).
3. Implémentez `useAPI` hook pour les appels à l'API Flask.
4. Créez le composant `LoginForm` qui obtient un JWT.
5. Créez le composant `IncidentList` qui charge et affiche les incidents.
6. Implémentez le vote depuis React.
7. **Bonus** : Ajoutez le formulaire de création d'incident depuis React.

---

<a name="chapitre-27"></a>
# [GUIDE] Chapitre 27 — Extensions Essentielles

## 27.1 Flask-Mail — Envoi d'emails

```bash
pip install flask-mail
```

```python
# app/__init__.py
from flask_mail import Mail

mail = Mail()

def create_app(config_name="development"):
    app = Flask(__name__)
    mail.init_app(app)
    return app
```

```python
# app/services/email_service.py

from flask_mail import Message
from flask import render_template
from app import mail

def send_welcome_email(user):
    """Envoie l'email de bienvenue à un nouvel inscrit."""
    msg = Message(
        subject="Bienvenue sur UrbanPulse !",
        recipients=[user.email],
        html=render_template("emails/welcome.html", user=user),
        body=render_template("emails/welcome.txt", user=user)   # Version texte
    )
    mail.send(msg)

def send_password_reset(user, reset_token):
    """Envoie le lien de réinitialisation du mot de passe."""
    reset_url = url_for("auth.reset_password", token=reset_token, _external=True)
    msg = Message(
        subject="Réinitialisation de votre mot de passe",
        recipients=[user.email],
        html=render_template("emails/reset_password.html", user=user, url=reset_url)
    )
    mail.send(msg)
```

---

## 27.2 Flask-Migrate vs migrations manuelles

| Situation | Recommandation |
|---|---|
| Projet simple, petite équipe | Flask-Migrate (automatique) |
| Projet complexe, migrations délicates | Migrations manuelles Alembic |
| Migration de données complexe | Script Python dédié |
| Production critique | Toujours tester sur staging d'abord |

---

## 27.3 Flask-Principal — Permissions granulaires

```bash
pip install flask-principal
```

```python
from flask_principal import Principal, Permission, RoleNeed

principal = Principal()

# Définition des permissions
admin_permission = Permission(RoleNeed("admin"))
moderator_permission = Permission(RoleNeed("moderator"), RoleNeed("admin"))

# Dans les routes
@incidents_bp.route("/<int:id>/delete", methods=["DELETE"])
@login_required
def delete(id):
    incident = Incident.query.get_or_404(id)
    
    # Vérification : auteur ou admin
    if incident.author_id != current_user.id:
        with admin_permission.require(http_exception=403):
            pass
    
    db.session.delete(incident)
    db.session.commit()
    return "", 204
```

---

## 27.4 Tableau de bord des extensions recommandées

| Extension | Usage | Priorité |
|---|---|---|
| Flask-SQLAlchemy | ORM | *** Indispensable |
| Flask-Migrate | Migrations | *** Indispensable |
| Flask-Login | Auth sessions | *** Indispensable |
| Flask-JWT-Extended | Auth API/JWT | *** Pour les APIs |
| Flask-WTF | Formulaires + CSRF | *** Indispensable |
| Flask-RESTX | API REST + Swagger | *** Pour les APIs |
| Flask-Limiter | Rate limiting | *** Sécurité |
| Flask-Mail | Emails | ** Très utile |
| Flask-Caching | Cache Redis | ** Performance |
| Flask-CORS | CORS pour SPAs | ** Si frontend séparé |
| Flask-Babel | i18n/l10n | * Si international |
| Flask-SocketIO | WebSockets | * Si temps réel |
| Flask-Admin | Panel admin auto | * Pour les MVPs |

---

<a name="chapitre-28"></a>
# [GUIDE] Chapitre 28 — Scalabilité Avancée

## 28.1 Read Replicas PostgreSQL

```python
# app/config.py — Réplication lecture/écriture

class ProductionConfig(Config):
    # Écriture sur le master
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_MASTER_URL")
    
    # Lecture sur les replicas (rotation)
    DATABASE_REPLICAS = [
        os.environ.get("DATABASE_REPLICA_1_URL"),
        os.environ.get("DATABASE_REPLICA_2_URL"),
    ]

# app/db_router.py — Routage intelligent lecture/écriture
import random

class DBRouter:
    """
    Route les requêtes READ vers les replicas,
    les requêtes WRITE vers le master.
    """
    
    @staticmethod
    def get_read_engine():
        """Retourne un engine de lecture aléatoire (round-robin)."""
        replicas = current_app.config.get("DATABASE_REPLICAS", [])
        if not replicas:
            return db.engine   # Fallback sur le master si pas de replica
        
        url = random.choice(replicas)
        return create_engine(url)
    
    @staticmethod
    def read_query(query):
        """Exécute une requête SELECT sur un replica."""
        with DBRouter.get_read_engine().connect() as conn:
            return conn.execute(query)
```

---

## 28.2 Celery : Queues prioritaires

```python
# Configuration des queues Celery

CELERY_TASK_ROUTES = {
    # Tâches critiques -> queue haute priorité
    "app.tasks.email_tasks.send_password_reset": {"queue": "high"},
    "app.tasks.email_tasks.send_incident_notification": {"queue": "default"},
    
    # Tâches non urgentes -> queue basse priorité
    "app.tasks.maintenance.cleanup_uploads": {"queue": "low"},
    "app.tasks.analytics.generate_report": {"queue": "low"},
}

# Lancer des workers spécialisés par queue
# Worker haute priorité (2 processus)
# celery -A app worker -Q high --concurrency=2

# Worker défaut (4 processus)
# celery -A app worker -Q default --concurrency=4

# Worker basse priorité (1 processus)
# celery -A app worker -Q low --concurrency=1
```

---

<a name="chapitre-29"></a>
# [GUIDE] Chapitre 29 — Documentation et OpenAPI

## 29.1 Swagger UI avec Flask-RESTX

Quand vous utilisez Flask-RESTX, la documentation est générée automatiquement. Améliorez-la :

```python
# app/api/incidents.py — Documentation enrichie

@ns.route("/")
class IncidentList(Resource):

    @ns.doc(
        "list_incidents",
        description="""
        Retourne la liste paginée des incidents citoyens.
        
        **Filtres disponibles :**
        - `status` : Filtrer par statut (open, in_progress, resolved, rejected)
        - `category` : Filtrer par catégorie
        - `q` : Recherche textuelle dans le titre, la description et l'adresse
        - `sort` : Tri par `vote_count` ou `created_at` (défaut: `created_at`)
        - `order` : `asc` ou `desc` (défaut: `desc`)
        
        **Exemple :**
        ```
        GET /api/v1/incidents/?status=open&category=voirie&sort=vote_count&order=desc&page=1
        ```
        """,
        params={
            "page": {"description": "Numéro de page", "type": "integer", "default": 1},
            "per_page": {"description": "Résultats par page (max 50)", "type": "integer", "default": 10},
            "status": {
                "description": "Filtrer par statut",
                "type": "string",
                "enum": ["open", "in_progress", "resolved", "rejected"]
            },
        },
        responses={
            200: "Liste des incidents",
            400: "Paramètres invalides",
        }
    )
    def get(self):
        pass
```

---

## 29.2 Versionning d'API

```python
# Stratégie 1 : URL versioning (recommandé)
# /api/v1/incidents
# /api/v2/incidents

# app/__init__.py
from app.api.v1 import api_v1_blueprint
from app.api.v2 import api_v2_blueprint

app.register_blueprint(api_v1_blueprint, url_prefix="/api/v1")
app.register_blueprint(api_v2_blueprint, url_prefix="/api/v2")

# Stratégie 2 : Header versioning
# Accept: application/vnd.urbanpulse.v2+json

@app.before_request
def route_by_version():
    accept = request.headers.get("Accept", "")
    if "v2" in accept:
        # Rediriger vers les handlers v2
        pass
```

---

<a name="chapitre-30"></a>
# [GUIDE] Chapitre 30 — Aller Plus Loin

## 30.1 WebSockets avec Flask-SocketIO

Pour les mises à jour en temps réel (ex: nouveau signalement sur la carte) :

```bash
pip install flask-socketio eventlet
```

```python
from flask_socketio import SocketIO, emit, join_room

socketio = SocketIO(cors_allowed_origins="*")

@socketio.on("connect")
def handle_connect():
    print(f"Client connecté : {request.sid}")

@socketio.on("join_city")
def handle_join(data):
    city = data.get("city", "all")
    join_room(city)
    emit("joined", {"message": f"Connecté à {city}"})

# Dans la route de création d'incident :
def notify_new_incident(incident):
    socketio.emit(
        "new_incident",
        incident.to_dict(),
        room=incident.city   # Envoyer seulement aux clients de cette ville
    )
```

---

## 30.2 Migration vers FastAPI

Si vos besoins en performance augmentent, FastAPI est le successeur naturel de Flask pour les APIs.

```python
# Même patterns, syntaxe modernisée

# Flask
@app.route("/incidents/<int:id>")
def get_incident(id):
    incident = Incident.query.get_or_404(id)
    return jsonify(incident.to_dict())

# FastAPI (équivalent)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

@app.get("/incidents/{id}")
async def get_incident(id: int) -> IncidentSchema:
    incident = await db.get(Incident, id)
    if not incident:
        raise HTTPException(status_code=404)
    return incident
```

**Quand migrer de Flask vers FastAPI ?**
- Performance : vous avez besoin de milliers de req/s avec de l'I/O async
- Validation : vous voulez la validation automatique avec Pydantic
- Documentation : OpenAPI auto généré est encore plus complet
- Standards : vous préférez les async/await natifs partout

---

## 30.3 Contribuer à l'open-source Flask

```bash
# Cloner Flask
git clone https://github.com/pallets/flask.git
cd flask

# Créer un environnement de développement
python -m venv venv && source venv/bin/activate
pip install -e ".[dev]"

# Lancer les tests Flask
pytest tests/

# Trouver des issues "good first issue"
# -> https://github.com/pallets/flask/issues?q=label%3A"good+first+issue"
```

---

## [TROPHEE] Récapitulatif Final — Parcours de la Formation

### Ce que vous savez faire maintenant

```
Niveau Débutant [OK]
├── Créer une route Flask simple
├── Rendre un template Jinja2
├── Définir un modèle SQLAlchemy
└── Faire des opérations CRUD

Niveau Intermédiaire [OK]
├── Organiser avec des Blueprints
├── Créer des formulaires WTForms
├── Gérer l'authentification Flask-Login
├── Écrire des migrations Alembic
├── Sécuriser contre CSRF/XSS/SQLi
└── Écrire des tests pytest

Niveau Avancé [OK]
├── Construire une API REST documentée (Flask-RESTX)
├── Gérer des tâches async (Celery + Redis)
├── Implémenter le caching (Redis)
├── Monitorer avec Prometheus + Grafana
├── Dockeriser l'application
├── Déployer avec Gunicorn + Nginx
├── Automatiser avec GitHub Actions CI/CD
└── Architécturer avec Service Layer

Niveau Expert [OK]
├── Architecture microservices événementielle
├── Scalabilité horizontale + load balancing
├── Read replicas + connection pooling
├── WebSockets temps réel
├── Fullstack React + Flask API
└── Bonnes pratiques production
```

---

## [LISTE] Projet Fil Rouge — État Final d'UrbanPulse

Voici ce que vous avez construit tout au long de la formation :

```
urbanpulse/                                    Fonctionnalité
├── [CITYSCAPE] Page d'accueil avec statistiques
├── [IMPORTANT] Signalement d'incidents
│   ├── Création avec formulaire + upload photo
│   ├── Liste paginée avec filtres et recherche
│   ├── Détail avec commentaires imbriqués
│   ├── Vote citoyen (1 vote/utilisateur)
│   └── Changement de statut (modérateur)
├── [UTILISATEUR] Authentification
│   ├── Inscription avec validation
│   ├── Connexion avec "se souvenir de moi"
│   ├── Profil utilisateur
│   └── Gestion des rôles (citizen/moderator/admin)
├── [PLUGIN] API REST
│   ├── CRUD complet avec JWT
│   ├── Documentation Swagger auto
│   ├── Pagination + filtres + recherche
│   └── Rate limiting
├── [EMAIL] Notifications
│   ├── Email de confirmation à la création
│   ├── Email de mise à jour de statut
│   └── Rapport hebdomadaire (Celery Beat)
├── [RAPIDE] Production
│   ├── Docker + Docker Compose
│   ├── Gunicorn + Nginx + SSL
│   ├── CI/CD GitHub Actions
│   ├── Monitoring Prometheus + Grafana
│   └── Logging structuré + Sentry
└── [TEST] Tests
    ├── 80%+ de couverture
    ├── Tests unitaires (modèles, services)
    └── Tests d'intégration (routes, API)
```

---

## [OBJECTIF] Feuille de route post-formation

### Semaine 1-2 : Consolider
- Reprendre UrbanPulse et ajouter une feature manquante de votre choix
- Atteindre 90% de couverture de tests
- Déployer sur un vrai serveur (DigitalOcean/Hetzner/OVH)

### Semaine 3-4 : Approfondir
- Lire les sources de Flask : `github.com/pallets/flask`
- Lire les sources de SQLAlchemy : comprendre les transactions
- Explorer FastAPI pour comparer

### Mois 2 : Pratiquer
- Construire un projet personnel de A à Z
- Contribuer à un projet open-source Flask
- Écrire un article de blog sur ce que vous avez appris

### Mois 3 : Spécialiser
Choisissez une spécialisation :
- **Backend API** -> FastAPI, GraphQL (Strawberry), gRPC
- **Data** -> SQLAlchemy Core, Pandas, Airflow
- **DevOps** -> Kubernetes, Terraform, ArgoCD
- **Cloud** -> AWS Lambda + Flask, Google Cloud Run

---

## [DOCS] Ressources pour aller plus loin

**Documentation officielle (indispensable) :**
- Flask : `https://flask.palletsprojects.com`
- SQLAlchemy : `https://docs.sqlalchemy.org`
- Celery : `https://docs.celeryq.dev`
- Alembic : `https://alembic.sqlalchemy.org`

**Livres :**
- *Flask Web Development* — Miguel Grinberg (O'Reilly)
- *Architecture Patterns with Python* — Harry Percival & Bob Gregory

**Cours complémentaires :**
- Real Python Flask tutorials : `realpython.com`
- Miguel Grinberg's blog : `blog.miguelgrinberg.com`

**Communautés :**
- Discord Pallets (maintient Flask) : `discord.gg/pallets`
- r/flask sur Reddit
- Stack Overflow tag `flask`

---

## [BRAVO] Félicitations !

Vous avez parcouru **30 chapitres**, construit **5 projets** (UrbanPulse + 4 projets pratiques), et maîtrisez maintenant Flask de A à Z — des premiers `@app.route` jusqu'au déploiement en production avec monitoring, CI/CD, et architecture microservices.

**UrbanPulse** n'est plus un projet fictif — c'est une vraie application production-ready que vous pouvez déployer, montrer dans votre portfolio, et faire évoluer.

*Bonne continuation dans votre aventure Flask ! [PYTHON][RAPIDE]*