# [RAPIDE] Projet Fil Rouge FastAPI — Guide Ultra-Détaillé pour Grand Débutant

> **Niveau :** Grand débutant (avec bases Python)
> **Objectif final :** Construire des APIs professionnelles, sécurisées et scalables avec FastAPI
> **Durée estimée :** 8 à 12 semaines selon votre rythme

---

## [IMPORTANT] Comment utiliser ce guide ?

Ce guide est structuré en **7 parties progressives**. Chaque chapitre contient :
- [OBJECTIF] **Les objectifs** clairs à atteindre
- [GUIDE] **Les explications ultra-détaillées** avec du code commenté
- [LOGIQUE] **Les clés de compréhension** pour bien retenir
- [OK] **Les bonnes pratiques** du monde professionnel
- [IDEE] **Des exercices pratiques** pour s'entraîner

> **Conseil :** Ne passez au chapitre suivant que lorsque vous êtes à l'aise avec le précédent. Tapez **toujours** le code vous-même, ne le copiez pas.

---

## [WORLD_MAP] Vue d'ensemble du programme

| Partie | Thème | Chapitres |
|--------|-------|-----------|
| 1 | Fondamentaux | 1 à 4 |
| 2 | Architecture & Structure | 5 à 9 |
| 3 | Sécurité & Authentification | 10 à 12 |
| 4 | Performance & Asynchronisme | 13 à 14 |
| 5 | Communication & Intégrations | 15 à 18 |
| 6 | Tests, CI/CD & Déploiement | 19 à 22 |
| 7 | Niveau Expert | 23 à 26 |

---

# [MODULE] PARTIE 1 — Fondamentaux de FastAPI

---

## Chapitre 1 : Introduction à FastAPI

### [OBJECTIF] Objectifs du chapitre
- Comprendre ce qu'est FastAPI et pourquoi l'utiliser
- Installer l'environnement de développement
- Créer et lancer votre première API fonctionnelle

---

### [GUIDE] 1.1 — Qu'est-ce que FastAPI ?

FastAPI est un **framework web Python moderne** conçu pour créer des **APIs** (Application Programming Interfaces) de manière rapide, simple et robuste.

Une **API** est une interface qui permet à deux applications de communiquer. Par exemple :
- Une application mobile qui demande la météo à un serveur -> c'est une API
- Un site web qui affiche une liste de produits depuis une base de données -> c'est une API

FastAPI a été créé en **2018** par Sebastián Ramírez. Il s'appuie sur deux piliers :
- **Starlette** : framework ASGI (gestion des requêtes HTTP, WebSockets)
- **Pydantic** : validation et sérialisation des données via les types Python

---

### [GUIDE] 1.2 — Pourquoi choisir FastAPI ?

#### [RAPIDE] Rapidité d'exécution
FastAPI est l'un des frameworks Python les plus rapides. Il est comparable en performance à **Node.js** et **Go** grâce à son support natif de la programmation asynchrone (async/await).

#### [VERROUILLE] Typage et validation automatique
FastAPI utilise les **annotations de types Python** (Python 3.7+) pour valider automatiquement les données entrantes. Si quelqu'un envoie une chaîne de caractères là où vous attendez un nombre, FastAPI renvoie une erreur claire sans que vous ayez à écrire du code de validation.

#### [FICHIER] Documentation automatique
Dès que vous écrivez un endpoint, FastAPI génère automatiquement une **documentation interactive** (Swagger UI) accessible dans le navigateur. C'est un gain de temps énorme.

#### 🆚 Comparaison avec d'autres frameworks

| Critère | FastAPI | Flask | Django |
|---------|---------|-------|--------|
| Performances | ***** | *** | ** |
| Courbe d'apprentissage | Moyenne | Faible | Élevée |
| Async natif | [OK] Oui | [X] Non | [ATTENTION] Partiel |
| Validation auto | [OK] Pydantic | [X] Manuel | [ATTENTION] Formulaires |
| Doc auto (OpenAPI) | [OK] Oui | [X] Non | [X] Non |
| Idéal pour | APIs modernes | APIs simples | Apps web complètes |

---

### [GUIDE] 1.3 — Installation de l'environnement

#### Étape 1 : Vérifier Python
```bash
python --version
# Doit afficher Python 3.8 ou supérieur
```

#### Étape 2 : Créer un environnement virtuel
Un environnement virtuel isole les dépendances de votre projet. C'est une bonne pratique indispensable.

```bash
# Créer un dossier pour votre projet
mkdir mon_api_fastapi
cd mon_api_fastapi

# Créer l'environnement virtuel
python -m venv venv

# Activer l'environnement virtuel
# Sur Windows :
venv\Scripts\activate
# Sur Mac/Linux :
source venv/bin/activate

# Votre terminal affiche maintenant (venv) devant le prompt
```

#### Étape 3 : Installer FastAPI et Uvicorn
```bash
pip install fastapi uvicorn

# Pour avoir toutes les dépendances optionnelles utiles :
pip install "fastapi[all]"
```

> **Qu'est-ce qu'Uvicorn ?**
> Uvicorn est un **serveur ASGI** (Asynchronous Server Gateway Interface). C'est lui qui "écoute" les requêtes HTTP et les passe à votre application FastAPI. Sans serveur, votre code Python ne peut pas recevoir de requêtes web.

---

### [GUIDE] 1.4 — Votre première API "Hello World"

Créez un fichier `main.py` dans votre dossier de projet :

```python
# main.py

from fastapi import FastAPI  # On importe la classe principale

# On crée une instance de l'application FastAPI
# C'est le "cœur" de votre API
app = FastAPI()

# On définit une route avec le décorateur @app.get()
# @app.get("/") signifie : quand quelqu'un fait un GET sur "/", exécuter cette fonction
@app.get("/")
def lire_racine():
    # On retourne un dictionnaire Python
    # FastAPI le convertit automatiquement en JSON
    return {"message": "Bonjour depuis FastAPI !"}
```

#### Lancer l'application

```bash
uvicorn main:app --reload
```

**Décodons cette commande :**
- `uvicorn` : le serveur
- `main` : le nom du fichier Python (sans `.py`)
- `app` : le nom de la variable FastAPI dans ce fichier
- `--reload` : redémarre automatiquement le serveur à chaque modification du code

**Sortie attendue dans le terminal :**
```
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [...]
INFO:     Started server process [...]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
```

Ouvrez votre navigateur sur **http://127.0.0.1:8000** et vous verrez :
```json
{"message": "Bonjour depuis FastAPI !"}
```

---

### [LOGIQUE] Clés à retenir

- `FastAPI()` crée l'instance de votre application
- Les **décorateurs** (`@app.get(...)`) définissent les routes
- FastAPI convertit automatiquement les dictionnaires Python en **JSON**
- Uvicorn est le serveur qui fait tourner votre app

### [OK] Bonne pratique

> Toujours garder `--reload` en **développement** uniquement. En production, ne jamais l'utiliser car cela ralentit le serveur et consomme des ressources inutilement.

### [IDEE] Exercice pratique

Ajoutez une deuxième route `/bonjour` qui retourne votre prénom et la date du jour. Utilisez le module `datetime` de Python.

---

## Chapitre 2 : Routes et Méthodes HTTP

### [OBJECTIF] Objectifs du chapitre
- Comprendre les méthodes HTTP (GET, POST, PUT, DELETE, PATCH)
- Maîtriser les paramètres de route, de requête et le corps de requête
- Structurer des endpoints clairs et bien nommés

---

### [GUIDE] 2.1 — Les méthodes HTTP

Le protocole HTTP définit des **méthodes** qui indiquent l'**intention** d'une requête. C'est comme un langage : chaque mot a un sens précis.

| Méthode | Usage | Exemple |
|---------|-------|---------|
| `GET` | Lire/récupérer des données | Obtenir la liste des utilisateurs |
| `POST` | Créer une nouvelle ressource | Créer un nouvel utilisateur |
| `PUT` | Remplacer entièrement une ressource | Mettre à jour toutes les infos d'un utilisateur |
| `PATCH` | Modifier partiellement une ressource | Changer uniquement l'email d'un utilisateur |
| `DELETE` | Supprimer une ressource | Supprimer un utilisateur |

```python
# main.py — Démonstration des méthodes HTTP

from fastapi import FastAPI

app = FastAPI()

# GET — Lire
@app.get("/utilisateurs")
def lire_utilisateurs():
    return [{"id": 1, "nom": "Alice"}, {"id": 2, "nom": "Bob"}]

# POST — Créer
@app.post("/utilisateurs")
def creer_utilisateur():
    return {"message": "Utilisateur créé"}

# PUT — Remplacer
@app.put("/utilisateurs/{id}")
def remplacer_utilisateur(id: int):
    return {"message": f"Utilisateur {id} remplacé"}

# PATCH — Modifier partiellement
@app.patch("/utilisateurs/{id}")
def modifier_utilisateur(id: int):
    return {"message": f"Utilisateur {id} modifié"}

# DELETE — Supprimer
@app.delete("/utilisateurs/{id}")
def supprimer_utilisateur(id: int):
    return {"message": f"Utilisateur {id} supprimé"}
```

---

### [GUIDE] 2.2 — Paramètres de route (Path Parameters)

Un **paramètre de route** est une partie variable de l'URL, entourée d'accolades `{}`.

```python
@app.get("/utilisateurs/{user_id}")
def lire_utilisateur(user_id: int):
    # user_id est automatiquement converti en entier grâce au type int
    # Si on passe une chaîne, FastAPI renvoie une erreur automatiquement
    return {"id": user_id, "nom": "Alice"}

# Exemple d'appel : GET /utilisateurs/42
# Résultat : {"id": 42, "nom": "Alice"}
```

**Ordre des routes — IMPORTANT !**
```python
# [ATTENTION] L'ordre des routes est crucial. FastAPI les évalue de haut en bas.

@app.get("/utilisateurs/moi")  # Cette route DOIT être AVANT la suivante
def lire_mon_profil():
    return {"nom": "moi-même"}

@app.get("/utilisateurs/{user_id}")  # Sinon "moi" serait capturé comme user_id
def lire_utilisateur(user_id: int):
    return {"id": user_id}
```

---

### [GUIDE] 2.3 — Paramètres de requête (Query Parameters)

Les **paramètres de requête** sont passés après un `?` dans l'URL : `/items?skip=0&limit=10`

```python
@app.get("/articles")
def lire_articles(skip: int = 0, limit: int = 10):
    # skip et limit ont des valeurs par défaut
    # Appel : GET /articles -> skip=0, limit=10
    # Appel : GET /articles?skip=5&limit=3 -> skip=5, limit=3
    articles = [{"id": i, "titre": f"Article {i}"} for i in range(1, 100)]
    return articles[skip : skip + limit]

@app.get("/recherche")
def rechercher(q: str):
    # q est obligatoire (pas de valeur par défaut)
    # Appel : GET /recherche?q=python
    return {"resultats": f"Recherche pour : {q}"}

@app.get("/filtrer")
def filtrer(actif: bool = True, categorie: str | None = None):
    # categorie est optionnelle (peut être None)
    return {"actif": actif, "categorie": categorie}
```

---

### [GUIDE] 2.4 — Corps de requête (Request Body)

Pour envoyer des données structurées (ex: lors d'un POST), on utilise un **corps de requête** en JSON. Nous verrons Pydantic en détail au chapitre 3, mais voici un aperçu :

```python
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# On définit la structure attendue dans le corps
class ArticleCreation(BaseModel):
    titre: str
    contenu: str
    publie: bool = False  # Valeur par défaut

@app.post("/articles")
def creer_article(article: ArticleCreation):
    # FastAPI lit automatiquement le JSON du corps
    # et le valide selon le modèle ArticleCreation
    return {
        "message": "Article créé",
        "titre": article.titre,
        "publie": article.publie
    }
```

**Corps JSON à envoyer (via Swagger ou Postman) :**
```json
{
    "titre": "Mon premier article",
    "contenu": "Contenu de l'article"
}
```

---

### [GUIDE] 2.5 — Combiner tous les types de paramètres

```python
@app.put("/articles/{article_id}")
def mettre_a_jour_article(
    article_id: int,           # Paramètre de route
    version: int = 1,          # Paramètre de requête
    article: ArticleCreation   # Corps de requête
):
    # FastAPI distingue automatiquement les trois types :
    # - article_id -> dans l'URL /articles/5
    # - version -> dans la query string ?version=2
    # - article -> dans le body JSON
    return {
        "id": article_id,
        "version": version,
        "titre": article.titre
    }
```

---

### [LOGIQUE] Clés à retenir

- **Route** : la structure de l'URL (`/utilisateurs/{id}`)
- **Path param** : dans l'URL entre accolades `{id}`
- **Query param** : après le `?` dans l'URL
- **Body** : données JSON dans le corps de la requête
- FastAPI distingue ces trois types **automatiquement** selon comment vous déclarez les paramètres

### [OK] Bonne pratique

> Utilisez des noms de ressources au **pluriel** et en **minuscules** pour vos endpoints : `/utilisateurs`, `/articles`, `/commandes`. Évitez les verbes dans les URLs (`/getUtilisateurs` est une mauvaise pratique).

### [IDEE] Exercice pratique

Créez une mini-API de gestion de livres avec les routes :
- `GET /livres` — liste tous les livres (avec pagination skip/limit)
- `GET /livres/{livre_id}` — obtient un livre par ID
- `POST /livres` — crée un livre (titre, auteur, année)
- `DELETE /livres/{livre_id}` — supprime un livre

---

## Chapitre 3 : Typage et Validation Automatique avec Pydantic

### [OBJECTIF] Objectifs du chapitre
- Comprendre Pydantic et son rôle dans FastAPI
- Créer des modèles de données robustes
- Utiliser les contraintes de validation
- Comprendre la sérialisation/désérialisation JSON

---

### [GUIDE] 3.1 — Introduction à Pydantic

**Pydantic** est une bibliothèque Python de **validation de données**. Elle utilise les annotations de types Python pour valider automatiquement que les données reçues correspondent au format attendu.

Sans Pydantic, vous devriez écrire manuellement :
```python
# [X] Sans Pydantic — code fastidieux et source d'erreurs
def creer_utilisateur(data: dict):
    if "nom" not in data:
        raise ValueError("Le champ 'nom' est requis")
    if not isinstance(data["nom"], str):
        raise ValueError("Le champ 'nom' doit être une chaîne")
    if len(data["nom"]) < 2:
        raise ValueError("Le nom doit faire au moins 2 caractères")
    # ... etc
```

Avec Pydantic :
```python
# [OK] Avec Pydantic — simple, lisible et robuste
from pydantic import BaseModel, Field

class Utilisateur(BaseModel):
    nom: str = Field(min_length=2, description="Prénom et nom de l'utilisateur")
    age: int = Field(ge=0, le=150)  # ge = greater or equal, le = less or equal
```

---

### [GUIDE] 3.2 — Modèles de base (`BaseModel`)

```python
from pydantic import BaseModel
from typing import Optional

class Produit(BaseModel):
    # Champ obligatoire (pas de valeur par défaut)
    nom: str
    prix: float
    
    # Champ avec valeur par défaut
    en_stock: bool = True
    
    # Champ optionnel (peut être None)
    description: Optional[str] = None
    # Depuis Python 3.10, on peut aussi écrire : description: str | None = None

# Utilisation en dehors de FastAPI :
produit = Produit(nom="Clavier", prix=49.99)
print(produit.nom)        # "Clavier"
print(produit.en_stock)   # True (valeur par défaut)
print(produit.description) # None

# Sérialisation en dictionnaire
print(produit.model_dump())
# {'nom': 'Clavier', 'prix': 49.99, 'en_stock': True, 'description': None}

# Sérialisation en JSON
print(produit.model_dump_json())
# '{"nom":"Clavier","prix":49.99,"en_stock":true,"description":null}'
```

---

### [GUIDE] 3.3 — Contraintes de validation avec `Field`

```python
from pydantic import BaseModel, Field, EmailStr
from typing import List

class UtilisateurComplet(BaseModel):
    # Contraintes sur les chaînes
    nom: str = Field(
        min_length=2,
        max_length=100,
        description="Nom complet de l'utilisateur"
    )
    
    # Validation d'email (nécessite pip install pydantic[email])
    email: EmailStr
    
    # Contraintes sur les nombres
    age: int = Field(ge=18, le=120)  # Entre 18 et 120
    score: float = Field(ge=0.0, le=10.0, default=0.0)
    
    # Contrainte de pattern (expression régulière)
    code_postal: str = Field(pattern=r"^\d{5}$")  # 5 chiffres exactement
    
    # Liste avec contraintes
    tags: List[str] = Field(default=[], max_length=10)
```

---

### [GUIDE] 3.4 — Validation personnalisée

```python
from pydantic import BaseModel, field_validator, model_validator

class Commande(BaseModel):
    article: str
    quantite: int
    prix_unitaire: float
    
    # Validateur sur un champ spécifique
    @field_validator("article")
    @classmethod
    def article_non_vide(cls, valeur):
        if not valeur.strip():
            raise ValueError("L'article ne peut pas être une chaîne vide")
        return valeur.strip()  # On peut aussi transformer la valeur
    
    # Validateur sur tout le modèle (pour validation croisée)
    @model_validator(mode="after")
    def verifier_commande(self):
        if self.quantite <= 0:
            raise ValueError("La quantité doit être positive")
        if self.prix_unitaire <= 0:
            raise ValueError("Le prix doit être positif")
        return self
    
    # Propriété calculée
    @property
    def total(self) -> float:
        return self.quantite * self.prix_unitaire
```

---

### [GUIDE] 3.5 — Modèles imbriqués et héritage

```python
from pydantic import BaseModel
from typing import List
from datetime import datetime

class Adresse(BaseModel):
    rue: str
    ville: str
    code_postal: str
    pays: str = "France"

class Client(BaseModel):
    id: int
    nom: str
    email: str
    adresse: Adresse  # Modèle imbriqué
    commandes: List[int] = []  # Liste d'IDs de commandes

# Désérialisation depuis JSON (FastAPI fait ça automatiquement)
donnees_json = {
    "id": 1,
    "nom": "Alice Dupont",
    "email": "alice@exemple.com",
    "adresse": {
        "rue": "10 rue de la Paix",
        "ville": "Paris",
        "code_postal": "75001"
    }
}
client = Client(**donnees_json)
print(client.adresse.ville)  # "Paris"
```

---

### [GUIDE] 3.6 — Modèles de réponse séparés

En pratique, on crée souvent **plusieurs modèles** pour le même objet :

```python
# schemas/utilisateur.py

from pydantic import BaseModel, EmailStr

# Champs communs
class UtilisateurBase(BaseModel):
    nom: str
    email: EmailStr

# Pour la création (mot de passe requis)
class UtilisateurCreer(UtilisateurBase):
    mot_de_passe: str

# Pour la réponse (jamais de mot de passe !)
class UtilisateurReponse(UtilisateurBase):
    id: int
    actif: bool = True
    
    class Config:
        from_attributes = True  # Pour convertir depuis un ORM
```

```python
# Utilisation dans une route
from fastapi import FastAPI

app = FastAPI()

@app.post("/utilisateurs", response_model=UtilisateurReponse)
def creer_utilisateur(utilisateur: UtilisateurCreer):
    # FastAPI filtre automatiquement la réponse selon UtilisateurReponse
    # Le mot de passe ne sera JAMAIS retourné dans la réponse
    return {
        "id": 1,
        "nom": utilisateur.nom,
        "email": utilisateur.email,
        "actif": True
    }
```

---

### [LOGIQUE] Clés à retenir

- Pydantic valide **automatiquement** les entrées et lance une erreur HTTP 422 si invalide
- Chaque champ typé dans un `BaseModel` est validé selon son type
- `Field()` permet d'ajouter des contraintes supplémentaires
- FastAPI lit le schéma Pydantic pour générer la **documentation OpenAPI automatiquement**
- Utiliser des modèles distincts (création vs réponse) est une bonne pratique de sécurité

### [OK] Bonne pratique

> Toujours définir des types **explicites** sur tous vos champs. Évitez `Any` sauf cas exceptionnel. Plus vous êtes précis dans vos types, plus votre API est robuste et auto-documentée.

### [IDEE] Exercice pratique

Créez un modèle Pydantic `Article` avec les contraintes suivantes :
- `titre` : entre 5 et 200 caractères
- `contenu` : minimum 100 caractères
- `note` : entre 0 et 5 (float)
- `tags` : liste de chaînes, max 5 tags, chaque tag max 20 caractères
- `date_publication` : type `datetime`, optionnelle

---

## Chapitre 4 : Documentation Automatique (Swagger UI & ReDoc)

### [OBJECTIF] Objectifs du chapitre
- Accéder et utiliser la documentation interactive
- Personnaliser l'apparence et les informations de votre API
- Documenter chaque route pour la maintenance

---

### [GUIDE] 4.1 — Accéder à la documentation

FastAPI génère **deux interfaces de documentation** automatiquement, sans aucune configuration :

| Interface | URL | Description |
|-----------|-----|-------------|
| **Swagger UI** | `http://127.0.0.1:8000/docs` | Interactive, permet de tester les endpoints |
| **ReDoc** | `http://127.0.0.1:8000/redoc` | Plus lisible, idéale pour partager |
| **OpenAPI JSON** | `http://127.0.0.1:8000/openapi.json` | Schéma JSON brut |

---

### [GUIDE] 4.2 — Personnaliser la documentation

```python
from fastapi import FastAPI

app = FastAPI(
    title="Mon API de Gestion de Livres",
    description="""
    ## API REST pour gérer une bibliothèque

    ### Fonctionnalités :
    * [DOCS] Gestion des livres (CRUD complet)
    * [UTILISATEUR] Gestion des auteurs
    * [RECHERCHE] Recherche avancée

    Développée avec FastAPI et Python.
    """,
    version="1.0.0",
    contact={
        "name": "Support Technique",
        "email": "support@mabibliotheque.com",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
)
```

---

### [GUIDE] 4.3 — Documenter les routes avec des tags et descriptions

```python
from fastapi import FastAPI

app = FastAPI(title="API Bibliothèque")

# Les tags permettent de regrouper les routes dans la doc
@app.get(
    "/livres",
    tags=["[DOCS] Livres"],  # Groupe dans la documentation
    summary="Liste tous les livres",  # Titre court
    description="""
    Retourne la liste complète des livres disponibles.
    
    Supports la pagination avec les paramètres **skip** et **limit**.
    """,
    response_description="Liste des livres avec leurs métadonnées",
)
def lire_livres(skip: int = 0, limit: int = 10):
    return []

@app.post(
    "/livres",
    tags=["[DOCS] Livres"],
    summary="Créer un nouveau livre",
    status_code=201,  # Code HTTP de succès (201 = Created)
)
def creer_livre():
    return {}

@app.get(
    "/auteurs",
    tags=["[UTILISATEUR] Auteurs"],
    summary="Liste tous les auteurs",
)
def lire_auteurs():
    return []
```

---

### [GUIDE] 4.4 — Documenter avec des docstrings

Une façon plus pythonique de documenter vos routes est d'utiliser les **docstrings** :

```python
@app.get("/livres/{livre_id}", tags=["[DOCS] Livres"])
def lire_livre(livre_id: int):
    """
    Récupère un livre par son identifiant unique.
    
    - **livre_id** : L'identifiant entier du livre
    - Retourne une erreur 404 si le livre n'existe pas
    """
    return {"id": livre_id, "titre": "Exemple"}
```

FastAPI convertit automatiquement la docstring en description dans Swagger UI.

---

### [GUIDE] 4.5 — Exemples dans la documentation

```python
from pydantic import BaseModel, Field

class LivreCreation(BaseModel):
    titre: str = Field(
        ...,  # ... signifie champ obligatoire
        examples=["Le Petit Prince"],  # Exemple dans la doc
        description="Titre complet du livre"
    )
    auteur: str = Field(examples=["Antoine de Saint-Exupéry"])
    annee: int = Field(ge=1000, le=2100, examples=[1943])
    
    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "titre": "Le Petit Prince",
                    "auteur": "Antoine de Saint-Exupéry",
                    "annee": 1943
                }
            ]
        }
    }
```

---

### [GUIDE] 4.6 — Désactiver la documentation en production

En production, il peut être souhaitable de désactiver la documentation pour des raisons de sécurité :

```python
from fastapi import FastAPI
import os

# Variable d'environnement pour contrôler l'environnement
env = os.getenv("ENVIRONNEMENT", "development")

app = FastAPI(
    title="Mon API",
    # Désactiver la doc en production
    docs_url="/docs" if env != "production" else None,
    redoc_url="/redoc" if env != "production" else None,
)
```

---

### [LOGIQUE] Clés à retenir

- La documentation est **générée automatiquement** à partir du code — zéro configuration
- Swagger UI (`/docs`) permet de **tester directement** les endpoints dans le navigateur
- Les `tags` regroupent les routes logiquement dans la documentation
- Les `summary` et `description` (ou docstrings) enrichissent la documentation

### [OK] Bonne pratique

> Documentez **toutes** vos routes, même les plus simples. Une API bien documentée est plus facile à maintenir, à comprendre pour les nouveaux développeurs, et à utiliser pour les clients de votre API. La documentation est une forme de respect envers les autres (et envers votre futur vous).

### [IDEE] Exercice final de la Partie 1

Créez une API complète "Gestion de Contacts" qui combine tout ce que vous avez appris :

1. Un modèle Pydantic `Contact` avec validation (nom, email, téléphone, adresse)
2. Des routes GET, POST, PUT, DELETE pour les contacts
3. Une pagination sur la liste des contacts
4. Une recherche par nom (`?nom=alice`)
5. Une documentation Swagger bien organisée avec tags et descriptions

---

*Fin de la Partie 1 — Fondamentaux*

> **Prochaine étape :** Partie 2 — Architecture et Structuration du projet


# [RAPIDE] FastAPI — Fil Rouge Étudiant
# [CONFIG] PARTIE 2 — Architecture & Structuration (Niveau Intermédiaire)

> **Prérequis :** Avoir complété la Partie 1 (Chapitres 1 à 4)
> **Objectif de cette partie :** Apprendre à structurer un projet professionnel, gérer des bases de données, implémenter le CRUD, et gérer les erreurs.

---

## Chapitre 5 : Structuration du Projet FastAPI

### [OBJECTIF] Objectifs du chapitre
- Organiser un projet FastAPI de manière modulaire et professionnelle
- Utiliser `APIRouter` pour séparer les routes
- Comprendre le principe de séparation des responsabilités

---

### [GUIDE] 5.1 — Pourquoi structurer son projet ?

Quand votre API grandit, mettre tout dans un seul `main.py` devient ingérable :
- Fichier énorme, difficile à lire
- Difficile de travailler en équipe
- Impossible de tester unitairement
- Couplage fort entre les composants

La solution est d'adopter une **architecture modulaire** : 1 fichier = 1 responsabilité.

---

### [GUIDE] 5.2 — Structure recommandée

```
mon_api/
├── main.py                    # Point d'entrée de l'application
├── requirements.txt           # Dépendances Python
├── .env                       # Variables d'environnement (ne jamais committer !)
├── .gitignore
│
├── app/
│   ├── __init__.py
│   │
│   ├── api/                   # Tout ce qui concerne l'API HTTP
│   │   ├── __init__.py
│   │   └── routes/            # Un fichier par ressource
│   │       ├── __init__.py
│   │       ├── utilisateurs.py
│   │       ├── articles.py
│   │       └── commandes.py
│   │
│   ├── core/                  # Configuration centrale
│   │   ├── __init__.py
│   │   ├── config.py          # Paramètres de l'application
│   │   └── security.py        # Utilitaires de sécurité
│   │
│   ├── db/                    # Tout ce qui concerne la base de données
│   │   ├── __init__.py
│   │   ├── session.py         # Connexion à la base
│   │   └── base.py            # Modèle de base ORM
│   │
│   ├── models/                # Modèles ORM (représentation en base de données)
│   │   ├── __init__.py
│   │   ├── utilisateur.py
│   │   └── article.py
│   │
│   └── schemas/               # Schémas Pydantic (validation des données API)
│       ├── __init__.py
│       ├── utilisateur.py
│       └── article.py
│
└── tests/                     # Tests automatisés
    ├── __init__.py
    └── test_utilisateurs.py
```

---

### [GUIDE] 5.3 — Utilisation d'`APIRouter`

`APIRouter` est l'équivalent d'un "mini-FastAPI" pour grouper des routes liées.

```python
# app/api/routes/utilisateurs.py

from fastapi import APIRouter, HTTPException

# Création du routeur avec préfixe et tag pour la doc
router = APIRouter(
    prefix="/utilisateurs",  # Toutes les routes commencent par /utilisateurs
    tags=["[UTILISATEUR] Utilisateurs"],
    responses={404: {"description": "Utilisateur non trouvé"}},
)

# Ces routes sont automatiquement préfixées
@router.get("/")                       # -> GET /utilisateurs/
def lire_utilisateurs():
    return [{"id": 1, "nom": "Alice"}]

@router.get("/{user_id}")              # -> GET /utilisateurs/{user_id}
def lire_utilisateur(user_id: int):
    return {"id": user_id, "nom": "Alice"}

@router.post("/", status_code=201)     # -> POST /utilisateurs/
def creer_utilisateur():
    return {"id": 2, "nom": "Bob"}
```

```python
# app/api/routes/articles.py

from fastapi import APIRouter

router = APIRouter(
    prefix="/articles",
    tags=["[NEWSPAPER] Articles"],
)

@router.get("/")
def lire_articles():
    return []
```

---

### [GUIDE] 5.4 — Assembler les routes dans `main.py`

```python
# main.py

from fastapi import FastAPI
from app.api.routes import utilisateurs, articles

app = FastAPI(
    title="Mon API Modulaire",
    version="1.0.0",
)

# Inclusion des routeurs
# Les préfixes et tags définis dans chaque routeur sont respectés
app.include_router(utilisateurs.router)
app.include_router(articles.router)

# Route racine dans main.py
@app.get("/", tags=["[ACCUEIL] Accueil"])
def accueil():
    return {"message": "Bienvenue sur Mon API Modulaire"}
```

---

### [GUIDE] 5.5 — Configuration centralisée

```python
# app/core/config.py

from pydantic_settings import BaseSettings  # pip install pydantic-settings

class Parametres(BaseSettings):
    # Ces valeurs sont lues depuis les variables d'environnement ou le fichier .env
    nom_app: str = "Mon API"
    debug: bool = False
    version: str = "1.0.0"
    
    # Base de données
    url_base_de_donnees: str = "sqlite:///./test.db"
    
    # Sécurité
    cle_secrete: str = "changez-moi-en-production"
    algorithme_jwt: str = "HS256"
    duree_token_minutes: int = 30
    
    class Config:
        env_file = ".env"  # Lit depuis le fichier .env
        case_sensitive = False

# Instance globale — importée depuis tous les modules
parametres = Parametres()
```

```bash
# .env — Variables d'environnement (ne JAMAIS committer ce fichier !)
NOM_APP="Mon API Prod"
DEBUG=false
URL_BASE_DE_DONNEES=postgresql://user:password@localhost/mabase
CLE_SECRETE=une-cle-tres-longue-et-aleatoire-ici
```

---

### [LOGIQUE] Clés à retenir

- `APIRouter` permet de **découpler** les routes par ressource
- `include_router()` dans `main.py` assemble les modules
- La configuration doit être **centralisée** dans `core/config.py`
- Les variables sensibles vivent dans `.env`, jamais dans le code

### [OK] Bonne pratique

> Un fichier = une responsabilité. Les modèles ORM (`models/`) ne contiennent que la structure de la base de données. Les schémas Pydantic (`schemas/`) ne contiennent que la validation API. Les routes (`api/routes/`) ne contiennent que la logique HTTP.

---

## Chapitre 6 : Gestion des Données — Bases de Données avec SQLAlchemy

### [OBJECTIF] Objectifs du chapitre
- Connecter FastAPI à une base de données (SQLite, PostgreSQL)
- Comprendre les ORM et SQLAlchemy
- Distinguer modèles ORM et schémas Pydantic

---

### [GUIDE] 6.1 — Qu'est-ce qu'un ORM ?

Un **ORM** (Object-Relational Mapper) permet de manipuler une base de données SQL **en utilisant du Python** au lieu d'écrire du SQL brut.

```python
# [X] Sans ORM — SQL brut
cursor.execute("SELECT * FROM utilisateurs WHERE id = ?", (user_id,))
row = cursor.fetchone()
utilisateur = {"id": row[0], "nom": row[1]}

# [OK] Avec SQLAlchemy ORM
utilisateur = session.get(Utilisateur, user_id)
```

---

### [GUIDE] 6.2 — Installation et configuration

```bash
pip install sqlalchemy aiosqlite  # Pour SQLite async
# ou
pip install sqlalchemy asyncpg    # Pour PostgreSQL async
```

```python
# app/db/session.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# URL de la base de données
# SQLite (développement) :
SQLITE_URL = "sqlite+aiosqlite:///./ma_base.db"
# PostgreSQL (production) :
# POSTGRES_URL = "postgresql+asyncpg://user:password@localhost/mabase"

# Moteur de base de données
engine = create_async_engine(
    SQLITE_URL,
    echo=True,  # Affiche les requêtes SQL dans le terminal (dev seulement)
)

# Factory de sessions
AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# Classe de base pour tous les modèles ORM
class Base(DeclarativeBase):
    pass

# Dépendance FastAPI pour obtenir une session
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # Fournit la session à la route
            await session.commit()  # Valide si tout s'est bien passé
        except Exception:
            await session.rollback()  # Annule en cas d'erreur
            raise
        finally:
            await session.close()  # Toujours fermer la session
```

---

### [GUIDE] 6.3 — Définir des modèles ORM

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

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from app.db.session import Base

class Utilisateur(Base):
    __tablename__ = "utilisateurs"  # Nom de la table en base
    
    id = Column(Integer, primary_key=True, index=True)
    nom = Column(String(100), nullable=False)
    email = Column(String(200), unique=True, index=True, nullable=False)
    mot_de_passe_hash = Column(String, nullable=False)
    actif = Column(Boolean, default=True)
    
    # Timestamps automatiques
    cree_le = Column(DateTime, server_default=func.now())
    modifie_le = Column(DateTime, onupdate=func.now())
```

---

### [GUIDE] 6.4 — Schémas Pydantic distincts des modèles ORM

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

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional

# Base commune
class UtilisateurBase(BaseModel):
    nom: str
    email: EmailStr

# Création (avec mot de passe)
class UtilisateurCreer(UtilisateurBase):
    mot_de_passe: str

# Mise à jour (tous les champs optionnels)
class UtilisateurModifier(BaseModel):
    nom: Optional[str] = None
    email: Optional[EmailStr] = None

# Réponse API (jamais de mot de passe !)
class UtilisateurReponse(UtilisateurBase):
    id: int
    actif: bool
    cree_le: datetime
    
    class Config:
        from_attributes = True  # Convertit l'objet ORM en schema Pydantic
```

---

### [GUIDE] 6.5 — Initialisation de la base de données

```python
# app/db/init_db.py

from sqlalchemy.ext.asyncio import create_async_engine
from app.db.session import Base, engine
from app.models import utilisateur  # Import pour que les modèles soient enregistrés

async def init_db():
    async with engine.begin() as conn:
        # Crée toutes les tables si elles n'existent pas
        await conn.run_sync(Base.metadata.create_all)
```

```python
# main.py — Ajouter le démarrage de la DB

from fastapi import FastAPI
from app.db.init_db import init_db

app = FastAPI()

@app.on_event("startup")
async def startup():
    await init_db()  # Créer les tables au démarrage
```

---

### [LOGIQUE] Clés à retenir

- **Modèle ORM** = représentation d'une table SQL en Python (dans `models/`)
- **Schéma Pydantic** = validation/sérialisation des données API (dans `schemas/`)
- Ces deux notions sont **distinctes** et ne doivent pas être mélangées
- La session de base de données est injectée via une **dépendance FastAPI**

### [OK] Bonne pratique

> Ne mettez jamais de logique métier dans les modèles ORM. La logique appartient à une couche de service (voir architecture hexagonale au Chapitre 23). Les modèles ORM ne font que décrire la structure de la table.

---

## Chapitre 7 : CRUD Complet

### [OBJECTIF] Objectifs du chapitre
- Implémenter les 4 opérations CRUD (Create, Read, Update, Delete)
- Gérer correctement les codes HTTP de réponse
- Utiliser l'injection de dépendances pour la base de données

---

### [GUIDE] 7.1 — Injection de dépendances

FastAPI dispose d'un système d'**injection de dépendances** puissant. Il permet d'injecter automatiquement des ressources (session DB, utilisateur courant, etc.) dans vos routes.

```python
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db

# La session DB est injectée automatiquement par FastAPI
@router.get("/utilisateurs/{user_id}")
async def lire_utilisateur(
    user_id: int,
    db: AsyncSession = Depends(get_db)  # Injection de la session
):
    # 'db' est maintenant disponible pour faire des requêtes
    utilisateur = await db.get(Utilisateur, user_id)
    return utilisateur
```

---

### [GUIDE] 7.2 — Implémenter le CRUD complet

```python
# app/api/routes/utilisateurs.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.db.session import get_db
from app.models.utilisateur import Utilisateur as UtilisateurORM
from app.schemas.utilisateur import (
    UtilisateurCreer, UtilisateurReponse, UtilisateurModifier
)
from typing import List

router = APIRouter(prefix="/utilisateurs", tags=["[UTILISATEUR] Utilisateurs"])

# ──────────────────────────────────────
# READ — Lire tous les utilisateurs
# ──────────────────────────────────────
@router.get("/", response_model=List[UtilisateurReponse])
async def lire_utilisateurs(
    skip: int = 0,
    limit: int = 10,
    db: AsyncSession = Depends(get_db)
):
    resultat = await db.execute(
        select(UtilisateurORM).offset(skip).limit(limit)
    )
    return resultat.scalars().all()

# ──────────────────────────────────────
# READ — Lire un utilisateur par ID
# ──────────────────────────────────────
@router.get("/{user_id}", response_model=UtilisateurReponse)
async def lire_utilisateur(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    utilisateur = await db.get(UtilisateurORM, user_id)
    
    if not utilisateur:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Utilisateur {user_id} non trouvé"
        )
    
    return utilisateur

# ──────────────────────────────────────
# CREATE — Créer un utilisateur
# ──────────────────────────────────────
@router.post("/", response_model=UtilisateurReponse, status_code=status.HTTP_201_CREATED)
async def creer_utilisateur(
    utilisateur_data: UtilisateurCreer,
    db: AsyncSession = Depends(get_db)
):
    # Vérifier si l'email existe déjà
    resultat = await db.execute(
        select(UtilisateurORM).where(UtilisateurORM.email == utilisateur_data.email)
    )
    if resultat.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Cet email est déjà utilisé"
        )
    
    # Hasher le mot de passe (voir Chapitre 10)
    mot_de_passe_hash = "hash_" + utilisateur_data.mot_de_passe  # Temporaire !
    
    # Créer l'objet ORM
    nouvel_utilisateur = UtilisateurORM(
        nom=utilisateur_data.nom,
        email=utilisateur_data.email,
        mot_de_passe_hash=mot_de_passe_hash,
    )
    
    db.add(nouvel_utilisateur)
    await db.commit()
    await db.refresh(nouvel_utilisateur)  # Recharge depuis la DB pour avoir l'ID
    
    return nouvel_utilisateur

# ──────────────────────────────────────
# UPDATE — Modifier partiellement
# ──────────────────────────────────────
@router.patch("/{user_id}", response_model=UtilisateurReponse)
async def modifier_utilisateur(
    user_id: int,
    donnees: UtilisateurModifier,
    db: AsyncSession = Depends(get_db)
):
    utilisateur = await db.get(UtilisateurORM, user_id)
    
    if not utilisateur:
        raise HTTPException(status_code=404, detail="Utilisateur non trouvé")
    
    # Mettre à jour uniquement les champs fournis
    donnees_dict = donnees.model_dump(exclude_unset=True)  # Ignore les champs non envoyés
    for champ, valeur in donnees_dict.items():
        setattr(utilisateur, champ, valeur)
    
    await db.commit()
    await db.refresh(utilisateur)
    return utilisateur

# ──────────────────────────────────────
# DELETE — Supprimer
# ──────────────────────────────────────
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def supprimer_utilisateur(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    utilisateur = await db.get(UtilisateurORM, user_id)
    
    if not utilisateur:
        raise HTTPException(status_code=404, detail="Utilisateur non trouvé")
    
    await db.delete(utilisateur)
    await db.commit()
    # 204 No Content ne retourne rien
```

---

### [GUIDE] 7.3 — Codes HTTP importants

| Code | Nom | Utilisation |
|------|-----|-------------|
| 200 | OK | Succès général (GET, PUT, PATCH) |
| 201 | Created | Ressource créée (POST) |
| 204 | No Content | Succès sans corps de réponse (DELETE) |
| 400 | Bad Request | Requête invalide (email déjà utilisé) |
| 401 | Unauthorized | Non authentifié |
| 403 | Forbidden | Authentifié mais pas autorisé |
| 404 | Not Found | Ressource introuvable |
| 422 | Unprocessable Entity | Erreur de validation Pydantic |
| 500 | Internal Server Error | Erreur serveur inattendue |

---

### [LOGIQUE] Clés à retenir

- `Depends(get_db)` injecte automatiquement la session de base de données
- Toujours vérifier que la ressource existe avant de la modifier/supprimer
- `model_dump(exclude_unset=True)` est crucial pour les opérations PATCH
- Le code HTTP doit être **cohérent** avec l'action réalisée

### [OK] Bonne pratique

> Séparez la logique métier des routes ! Idéalement, créez un fichier `app/services/utilisateurs.py` avec des fonctions comme `get_utilisateur_by_id()`, `creer_utilisateur()`, etc. Les routes ne font qu'appeler ces services.

---

## Chapitre 8 : Gestion des Erreurs & Exceptions

### [OBJECTIF] Objectifs du chapitre
- Gérer les erreurs de manière globale et cohérente
- Personnaliser les messages d'erreur
- Implémenter un middleware de gestion d'erreurs
- Logger les erreurs correctement

---

### [GUIDE] 8.1 — `HTTPException` — L'exception de base

```python
from fastapi import HTTPException, status

# Utilisation simple
raise HTTPException(
    status_code=status.HTTP_404_NOT_FOUND,
    detail="Ressource non trouvée"
)

# Avec des headers personnalisés
raise HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Token invalide ou expiré",
    headers={"WWW-Authenticate": "Bearer"},
)
```

---

### [GUIDE] 8.2 — Exceptions personnalisées

```python
# app/core/exceptions.py

class APIException(Exception):
    """Exception de base pour l'API"""
    def __init__(self, message: str, code: int = 400):
        self.message = message
        self.code = code
        super().__init__(self.message)

class RessourceNonTrouvee(APIException):
    def __init__(self, ressource: str, id: int):
        super().__init__(f"{ressource} avec l'id {id} n'existe pas", code=404)

class EmailDejaUtilise(APIException):
    def __init__(self, email: str):
        super().__init__(f"L'email {email} est déjà enregistré", code=409)

class PermissionRefusee(APIException):
    def __init__(self, action: str):
        super().__init__(f"Vous n'êtes pas autorisé à {action}", code=403)
```

---

### [GUIDE] 8.3 — Gestionnaires d'exceptions globaux

```python
# main.py

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.core.exceptions import APIException
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

# Gestionnaire pour nos exceptions personnalisées
@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
    logger.warning(f"APIException: {exc.message} - URL: {request.url}")
    return JSONResponse(
        status_code=exc.code,
        content={
            "erreur": True,
            "message": exc.message,
            "chemin": str(request.url.path)
        }
    )

# Gestionnaire pour les erreurs de validation Pydantic
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    erreurs = []
    for erreur in exc.errors():
        erreurs.append({
            "champ": " -> ".join(str(x) for x in erreur["loc"]),
            "message": erreur["msg"],
            "type": erreur["type"]
        })
    
    logger.warning(f"Erreur de validation: {erreurs}")
    
    return JSONResponse(
        status_code=422,
        content={
            "erreur": True,
            "message": "Données invalides",
            "details": erreurs
        }
    )

# Gestionnaire pour toutes les autres exceptions
@app.exception_handler(Exception)
async def exception_inconnue_handler(request: Request, exc: Exception):
    logger.error(f"Erreur inattendue: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={
            "erreur": True,
            "message": "Une erreur interne s'est produite. Veuillez réessayer."
            # Ne jamais exposer les détails de l'erreur en production !
        }
    )
```

---

### [GUIDE] 8.4 — Configuration du logging

```python
# app/core/logging.py

import logging
import sys

def configurer_logging(niveau: str = "INFO"):
    logging.basicConfig(
        level=getattr(logging, niveau.upper()),
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        handlers=[
            logging.StreamHandler(sys.stdout),  # Console
        ]
    )
    
    # Réduire le bruit des bibliothèques tierces
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
```

---

### [LOGIQUE] Clés à retenir

- Ne jamais exposer les **détails techniques** des erreurs internes en production
- Créer des exceptions **personnalisées** pour une meilleure lisibilité du code
- Les gestionnaires globaux centralisent la gestion des erreurs
- Logger toutes les erreurs pour faciliter le débogage

### [OK] Bonne pratique

> En production : messages d'erreur vagues côté client, erreurs détaillées côté logs. En développement : l'inverse. Le flag `debug` dans la config vous permet de basculer entre les deux modes.

---

## Chapitre 9 : Middleware & Hooks

### [OBJECTIF] Objectifs du chapitre
- Comprendre le rôle des middlewares dans FastAPI
- Configurer CORS pour les applications frontend
- Créer des middlewares personnalisés (logging, timing)

---

### [GUIDE] 9.1 — Qu'est-ce qu'un middleware ?

Un **middleware** est un composant qui s'exécute **avant ET après** chaque requête HTTP. C'est comme un gardien qui intercepte toutes les requêtes.

```
Requête -> [Middleware 1] -> [Middleware 2] -> Route FastAPI
                                               v
Réponse <- [Middleware 1] <- [Middleware 2] <- Réponse
```

---

### [GUIDE] 9.2 — Middleware CORS

**CORS** (Cross-Origin Resource Sharing) est un mécanisme de sécurité des navigateurs qui bloque les requêtes venant d'un domaine différent. Si votre frontend React tourne sur `http://localhost:3000` et votre API sur `http://localhost:8000`, CORS doit être configuré.

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",      # Frontend React en développement
        "https://monsite.com",        # Frontend en production
    ],
    allow_credentials=True,           # Autoriser les cookies/tokens
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)
```

---

### [GUIDE] 9.3 — Middleware de timing

```python
import time
import logging
from fastapi import FastAPI, Request

app = FastAPI()
logger = logging.getLogger(__name__)

@app.middleware("http")
async def middleware_timing(request: Request, call_next):
    # Code exécuté AVANT la route
    debut = time.time()
    
    # Appel de la route (ou middleware suivant)
    reponse = await call_next(request)
    
    # Code exécuté APRÈS la route
    duree = (time.time() - debut) * 1000  # En millisecondes
    
    # Ajouter un header de timing à la réponse
    reponse.headers["X-Temps-Traitement"] = f"{duree:.2f}ms"
    
    logger.info(
        f"{request.method} {request.url.path} -> {reponse.status_code} ({duree:.2f}ms)"
    )
    
    return reponse
```

---

### [GUIDE] 9.4 — Middleware de logging

```python
@app.middleware("http")
async def middleware_logging(request: Request, call_next):
    # Informations sur la requête entrante
    logger.info(
        f"Requête: {request.method} {request.url} "
        f"| IP: {request.client.host} "
        f"| User-Agent: {request.headers.get('user-agent', 'inconnu')}"
    )
    
    reponse = await call_next(request)
    
    # Informations sur la réponse
    logger.info(f"Réponse: {reponse.status_code}")
    
    return reponse
```

---

### [GUIDE] 9.5 — Événements de démarrage et d'arrêt

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def duree_de_vie(app: FastAPI):
    # Code exécuté au DÉMARRAGE
    print("[RAPIDE] Démarrage de l'application...")
    await init_db()
    await connecter_redis()
    
    yield  # L'application fonctionne ici
    
    # Code exécuté à l'ARRÊT
    print("[STOP] Arrêt de l'application...")
    await deconnecter_redis()

app = FastAPI(lifespan=duree_de_vie)
```

---

### [LOGIQUE] Clés à retenir

- Un middleware intercepte **toutes** les requêtes et réponses
- CORS est indispensable si votre frontend et votre API sont sur des domaines/ports différents
- L'ordre des middlewares est important : ils s'appliquent dans l'ordre inverse de leur déclaration

### [OK] Bonne pratique

> N'ajoutez que les middlewares dont vous avez **réellement besoin**. Chaque middleware ajoute une latence. Un middleware de compression, de logging, de CORS et d'authentification est généralement suffisant.

---

*Fin de la Partie 2 — Architecture & Structuration*

> **Prochaine étape :** Partie 3 — Sécurité & Authentification


# [RAPIDE] FastAPI — Fil Rouge Étudiant
# [SCIENCE] PARTIE 7 — Niveau Expert : Architecture & Scalabilité

> **Prérequis :** Avoir complété les Parties 1 à 6
> **Niveau :** Expert / Professionnel
> **Objectif :** Concevoir des systèmes FastAPI robustes, scalables et maintenables

---

## Chapitre 23 : Architecture Avancée

### [OBJECTIF] Objectifs du chapitre
- Comprendre l'architecture hexagonale
- Implémenter le Repository Pattern
- Découpler complètement l'infrastructure du domaine

---

### [GUIDE] 23.1 — Le problème des architectures plates

Dans une architecture classique, vos routes FastAPI appellent directement la base de données via SQLAlchemy. C'est fonctionnel mais pose des problèmes :

- **Tests difficiles** : vous devez toujours avoir une vraie DB
- **Couplage fort** : changer de base de données nécessite de tout réécrire
- **Logique métier éparpillée** : dans les routes, les modèles, partout...

---

### [GUIDE] 23.2 — Architecture Hexagonale (Ports & Adapters)

L'architecture hexagonale sépare le code en trois zones :

```
┌─────────────────────────────────────────────────────────┐
│                    INFRASTRUCTURE                        │
│  (FastAPI routes, SQLAlchemy, Redis, SMTP, Stripe...)   │
├─────────────────────────────────────────────────────────┤
│                  APPLICATION (Use Cases)                 │
│    (Logique d'application : créer commande, etc.)       │
├─────────────────────────────────────────────────────────┤
│                      DOMAINE                            │
│    (Entités, règles métier, interfaces = ports)         │
└─────────────────────────────────────────────────────────┘
```

Le **domaine** ne connaît rien de FastAPI, SQLAlchemy, ou Redis. Il définit des **interfaces (ports)** que l'infrastructure implémente (adapters).

---

### [GUIDE] 23.3 — Implémentation du Repository Pattern

```python
# app/domain/repositories/utilisateur.py — INTERFACE (Port)

from abc import ABC, abstractmethod
from typing import Optional, List

class UtilisateurRepositoryInterface(ABC):
    """Interface que TOUS les repositories utilisateur doivent implémenter."""
    
    @abstractmethod
    async def trouver_par_id(self, user_id: int) -> Optional["Utilisateur"]:
        pass
    
    @abstractmethod
    async def trouver_par_email(self, email: str) -> Optional["Utilisateur"]:
        pass
    
    @abstractmethod
    async def sauvegarder(self, utilisateur: "Utilisateur") -> "Utilisateur":
        pass
    
    @abstractmethod
    async def supprimer(self, user_id: int) -> bool:
        pass
    
    @abstractmethod
    async def lister(self, skip: int, limit: int) -> List["Utilisateur"]:
        pass
```

```python
# app/infrastructure/repositories/utilisateur_sqlalchemy.py — IMPLEMENTATION (Adapter)

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.domain.repositories.utilisateur import UtilisateurRepositoryInterface
from app.infrastructure.models.utilisateur import UtilisateurORM
from app.domain.entities.utilisateur import Utilisateur

class UtilisateurRepositorySQLAlchemy(UtilisateurRepositoryInterface):
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def trouver_par_id(self, user_id: int) -> Optional[Utilisateur]:
        orm = await self.db.get(UtilisateurORM, user_id)
        return self._orm_vers_entite(orm) if orm else None
    
    async def trouver_par_email(self, email: str) -> Optional[Utilisateur]:
        resultat = await self.db.execute(
            select(UtilisateurORM).where(UtilisateurORM.email == email)
        )
        orm = resultat.scalar_one_or_none()
        return self._orm_vers_entite(orm) if orm else None
    
    async def sauvegarder(self, utilisateur: Utilisateur) -> Utilisateur:
        if utilisateur.id:
            orm = await self.db.get(UtilisateurORM, utilisateur.id)
            # Mettre à jour
        else:
            orm = self._entite_vers_orm(utilisateur)
            self.db.add(orm)
        
        await self.db.commit()
        await self.db.refresh(orm)
        return self._orm_vers_entite(orm)
    
    def _orm_vers_entite(self, orm: UtilisateurORM) -> Utilisateur:
        return Utilisateur(id=orm.id, nom=orm.nom, email=orm.email)
    
    def _entite_vers_orm(self, entite: Utilisateur) -> UtilisateurORM:
        return UtilisateurORM(nom=entite.nom, email=entite.email)
    
    # ... autres méthodes
```

```python
# app/application/services/utilisateur_service.py — COUCHE APPLICATION

from app.domain.repositories.utilisateur import UtilisateurRepositoryInterface
from app.domain.entities.utilisateur import Utilisateur
from app.core.security import hasher_mot_de_passe

class UtilisateurService:
    """Logique d'application pour la gestion des utilisateurs."""
    
    def __init__(self, repo: UtilisateurRepositoryInterface):
        # On reçoit l'INTERFACE, pas l'implémentation
        # On peut donc tester avec un Mock ou une implémentation en mémoire !
        self.repo = repo
    
    async def inscrire_utilisateur(self, nom: str, email: str, mot_de_passe: str) -> Utilisateur:
        # Règle métier : email unique
        existant = await self.repo.trouver_par_email(email)
        if existant:
            raise ValueError(f"L'email {email} est déjà utilisé")
        
        # Règle métier : mot de passe minimal
        if len(mot_de_passe) < 8:
            raise ValueError("Le mot de passe doit faire au moins 8 caractères")
        
        utilisateur = Utilisateur(
            nom=nom,
            email=email,
            mot_de_passe_hash=hasher_mot_de_passe(mot_de_passe)
        )
        
        return await self.repo.sauvegarder(utilisateur)
```

```python
# app/api/routes/utilisateurs.py — Injection dans les routes

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db
from app.infrastructure.repositories.utilisateur_sqlalchemy import UtilisateurRepositorySQLAlchemy
from app.application.services.utilisateur_service import UtilisateurService

router = APIRouter()

def get_utilisateur_service(db: AsyncSession = Depends(get_db)) -> UtilisateurService:
    repo = UtilisateurRepositorySQLAlchemy(db)
    return UtilisateurService(repo)

@router.post("/utilisateurs/")
async def creer_utilisateur(
    data: UtilisateurCreer,
    service: UtilisateurService = Depends(get_utilisateur_service)
):
    utilisateur = await service.inscrire_utilisateur(data.nom, data.email, data.mot_de_passe)
    return utilisateur
```

**Avantage pour les tests :**
```python
# test — On peut tester le service sans base de données !

class UtilisateurRepositoryEnMemoire(UtilisateurRepositoryInterface):
    """Implémentation en mémoire pour les tests."""
    def __init__(self):
        self.utilisateurs = {}
    
    async def trouver_par_email(self, email: str):
        return next((u for u in self.utilisateurs.values() if u.email == email), None)
    
    async def sauvegarder(self, u):
        u.id = len(self.utilisateurs) + 1
        self.utilisateurs[u.id] = u
        return u

async def test_inscription_email_duplique():
    repo = UtilisateurRepositoryEnMemoire()
    service = UtilisateurService(repo)
    
    await service.inscrire_utilisateur("Alice", "alice@test.com", "Pass123!")
    
    with pytest.raises(ValueError, match="déjà utilisé"):
        await service.inscrire_utilisateur("Bob", "alice@test.com", "Pass456!")
```

---

## Chapitre 24 : Microservices avec FastAPI

### [OBJECTIF] Objectifs du chapitre
- Comprendre l'architecture microservices
- Faire communiquer des services entre eux
- Gérer une API Gateway

---

### [GUIDE] 24.1 — Monolithe vs Microservices

| | Monolithe | Microservices |
|--|--|--|
| **Déploiement** | Simple | Complexe |
| **Scalabilité** | Tout ou rien | Service par service |
| **Tests** | Simples | Complexes |
| **Pannes** | Tout s'arrête | Service isolé |
| **Idéal pour** | < 10 devs | > 10 devs, grande scale |

**Ne migrez pas vers les microservices prématurément.** Un monolithe bien structuré est souvent la meilleure solution.

---

### [GUIDE] 24.2 — Communication entre services (REST)

```python
# service_commandes/app/clients/service_inventaire.py
# Le service Commandes communique avec le service Inventaire

import httpx
from app.core.config import parametres

class ClientServiceInventaire:
    """Client HTTP pour appeler le service Inventaire."""
    
    BASE_URL = parametres.url_service_inventaire  # "http://service-inventaire:8001"
    
    async def verifier_stock(self, produit_id: int, quantite: int) -> bool:
        async with httpx.AsyncClient(timeout=5.0) as client:
            try:
                reponse = await client.get(
                    f"{self.BASE_URL}/produits/{produit_id}/stock",
                    headers={"X-API-Key": parametres.cle_inter_services}
                )
                reponse.raise_for_status()
                stock = reponse.json()
                return stock["disponible"] >= quantite
            except httpx.TimeoutException:
                raise ServiceIndisponible("Service inventaire timeout")
            except httpx.HTTPStatusError as e:
                raise ServiceIndisponible(f"Erreur inventaire: {e.response.status_code}")
    
    async def reserver_stock(self, produit_id: int, quantite: int, commande_id: str):
        async with httpx.AsyncClient(timeout=10.0) as client:
            reponse = await client.post(
                f"{self.BASE_URL}/reservations",
                json={
                    "produit_id": produit_id,
                    "quantite": quantite,
                    "commande_id": commande_id
                }
            )
            reponse.raise_for_status()
            return reponse.json()
```

---

### [GUIDE] 24.3 — Communication asynchrone avec RabbitMQ/Redis Pub-Sub

```bash
pip install aio-pika  # Pour RabbitMQ
```

```python
# Producteur (service Commandes — publie un événement)
import aio_pika
import json

async def publier_commande_cree(commande_id: str, donnees: dict):
    """Publie un événement quand une commande est créée."""
    connection = await aio_pika.connect_robust("amqp://user:pass@rabbitmq/")
    
    async with connection:
        channel = await connection.channel()
        exchange = await channel.declare_exchange("commandes", aio_pika.ExchangeType.TOPIC)
        
        message = aio_pika.Message(
            body=json.dumps({"id": commande_id, "data": donnees}).encode(),
            content_type="application/json"
        )
        
        await exchange.publish(message, routing_key="commande.cree")

# Consommateur (service Emails — écoute les événements)
async def consommer_evenements():
    connection = await aio_pika.connect_robust("amqp://user:pass@rabbitmq/")
    
    async with connection:
        channel = await connection.channel()
        exchange = await channel.declare_exchange("commandes", aio_pika.ExchangeType.TOPIC)
        queue = await channel.declare_queue("emails.commandes", durable=True)
        
        await queue.bind(exchange, routing_key="commande.cree")
        
        async with queue.iterator() as messages:
            async for message in messages:
                async with message.process():
                    donnees = json.loads(message.body)
                    await envoyer_confirmation_commande(donnees)
```

---

### [GUIDE] 24.4 — Trace ID pour la traçabilité

```python
# middleware qui propage le trace ID à travers tous les services

import uuid
from fastapi import Request

@app.middleware("http")
async def middleware_trace_id(request: Request, call_next):
    # Récupérer ou générer un trace ID
    trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4()))
    
    # Rendre disponible dans tout le contexte de la requête
    request.state.trace_id = trace_id
    
    reponse = await call_next(request)
    
    # Retourner le trace ID dans la réponse
    reponse.headers["X-Trace-ID"] = trace_id
    
    return reponse
```

---

## Chapitre 25 : Observabilité & Performance

### [OBJECTIF] Objectifs du chapitre
- Mettre en place le tracing distribué avec OpenTelemetry
- Collecter des métriques avec Prometheus
- Faire des tests de charge avec Locust

---

### [GUIDE] 25.1 — OpenTelemetry (Tracing)

```bash
pip install opentelemetry-sdk opentelemetry-instrumentation-fastapi opentelemetry-exporter-jaeger
```

```python
# app/core/telemetry.py

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

def configurer_telemetrie(app):
    # Exporter vers Jaeger (visualisation des traces)
    jaeger_exporter = JaegerExporter(
        agent_host_name="localhost",
        agent_port=6831,
    )
    
    provider = TracerProvider()
    provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
    trace.set_tracer_provider(provider)
    
    # Instrumenter automatiquement FastAPI
    FastAPIInstrumentor.instrument_app(app)
```

---

### [GUIDE] 25.2 — Prometheus Metrics

```bash
pip install prometheus-fastapi-instrumentator
```

```python
# main.py

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

# Active la collecte de métriques automatique
Instrumentator().instrument(app).expose(app, endpoint="/metrics")

# Les métriques sont maintenant disponibles sur GET /metrics
# Configurez Prometheus pour scraper cette URL toutes les 15 secondes
```

```yaml
# prometheus.yml
scrape_configs:
  - job_name: fastapi
    static_configs:
      - targets: ["localhost:8000"]
    metrics_path: /metrics
    scrape_interval: 15s
```

---

### [GUIDE] 25.3 — Tests de charge avec Locust

```bash
pip install locust
```

```python
# locustfile.py

from locust import HttpUser, task, between

class UtilisateurSimule(HttpUser):
    wait_time = between(1, 3)  # Attente entre 1 et 3 secondes
    token = None
    
    def on_start(self):
        """Connexion au démarrage de chaque utilisateur simulé."""
        reponse = self.client.post("/login", data={
            "username": "test@test.com",
            "password": "Pass123!"
        })
        self.token = reponse.json()["access_token"]
    
    @task(3)  # 3x plus fréquent que les autres
    def lire_articles(self):
        self.client.get("/articles?page=1&taille=10")
    
    @task(1)
    def lire_profil(self):
        self.client.get(
            "/mon-profil",
            headers={"Authorization": f"Bearer {self.token}"}
        )
    
    @task(1)
    def creer_article(self):
        self.client.post(
            "/articles",
            json={"titre": "Test", "contenu": "Contenu de test"},
            headers={"Authorization": f"Bearer {self.token}"}
        )
```

```bash
# Lancer le test de charge
locust -f locustfile.py --host=http://localhost:8000 --users=100 --spawn-rate=10
# Ouvrir http://localhost:8089 pour l'interface graphique
```

---

## Chapitre 26 : Bonnes Pratiques Finales et Revue d'Expert

### [OBJECTIF] Objectifs du chapitre
- Consolider toutes les bonnes pratiques apprises
- Structurer un projet professionnel complet
- Préparer votre API pour la production

---

### [GUIDE] 26.1 — Checklist d'un projet FastAPI professionnel

#### [OK] Structure & Code

```
[WHITE_SQUARE] Architecture modulaire (routes, services, repositories, domaine)
[WHITE_SQUARE] 1 fichier = 1 responsabilité (Single Responsibility Principle)
[WHITE_SQUARE] Types Python partout (mypy pour vérification statique)
[WHITE_SQUARE] Pas de code "mort" ou commenté
[WHITE_SQUARE] Nommage clair et cohérent (français ou anglais, pas les deux !)
[WHITE_SQUARE] Docstrings sur toutes les fonctions publiques
[WHITE_SQUARE] Constants dans un fichier dédié, jamais en dur dans le code
```

#### [OK] Sécurité

```
[WHITE_SQUARE] Toutes les entrées validées avec Pydantic
[WHITE_SQUARE] Mots de passe hashés avec bcrypt
[WHITE_SQUARE] JWT avec expiration courte + refresh tokens
[WHITE_SQUARE] CORS configuré strictement
[WHITE_SQUARE] Rate limiting sur les routes sensibles
[WHITE_SQUARE] Secrets dans .env, jamais dans le code
[WHITE_SQUARE] Headers de sécurité (X-Frame-Options, HSTS...)
[WHITE_SQUARE] Logs sans données sensibles (pas de mots de passe, tokens)
[WHITE_SQUARE] SQL via ORM (protection injection SQL)
```

#### [OK] Performance

```
[WHITE_SQUARE] Toutes les routes I/O sont async
[WHITE_SQUARE] Cache Redis pour les données fréquentes et lentes
[WHITE_SQUARE] Pagination sur toutes les listes
[WHITE_SQUARE] Index sur les colonnes fréquemment filtrées
[WHITE_SQUARE] Compression GZip activée
[WHITE_SQUARE] N+1 queries évitées (utiliser joinedload avec SQLAlchemy)
```

#### [OK] Tests

```
[WHITE_SQUARE] Couverture de tests > 80% sur les routes critiques
[WHITE_SQUARE] Tests unitaires pour la logique métier
[WHITE_SQUARE] Tests d'intégration pour les routes API
[WHITE_SQUARE] Tests avec DB en mémoire (pas la vraie DB)
[WHITE_SQUARE] Fixtures pour les données de test
[WHITE_SQUARE] Tests des cas d'erreur (pas que le happy path)
```

#### [OK] CI/CD & Déploiement

```
[WHITE_SQUARE] Pipeline CI : lint -> tests -> build -> deploy
[WHITE_SQUARE] Docker multi-stage (image légère en prod)
[WHITE_SQUARE] Variables d'environnement pour tous les paramètres
[WHITE_SQUARE] Health check endpoint (/health)
[WHITE_SQUARE] Logs JSON structurés
[WHITE_SQUARE] Monitoring (Sentry, Prometheus)
[WHITE_SQUARE] Alertes configurées
[WHITE_SQUARE] Procédure de rollback documentée
```

---

### [GUIDE] 26.2 — Structure finale d'un projet professionnel

```
mon_api_pro/
├── .github/
│   └── workflows/
│       ├── ci.yml
│       └── cd.yml
│
├── docker/
│   ├── Dockerfile
│   ├── docker-compose.yml
│   └── docker-compose.prod.yml
│
├── nginx/
│   └── nginx.conf
│
├── app/
│   ├── __init__.py
│   │
│   ├── domain/                    # Aucune dépendance externe
│   │   ├── entities/              # Objets métier purs
│   │   ├── repositories/          # Interfaces (ports)
│   │   └── exceptions.py          # Exceptions métier
│   │
│   ├── application/               # Orchestration
│   │   └── services/              # Use cases / services
│   │
│   ├── infrastructure/            # Implémentations concrètes
│   │   ├── db/
│   │   │   ├── session.py
│   │   │   ├── models/            # Modèles ORM
│   │   │   └── repositories/      # Implémentations SQLAlchemy
│   │   ├── cache/                 # Redis
│   │   └── external/              # Clients d'APIs tierces
│   │
│   ├── api/                       # Couche HTTP
│   │   ├── routes/
│   │   ├── schemas/               # Pydantic schemas
│   │   ├── dependencies.py
│   │   └── middleware.py
│   │
│   └── core/
│       ├── config.py
│       ├── security.py
│       ├── logging.py
│       └── exceptions.py
│
├── tests/
│   ├── unit/                      # Tests sans DB
│   ├── integration/               # Tests avec DB de test
│   └── conftest.py
│
├── scripts/
│   ├── migration.py
│   └── seed_data.py
│
├── docs/
│   ├── architecture.md
│   └── api-guide.md
│
├── main.py
├── requirements.txt
├── requirements-dev.txt
├── .env.example                   # Template .env (committez-le !)
├── .env                           # Variables réelles (NE JAMAIS committer)
├── .gitignore
├── pyproject.toml                 # Config mypy, black, ruff
└── README.md
```

---

### [GUIDE] 26.3 — Endpoint de health check

```python
# app/api/routes/health.py

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
import redis.asyncio as aioredis
from app.db.session import get_db

router = APIRouter(tags=["[HOPITAL] Health"])

@router.get("/health")
async def health_check(db: AsyncSession = Depends(get_db)):
    """
    Vérifie la santé de l'application et ses dépendances.
    Utilisé par les load balancers et le monitoring.
    """
    statuts = {
        "api": "ok",
        "database": "unknown",
        "cache": "unknown"
    }
    
    # Vérifier la base de données
    try:
        await db.execute(text("SELECT 1"))
        statuts["database"] = "ok"
    except Exception:
        statuts["database"] = "error"
    
    # Vérifier Redis
    try:
        redis = aioredis.from_url("redis://localhost")
        await redis.ping()
        statuts["cache"] = "ok"
    except Exception:
        statuts["cache"] = "error"
    
    # Code HTTP 200 si tout ok, 503 si un service est défaillant
    code = 200 if all(v == "ok" for v in statuts.values()) else 503
    
    from fastapi.responses import JSONResponse
    return JSONResponse(content=statuts, status_code=code)
```

---

### [GUIDE] 26.4 — Conventions de code

```python
# pyproject.toml — Configuration des outils de qualité de code

[tool.black]
line-length = 88
target-version = ["py312"]

[tool.ruff]
line-length = 88
select = ["E", "F", "I", "N", "UP"]  # Règles activées

[tool.mypy]
python_version = "3.12"
strict = true
ignore_missing_imports = true

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
```

```bash
# Commandes qualité de code
black .               # Formater le code
ruff check . --fix    # Linter + corrections auto
mypy app/             # Vérification des types
pytest --cov=app      # Tests avec couverture
```

---

### [GUIDE] 26.5 — Les 10 commandements du développeur FastAPI

1. **Tu typeras toujours** — Pas de variables sans type, jamais de `Any`
2. **Tu séparerastes responsabilités** — Une fonction = une chose
3. **Tu ne stockeras jamais de secret dans le code** — Toujours dans `.env`
4. **Tu hasheras les mots de passe** — Jamais de clair, jamais de MD5/SHA1
5. **Tu testeras les cas d'erreur** — Le happy path ne suffit pas
6. **Tu loggeras, mais sans données sensibles** — Logs utiles, pas dangereux
7. **Tu pagineras tes listes** — Jamais de `SELECT *` sans LIMIT
8. **Tu utiliseras async pour l'I/O** — Jamais de blocking call en async
9. **Tu documenteras tes routes** — Pour toi dans 6 mois, et pour tes collègues
10. **Tu profileras avant d'optimiser** — Les prématurées optimisations sont la source de tous les maux

---

### [COURS] Récapitulatif du parcours complet

```
PARTIE 1 — Fondamentaux
├── Chapitre 1  : Hello World, Uvicorn, structure minimale
├── Chapitre 2  : Routes, méthodes HTTP, path/query params
├── Chapitre 3  : Pydantic, validation, modèles de données
└── Chapitre 4  : Documentation auto (Swagger, ReDoc)

PARTIE 2 — Architecture
├── Chapitre 5  : APIRouter, structure modulaire, config
├── Chapitre 6  : SQLAlchemy, ORM, sessions async
├── Chapitre 7  : CRUD complet, injection de dépendances
├── Chapitre 8  : Exceptions custom, gestionnaires globaux
└── Chapitre 9  : Middlewares (CORS, logging, timing)

PARTIE 3 — Sécurité
├── Chapitre 10 : JWT, bcrypt, endpoint login
├── Chapitre 11 : Refresh tokens, rôles, permissions
└── Chapitre 12 : CORS, rate limiting, headers sécurité

PARTIE 4 — Performance
├── Chapitre 13 : async/await, BackgroundTasks
└── Chapitre 14 : Cache Redis, pagination, compression

PARTIE 5 — Communication
├── Chapitre 15 : Upload/download fichiers
├── Chapitre 16 : WebSockets, chat temps réel
├── Chapitre 17 : Celery, tâches distribuées
└── Chapitre 18 : Emails HTML, notifications

PARTIE 6 — Tests & Déploiement
├── Chapitre 19 : pytest, tests d'intégration, couverture
├── Chapitre 20 : Logging JSON, Sentry, Prometheus
├── Chapitre 21 : Docker, Gunicorn, Nginx
└── Chapitre 22 : GitHub Actions, CI/CD complet

PARTIE 7 — Expert
├── Chapitre 23 : Architecture hexagonale, Repository Pattern
├── Chapitre 24 : Microservices, RabbitMQ, Trace ID
├── Chapitre 25 : OpenTelemetry, Prometheus, Locust
└── Chapitre 26 : Bonnes pratiques, checklist prod, conventions
```

---

### [RAPIDE] Et maintenant ?

Vous avez parcouru l'intégralité du programme. Pour aller encore plus loin :

- **Projet personnel** : Construisez une vraie API de A à Z (idée perso, portfolio)
- **Open source** : Contribuez à des projets FastAPI sur GitHub
- **Veille technologique** : Suivez le blog de Tiangolo (créateur de FastAPI)
- **Spécialisations** :
  - **Data APIs** : combiner FastAPI avec Pandas, Polars, ou DuckDB
  - **ML APIs** : servir des modèles ML avec FastAPI
  - **GraphQL** : explorer Strawberry avec FastAPI
  - **gRPC** : communication haute performance entre microservices

---

*Fin du Fil Rouge FastAPI — De Débutant à Expert*

> **"Le code propre fait une seule chose et la fait bien."** — Robert C. Martin
>
> Bonne continuation dans votre aventure FastAPI ! [RAPIDE]


# [RAPIDE] FastAPI — Fil Rouge Étudiant
# [VERROUILLE] PARTIE 3 — Sécurité & Authentification
# [RAPIDE] PARTIE 4 — Performance & Asynchronisme

> **Prérequis :** Avoir complété les Parties 1 et 2

---

## Chapitre 10 : Authentification Basique (JWT)

### [OBJECTIF] Objectifs du chapitre
- Comprendre le fonctionnement des tokens JWT
- Implémenter un endpoint de login
- Hasher les mots de passe avec bcrypt
- Protéger des routes avec OAuth2

---

### [GUIDE] 10.1 — Comment fonctionne l'authentification par token ?

```
1. L'utilisateur envoie son email + mot de passe
2. L'API vérifie les identifiants
3. Si valide, l'API génère un token JWT et le retourne
4. L'utilisateur stocke le token (localStorage ou cookie)
5. Pour chaque requête protégée, l'utilisateur envoie le token dans l'en-tête
6. L'API vérifie et décode le token à chaque requête
```

Un **token JWT** (JSON Web Token) est composé de 3 parties séparées par des points :
- **Header** : algorithme de signature
- **Payload** : données (user_id, expiration...)
- **Signature** : garantit l'authenticité

---

### [GUIDE] 10.2 — Installation des dépendances

```bash
pip install python-jose[cryptography] passlib[bcrypt] python-multipart
```

- `python-jose` : génération et vérification des tokens JWT
- `passlib[bcrypt]` : hashage sécurisé des mots de passe
- `python-multipart` : nécessaire pour les formulaires OAuth2

---

### [GUIDE] 10.3 — Hashage des mots de passe

```python
# app/core/security.py

from passlib.context import CryptContext

# Contexte de hashage — bcrypt est l'algorithme recommandé
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hasher_mot_de_passe(mot_de_passe: str) -> str:
    """Transforme un mot de passe en clair en hash bcrypt."""
    return pwd_context.hash(mot_de_passe)

def verifier_mot_de_passe(mot_de_passe_clair: str, mot_de_passe_hash: str) -> bool:
    """Vérifie qu'un mot de passe correspond à son hash."""
    return pwd_context.verify(mot_de_passe_clair, mot_de_passe_hash)

# Exemple d'utilisation :
# hash = hasher_mot_de_passe("monsupermotdepasse")
# # -> "$2b$12$..." (chaîne de 60 caractères)
# verifier_mot_de_passe("monsupermotdepasse", hash)
# # -> True
# verifier_mot_de_passe("mauvais", hash)
# # -> False
```

---

### [GUIDE] 10.4 — Génération et vérification des tokens JWT

```python
# app/core/security.py (suite)

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from app.core.config import parametres

# Ces valeurs viennent de votre fichier .env
CLE_SECRETE = parametres.cle_secrete        # Doit être longue et aléatoire
ALGORITHME = parametres.algorithme_jwt       # "HS256"
DUREE_TOKEN = parametres.duree_token_minutes # 30 minutes

def creer_token_acces(donnees: dict) -> str:
    """Crée un token JWT avec une durée d'expiration."""
    payload = donnees.copy()
    
    # Ajouter l'heure d'expiration
    expiration = datetime.now(timezone.utc) + timedelta(minutes=DUREE_TOKEN)
    payload.update({"exp": expiration})
    
    # Encoder le token
    token = jwt.encode(payload, CLE_SECRETE, algorithm=ALGORITHME)
    return token

def decoder_token(token: str) -> dict:
    """Décode et vérifie un token JWT. Lève une exception si invalide."""
    try:
        payload = jwt.decode(token, CLE_SECRETE, algorithms=[ALGORITHME])
        return payload
    except JWTError:
        raise ValueError("Token invalide ou expiré")
```

---

### [GUIDE] 10.5 — L'endpoint de login

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

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.db.session import get_db
from app.models.utilisateur import Utilisateur
from app.core.security import verifier_mot_de_passe, creer_token_acces

router = APIRouter(tags=["[SECURISE] Authentification"])

@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    # OAuth2PasswordRequestForm attend les champs "username" et "password"
    db: AsyncSession = Depends(get_db)
):
    # 1. Chercher l'utilisateur par email
    resultat = await db.execute(
        select(Utilisateur).where(Utilisateur.email == form_data.username)
    )
    utilisateur = resultat.scalar_one_or_none()
    
    # 2. Vérifier les identifiants
    # On utilise la MÊME réponse d'erreur pour email et mot de passe incorrects
    # Pour éviter d'indiquer à un attaquant si l'email existe ou non
    erreur_auth = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Email ou mot de passe incorrect",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    if not utilisateur:
        raise erreur_auth
    
    if not verifier_mot_de_passe(form_data.password, utilisateur.mot_de_passe_hash):
        raise erreur_auth
    
    if not utilisateur.actif:
        raise HTTPException(status_code=400, detail="Compte désactivé")
    
    # 3. Générer et retourner le token
    token = creer_token_acces({"sub": str(utilisateur.id)})
    # "sub" (subject) est une convention JWT pour l'identifiant de l'utilisateur
    
    return {
        "access_token": token,
        "token_type": "bearer"
    }
```

---

### [GUIDE] 10.6 — Protéger des routes

```python
# app/core/dependencies.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db
from app.core.security import decoder_token
from app.models.utilisateur import Utilisateur

# FastAPI cherchera le token dans l'en-tête "Authorization: Bearer <token>"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")

async def get_utilisateur_courant(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> Utilisateur:
    """Dépendance qui récupère l'utilisateur authentifié depuis son token."""
    
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Impossible de valider les identifiants",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = decoder_token(token)
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except ValueError:
        raise credentials_exception
    
    utilisateur = await db.get(Utilisateur, int(user_id))
    if utilisateur is None:
        raise credentials_exception
    
    return utilisateur

# ── Utilisation dans une route ──────────────────────────────
from fastapi import APIRouter

router = APIRouter()

@router.get("/mon-profil")
async def lire_mon_profil(
    utilisateur: Utilisateur = Depends(get_utilisateur_courant)
    # Si le token est invalide, FastAPI lève automatiquement une 401
):
    return {"id": utilisateur.id, "nom": utilisateur.nom}
```

---

### [LOGIQUE] Clés à retenir

- Ne jamais stocker de mot de passe en **clair** — toujours hasher avec bcrypt
- Le token JWT est **signé**, pas chiffré — ne pas y mettre d'informations sensibles
- Utiliser la **même message d'erreur** pour email et mot de passe invalides (sécurité)
- Centraliser les dépendances d'auth dans `dependencies.py`

### [OK] Bonne pratique

> La clé secrète JWT doit être générée aléatoirement et être longue (minimum 32 caractères). Générez-la avec : `openssl rand -hex 32` et stockez-la dans `.env`.

---

## Chapitre 11 : Authentification Avancée & Permissions

### [OBJECTIF] Objectifs du chapitre
- Implémenter les refresh tokens
- Créer un système de rôles et permissions
- Protéger des routes par rôle

---

### [GUIDE] 11.1 — Access Token vs Refresh Token

| | Access Token | Refresh Token |
|--|--|--|
| **Durée** | Courte (15-30 min) | Longue (7-30 jours) |
| **Stockage** | Mémoire / Header | Base de données / Cookie httpOnly |
| **Usage** | Chaque requête API | Générer un nouvel access token |
| **Révocation** | Impossible (stateless) | Possible (on le supprime en DB) |

```python
# app/core/security.py

def creer_refresh_token(user_id: int) -> str:
    """Crée un refresh token avec longue durée de vie."""
    payload = {
        "sub": str(user_id),
        "type": "refresh",  # Important : distinguer les types de tokens
        "exp": datetime.now(timezone.utc) + timedelta(days=30)
    }
    return jwt.encode(payload, CLE_SECRETE, algorithm=ALGORITHME)
```

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

@router.post("/refresh")
async def rafraichir_token(
    refresh_token: str,
    db: AsyncSession = Depends(get_db)
):
    """Génère un nouvel access token à partir d'un refresh token."""
    try:
        payload = decoder_token(refresh_token)
        
        # Vérifier que c'est bien un refresh token
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Type de token invalide")
        
        user_id = int(payload["sub"])
        utilisateur = await db.get(Utilisateur, user_id)
        
        if not utilisateur or not utilisateur.actif:
            raise HTTPException(status_code=401, detail="Utilisateur invalide")
        
        # Générer un nouvel access token
        nouveau_token = creer_token_acces({"sub": str(user_id)})
        return {"access_token": nouveau_token, "token_type": "bearer"}
        
    except (ValueError, KeyError):
        raise HTTPException(status_code=401, detail="Refresh token invalide")
```

---

### [GUIDE] 11.2 — Système de rôles et permissions

```python
# app/models/utilisateur.py (ajout du rôle)

from enum import Enum as PyEnum

class Role(str, PyEnum):
    ADMIN = "admin"
    MODERATEUR = "moderateur"
    UTILISATEUR = "utilisateur"

class Utilisateur(Base):
    __tablename__ = "utilisateurs"
    # ... autres colonnes ...
    role = Column(String, default=Role.UTILISATEUR)
```

```python
# app/core/dependencies.py (ajout de la vérification des rôles)

from app.models.utilisateur import Role, Utilisateur
from functools import wraps

def requis_role(*roles: Role):
    """
    Factory de dépendances pour vérifier les rôles.
    
    Usage : role_requis = Depends(requis_role(Role.ADMIN))
    """
    async def verifier_role(
        utilisateur: Utilisateur = Depends(get_utilisateur_courant)
    ) -> Utilisateur:
        if utilisateur.role not in roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Cette action nécessite l'un des rôles : {[r.value for r in roles]}"
            )
        return utilisateur
    
    return verifier_role

# ── Utilisation ──────────────────────────────────────────────
admin_requis = Depends(requis_role(Role.ADMIN))
modero_ou_admin = Depends(requis_role(Role.ADMIN, Role.MODERATEUR))

@router.delete("/utilisateurs/{user_id}")
async def supprimer_utilisateur(
    user_id: int,
    admin: Utilisateur = admin_requis,  # Seul un admin peut supprimer
    db: AsyncSession = Depends(get_db)
):
    # ...
    pass

@router.patch("/articles/{article_id}/moderer")
async def moderer_article(
    article_id: int,
    moderateur: Utilisateur = modero_ou_admin,
):
    # ...
    pass
```

---

### [LOGIQUE] Clés à retenir

- Les **refresh tokens** permettent une expérience utilisateur sans re-connexion fréquente
- Les **rôles** contrôlent ce qu'un utilisateur **peut faire** (autorisation)
- L'**authentification** répond à "qui es-tu ?"
- L'**autorisation** répond à "as-tu le droit de faire ça ?"

### [OK] Bonne pratique

> Centralisez toutes les dépendances d'authentification et d'autorisation dans un fichier `dependencies.py`. Ainsi, si vous changez de système d'auth, vous n'avez qu'un seul fichier à modifier.

---

## Chapitre 12 : Sécurité Applicative

### [OBJECTIF] Objectifs du chapitre
- Configurer CORS correctement
- Se protéger contre les vulnérabilités courantes
- Valider strictement toutes les entrées

---

### [GUIDE] 12.1 — CORS en détail

CORS (Cross-Origin Resource Sharing) est géré par le navigateur. En production, soyez **restrictif** :

```python
# app/core/config.py

class Parametres(BaseSettings):
    # En développement : accepter localhost
    origines_autorisees: list[str] = [
        "http://localhost:3000",
        "http://localhost:8080",
    ]
    # En production, remplacer par vos vrais domaines :
    # origines_autorisees: list[str] = ["https://monapp.com"]
```

```python
# main.py

from app.core.config import parametres

app.add_middleware(
    CORSMiddleware,
    allow_origins=parametres.origines_autorisees,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type", "X-Requested-With"],
    max_age=3600,  # Cache CORS 1 heure
)
```

---

### [GUIDE] 12.2 — Protection contre les injections SQL

SQLAlchemy avec l'ORM vous protège automatiquement des injections SQL car il **paramétrise** toutes les requêtes. Évitez cependant de construire des requêtes SQL manuellement :

```python
# [X] DANGEREUX — injection SQL possible
requete = f"SELECT * FROM utilisateurs WHERE nom = '{nom_entree}'"

# [OK] SÉCURISÉ — SQLAlchemy paramétrise automatiquement
resultat = await db.execute(
    select(Utilisateur).where(Utilisateur.nom == nom_entree)
)
```

---

### [GUIDE] 12.3 — Rate Limiting

Le **rate limiting** empêche les attaques par force brute et le spam d'API.

```bash
pip install slowapi
```

```python
# main.py

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# ── Dans les routes ──────────────────────────────────────────
@app.post("/login")
@limiter.limit("5/minute")  # Maximum 5 tentatives par minute par IP
async def login(request: Request, ...):
    # ...
    pass
```

---

### [GUIDE] 12.4 — Headers de sécurité

```python
# middleware de headers de sécurité

@app.middleware("http")
async def ajouter_headers_securite(request: Request, call_next):
    reponse = await call_next(request)
    reponse.headers["X-Content-Type-Options"] = "nosniff"
    reponse.headers["X-Frame-Options"] = "DENY"
    reponse.headers["X-XSS-Protection"] = "1; mode=block"
    reponse.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    return reponse
```

---

### [OK] Bonne pratique

> Isolez toute la configuration dans `core/config.py`. Ne mettez jamais de valeurs sensibles (clés API, mots de passe, tokens) directement dans le code. Utilisez **toujours** des variables d'environnement.

---

# [RAPIDE] PARTIE 4 — Performance & Asynchronisme

---

## Chapitre 13 : Programmation Asynchrone

### [OBJECTIF] Objectifs du chapitre
- Comprendre la différence entre synchrone et asynchrone
- Utiliser `async def` correctement dans FastAPI
- Implémenter des tâches en arrière-plan

---

### [GUIDE] 13.1 — Synchrone vs Asynchrone

**Synchrone** : les opérations s'exécutent les unes après les autres. Chaque opération bloque l'exécution jusqu'à sa fin.

**Asynchrone** : pendant qu'une opération attend (ex: requête DB), Python peut traiter d'autres requêtes.

```python
# [X] SYNCHRONE — bloque le serveur pendant 2 secondes
import time

@app.get("/lent")
def route_lente():
    time.sleep(2)  # Bloque tout le serveur !
    return {"resultat": "fini"}

# [OK] ASYNCHRONE — libère le serveur pendant l'attente
import asyncio

@app.get("/rapide")
async def route_rapide():
    await asyncio.sleep(2)  # Libère le serveur, qui peut traiter d'autres requêtes
    return {"resultat": "fini"}
```

---

### [GUIDE] 13.2 — Règles d'utilisation `async def` vs `def`

```python
# [OK] Utilisez async def quand vous faites des opérations I/O :
# - Requêtes base de données (avec un driver async)
# - Appels HTTP vers d'autres APIs
# - Lecture/écriture de fichiers (avec aiofiles)
# - WebSockets

@app.get("/donnees")
async def obtenir_donnees(db: AsyncSession = Depends(get_db)):
    resultat = await db.execute(select(Article))  # I/O -> async
    return resultat.scalars().all()

# [OK] Utilisez def pour la logique pure (pas d'I/O) :
@app.get("/calcul")
def calculer():
    return {"resultat": sum(range(1_000_000))}  # CPU -> sync
```

---

### [GUIDE] 13.3 — Requêtes HTTP asynchrones vers des APIs externes

```bash
pip install httpx  # Client HTTP async (alternative à requests)
```

```python
import httpx

@app.get("/meteo/{ville}")
async def obtenir_meteo(ville: str):
    async with httpx.AsyncClient() as client:
        reponse = await client.get(
            f"https://api.openweathermap.org/data/2.5/weather",
            params={"q": ville, "appid": "votre_cle_api"}
        )
        reponse.raise_for_status()
        return reponse.json()
```

---

### [GUIDE] 13.4 — BackgroundTasks — Tâches en arrière-plan

Les `BackgroundTasks` permettent d'exécuter du code **après** que la réponse HTTP est envoyée, sans faire attendre le client.

```python
from fastapi import BackgroundTasks

def envoyer_email_bienvenue(email: str, nom: str):
    """Cette fonction s'exécute en arrière-plan."""
    # Simulation d'envoi d'email
    import time
    time.sleep(5)  # L'envoi d'email peut prendre du temps
    print(f"Email de bienvenue envoyé à {email}")

@app.post("/inscription")
async def inscription(
    utilisateur_data: UtilisateurCreer,
    background_tasks: BackgroundTasks,  # Injection automatique
    db: AsyncSession = Depends(get_db)
):
    # 1. Créer l'utilisateur en base (rapide)
    utilisateur = await creer_utilisateur_en_db(db, utilisateur_data)
    
    # 2. Ajouter l'envoi d'email en tâche de fond (ne bloque pas)
    background_tasks.add_task(
        envoyer_email_bienvenue,
        email=utilisateur.email,
        nom=utilisateur.nom
    )
    
    # 3. Retourner immédiatement la réponse
    # L'email sera envoyé APRÈS que cette réponse est reçue par le client
    return {"message": "Compte créé ! Un email de confirmation a été envoyé."}
```

---

### [LOGIQUE] Clés à retenir

- `async def` + `await` pour toutes les opérations I/O (DB, HTTP, fichiers)
- `def` pour les calculs purs sans I/O
- `BackgroundTasks` pour les traitements qui ne doivent pas bloquer la réponse
- FastAPI gère l'event loop automatiquement — ne créez pas le vôtre

### [OK] Bonne pratique

> Toujours utiliser `async` dès qu'il y a une interaction avec une base de données ou un service externe. Même si ça semble plus complexe, ça multiplie la capacité de traitement de votre serveur.

---

## Chapitre 14 : Optimisation & Caching

### [OBJECTIF] Objectifs du chapitre
- Comprendre pourquoi le caching améliore les performances
- Implémenter un cache avec Redis
- Paginer efficacement les résultats

---

### [GUIDE] 14.1 — Pourquoi le caching ?

Sans cache, chaque requête vers `/articles/populaires` execute une requête SQL complexe. Si 1000 utilisateurs font cette requête en même temps, la DB est saturée.

Avec un cache, la première requête calcule le résultat et le stocke en Redis (mémoire). Les 999 autres requêtes lisent depuis Redis (microseconde) sans toucher la DB.

---

### [GUIDE] 14.2 — Configuration Redis

```bash
pip install redis aioredis
# Installer Redis : https://redis.io/download
# Sur Mac : brew install redis && brew services start redis
# Sur Ubuntu : sudo apt install redis-server
```

```python
# app/core/cache.py

import json
import redis.asyncio as aioredis
from functools import wraps
from typing import Callable, Optional

redis_client = aioredis.from_url("redis://localhost", decode_responses=True)

async def obtenir_cache(cle: str) -> Optional[dict]:
    """Récupère une valeur du cache Redis."""
    valeur = await redis_client.get(cle)
    if valeur:
        return json.loads(valeur)
    return None

async def definir_cache(cle: str, valeur: dict, expiration_secondes: int = 300):
    """Stocke une valeur dans Redis avec expiration."""
    await redis_client.setex(
        cle,
        expiration_secondes,
        json.dumps(valeur, default=str)  # default=str pour les dates
    )

async def invalider_cache(cle: str):
    """Supprime une valeur du cache."""
    await redis_client.delete(cle)
```

```python
# Utilisation dans une route

@router.get("/articles/populaires")
async def articles_populaires(db: AsyncSession = Depends(get_db)):
    # Essayer de récupérer depuis le cache
    cle_cache = "articles:populaires"
    donnees_cachees = await obtenir_cache(cle_cache)
    
    if donnees_cachees:
        return {"source": "cache", "data": donnees_cachees}
    
    # Pas de cache -> requête DB
    articles = await calculer_articles_populaires(db)
    
    # Stocker en cache pour 10 minutes
    await definir_cache(cle_cache, articles, expiration_secondes=600)
    
    return {"source": "database", "data": articles}
```

---

### [GUIDE] 14.3 — Pagination efficace

```python
from pydantic import BaseModel
from typing import TypeVar, Generic, List

T = TypeVar("T")

class ReponsePagee(BaseModel, Generic[T]):
    """Modèle générique pour les réponses paginées."""
    items: List[T]
    total: int
    page: int
    taille_page: int
    total_pages: int
    
    @classmethod
    def creer(cls, items, total, page, taille_page):
        return cls(
            items=items,
            total=total,
            page=page,
            taille_page=taille_page,
            total_pages=(total + taille_page - 1) // taille_page
        )

@router.get("/articles", response_model=ReponsePagee[ArticleReponse])
async def lire_articles(
    page: int = 1,
    taille: int = 20,
    db: AsyncSession = Depends(get_db)
):
    if page < 1:
        raise HTTPException(status_code=400, detail="La page doit être >= 1")
    if taille > 100:
        raise HTTPException(status_code=400, detail="Taille max : 100")
    
    skip = (page - 1) * taille
    
    # Compter le total
    total = await db.scalar(select(func.count()).select_from(Article))
    
    # Récupérer les articles de la page
    resultat = await db.execute(
        select(Article).offset(skip).limit(taille).order_by(Article.cree_le.desc())
    )
    articles = resultat.scalars().all()
    
    return ReponsePagee.creer(articles, total, page, taille)
```

---

### [GUIDE] 14.4 — Compression des réponses

```python
from fastapi.middleware.gzip import GZipMiddleware

# Compresse automatiquement les réponses > 1000 bytes
app.add_middleware(GZipMiddleware, minimum_size=1000)
```

---

### [LOGIQUE] Clés à retenir

- Redis est une base de données **en mémoire** — beaucoup plus rapide qu'une DB SQL
- Le cache a une durée de vie (TTL) — les données expirées sont automatiquement supprimées
- La pagination évite de charger des milliers d'enregistrements inutilement
- La compression réduit la bande passante (utile pour les gros JSON)

### [OK] Bonne pratique

> **Profiler avant d'optimiser.** N'ajoutez du cache que pour les routes qui en ont vraiment besoin (routes lentes et fréquemment appelées). Un cache mal géré peut causer des bugs difficiles à détecter (données obsolètes).

### [IDEE] Exercice de synthèse — Parties 3 & 4

Créez une API sécurisée de gestion de blog :
1. Inscription/Login avec JWT
2. Création d'articles (authentification requise)
3. Publication d'articles (rôle "auteur" ou "admin")
4. Cache Redis sur la liste des articles populaires
5. Pagination sur la liste des articles
6. Rate limiting sur l'endpoint de login (5 tentatives/minute)

---

*Fin des Parties 3 & 4 — Sécurité, Authentification, Performance*

> **Prochaine étape :** Partie 5 — Communication & Intégrations (Uploads, WebSockets, Emails)


# [RAPIDE] FastAPI — Fil Rouge Étudiant
# [RESEAU] PARTIE 5 — Communication & Intégrations
# [LOGIQUE] PARTIE 6 — Tests, CI/CD et Déploiement

> **Prérequis :** Avoir complété les Parties 1 à 4

---

## Chapitre 15 : Uploads & Downloads de Fichiers

### [OBJECTIF] Objectifs du chapitre
- Gérer l'upload de fichiers avec `UploadFile`
- Télécharger des fichiers depuis l'API
- Limiter la taille et valider le type des fichiers

---

### [GUIDE] 15.1 — Upload de fichiers

```python
from fastapi import APIRouter, UploadFile, File, HTTPException
from fastapi.responses import FileResponse
import aiofiles
import os
import uuid
from pathlib import Path

router = APIRouter(prefix="/fichiers", tags=["[DOSSIER] Fichiers"])

DOSSIER_UPLOADS = Path("uploads")
DOSSIER_UPLOADS.mkdir(exist_ok=True)

# Types MIME autorisés
TYPES_AUTORISES = {"image/jpeg", "image/png", "image/gif", "application/pdf"}
TAILLE_MAX = 5 * 1024 * 1024  # 5 Mo en bytes

@router.post("/upload")
async def uploader_fichier(
    fichier: UploadFile = File(..., description="Fichier à uploader (max 5 Mo)")
):
    # Vérifier le type MIME
    if fichier.content_type not in TYPES_AUTORISES:
        raise HTTPException(
            status_code=400,
            detail=f"Type de fichier non autorisé. Types acceptés : {TYPES_AUTORISES}"
        )
    
    # Lire le contenu pour vérifier la taille
    contenu = await fichier.read()
    
    if len(contenu) > TAILLE_MAX:
        raise HTTPException(
            status_code=413,
            detail=f"Fichier trop grand. Taille max : {TAILLE_MAX // 1024 // 1024} Mo"
        )
    
    # Générer un nom de fichier unique (évite les conflits et les path traversal)
    extension = os.path.splitext(fichier.filename)[1].lower()
    nom_fichier = f"{uuid.uuid4()}{extension}"
    chemin = DOSSIER_UPLOADS / nom_fichier
    
    # Écrire le fichier de manière asynchrone
    async with aiofiles.open(chemin, "wb") as f:
        await f.write(contenu)
    
    return {
        "nom_original": fichier.filename,
        "nom_stocke": nom_fichier,
        "taille": len(contenu),
        "type": fichier.content_type,
        "url": f"/fichiers/{nom_fichier}"
    }

@router.post("/upload-multiple")
async def uploader_plusieurs_fichiers(
    fichiers: list[UploadFile] = File(...)
):
    if len(fichiers) > 10:
        raise HTTPException(status_code=400, detail="Maximum 10 fichiers à la fois")
    
    resultats = []
    for fichier in fichiers:
        # ... même logique que ci-dessus ...
        resultats.append({"nom": fichier.filename})
    
    return resultats
```

---

### [GUIDE] 15.2 — Téléchargement et streaming de fichiers

```python
from fastapi.responses import FileResponse, StreamingResponse

@router.get("/{nom_fichier}")
async def telecharger_fichier(nom_fichier: str):
    chemin = DOSSIER_UPLOADS / nom_fichier
    
    if not chemin.exists():
        raise HTTPException(status_code=404, detail="Fichier non trouvé")
    
    # Sécurité : empêcher path traversal
    if not chemin.resolve().is_relative_to(DOSSIER_UPLOADS.resolve()):
        raise HTTPException(status_code=403, detail="Accès refusé")
    
    return FileResponse(
        path=chemin,
        filename=nom_fichier,  # Nom affiché dans le navigateur
        media_type="application/octet-stream"  # Force le téléchargement
    )

# Streaming pour les gros fichiers
@router.get("/stream/{nom_fichier}")
async def streamer_fichier(nom_fichier: str):
    chemin = DOSSIER_UPLOADS / nom_fichier
    
    async def generateur():
        async with aiofiles.open(chemin, "rb") as f:
            while chunk := await f.read(65536):  # 64 Ko par chunk
                yield chunk
    
    return StreamingResponse(
        generateur(),
        media_type="application/octet-stream",
        headers={"Content-Disposition": f"attachment; filename={nom_fichier}"}
    )
```

---

### [GUIDE] 15.3 — Génération dynamique de CSV

```python
import csv
import io
from fastapi.responses import StreamingResponse

@router.get("/export/utilisateurs.csv")
async def exporter_utilisateurs_csv(db: AsyncSession = Depends(get_db)):
    utilisateurs = await db.execute(select(Utilisateur))
    
    # Créer le CSV en mémoire
    sortie = io.StringIO()
    writer = csv.DictWriter(sortie, fieldnames=["id", "nom", "email", "cree_le"])
    writer.writeheader()
    
    for u in utilisateurs.scalars():
        writer.writerow({
            "id": u.id,
            "nom": u.nom,
            "email": u.email,
            "cree_le": u.cree_le.isoformat()
        })
    
    sortie.seek(0)
    
    return StreamingResponse(
        iter([sortie.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=utilisateurs.csv"}
    )
```

---

## Chapitre 16 : WebSockets & Streaming Temps Réel

### [OBJECTIF] Objectifs du chapitre
- Comprendre le protocole WebSocket
- Implémenter un chat temps réel
- Gérer plusieurs connexions simultanées

---

### [GUIDE] 16.1 — Introduction aux WebSockets

HTTP est un protocole **requête-réponse** : le client demande, le serveur répond, la connexion se ferme. Impossible de pusher des données côté serveur.

WebSocket est un protocole **bidirectionnel persistant** : une fois la connexion établie, serveur et client peuvent s'envoyer des messages à tout moment.

**Cas d'utilisation :**
- Chat en temps réel
- Notifications en direct
- Tableaux de bord en temps réel (prix, stocks...)
- Jeux multijoueurs

---

### [GUIDE] 16.2 — Implémentation d'un chat simple

```python
from fastapi import WebSocket, WebSocketDisconnect
from typing import Dict

router = APIRouter(prefix="/ws", tags=["[SPEECH_BALLOON] WebSocket"])

# Gestionnaire de connexions
class GestionnaireConnexions:
    def __init__(self):
        # Dictionnaire : room_id -> liste des connexions
        self.connexions_actives: Dict[str, list[WebSocket]] = {}
    
    async def connecter(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.connexions_actives:
            self.connexions_actives[room] = []
        self.connexions_actives[room].append(websocket)
    
    def deconnecter(self, websocket: WebSocket, room: str):
        if room in self.connexions_actives:
            self.connexions_actives[room].remove(websocket)
    
    async def diffuser_message(self, message: str, room: str, expediteur: WebSocket = None):
        """Envoie un message à tous les clients d'une room (sauf l'expéditeur)."""
        if room in self.connexions_actives:
            connexions_mortes = []
            for connexion in self.connexions_actives[room]:
                if connexion != expediteur:
                    try:
                        await connexion.send_text(message)
                    except Exception:
                        connexions_mortes.append(connexion)
            
            # Nettoyer les connexions fermées
            for c in connexions_mortes:
                self.connexions_actives[room].remove(c)

gestionnaire = GestionnaireConnexions()

@router.websocket("/chat/{room_id}")
async def websocket_chat(
    websocket: WebSocket,
    room_id: str,
    utilisateur: str = "Anonyme"
):
    await gestionnaire.connecter(websocket, room_id)
    
    try:
        # Notifier l'arrivée
        await gestionnaire.diffuser_message(
            f"[VERT] {utilisateur} a rejoint la room",
            room_id
        )
        
        # Boucle de réception des messages
        while True:
            message = await websocket.receive_text()
            await gestionnaire.diffuser_message(
                f"{utilisateur}: {message}",
                room_id,
                expediteur=websocket  # Ne pas renvoyer à l'expéditeur
            )
    
    except WebSocketDisconnect:
        gestionnaire.deconnecter(websocket, room_id)
        await gestionnaire.diffuser_message(
            f"[ROUGE] {utilisateur} a quitté la room",
            room_id
        )
```

**Client JavaScript pour tester :**
```javascript
const ws = new WebSocket("ws://localhost:8000/ws/chat/room1?utilisateur=Alice");

ws.onmessage = (event) => {
    console.log("Reçu:", event.data);
};

ws.send("Bonjour tout le monde !");
```

---

## Chapitre 17 : Background Tasks & Scheduling

### [OBJECTIF] Objectifs du chapitre
- Exécuter des tâches lourdes sans bloquer l'API
- Intégrer Celery pour les tâches distribuées
- Planifier des tâches périodiques

---

### [GUIDE] 17.1 — BackgroundTasks (simple)

Vu au Chapitre 13 — idéal pour les tâches légères (envoi d'email, log, notification).

```python
@router.post("/commandes")
async def passer_commande(
    commande: CommandeCreation,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db)
):
    nouvelle_commande = await creer_commande_db(db, commande)
    
    # Ces tâches s'exécutent après l'envoi de la réponse
    background_tasks.add_task(envoyer_confirmation_email, commande.email)
    background_tasks.add_task(notifier_entrepot, nouvelle_commande.id)
    background_tasks.add_task(mettre_a_jour_stock, commande.produit_id)
    
    return {"id": nouvelle_commande.id, "statut": "en_cours"}
```

---

### [GUIDE] 17.2 — Celery pour les tâches lourdes

Celery est un système de **file de tâches distribuées**. Idéal pour les traitements long (encodage vidéo, génération de rapports, envois massifs d'emails).

```bash
pip install celery redis
```

```python
# app/worker/celery_app.py

from celery import Celery

celery_app = Celery(
    "taches",
    broker="redis://localhost:6379/0",   # File d'attente
    backend="redis://localhost:6379/1",  # Stockage des résultats
    include=["app.worker.taches"]
)

celery_app.conf.update(
    task_serializer="json",
    result_expires=3600,  # Résultats gardés 1 heure
)
```

```python
# app/worker/taches.py

from app.worker.celery_app import celery_app

@celery_app.task(bind=True, max_retries=3)
def generer_rapport_pdf(self, user_id: int, filtres: dict):
    """Génère un rapport PDF complexe."""
    try:
        # Traitement long...
        rapport = construire_rapport(user_id, filtres)
        return {"chemin": rapport.chemin, "statut": "succes"}
    except Exception as exc:
        # Réessayer après 60 secondes
        raise self.retry(exc=exc, countdown=60)
```

```python
# Dans une route FastAPI
from app.worker.taches import generer_rapport_pdf

@router.post("/rapports")
async def demander_rapport(user_id: int):
    # Lance la tâche en arrière-plan (Celery worker)
    tache = generer_rapport_pdf.delay(user_id, {})
    
    return {
        "tache_id": tache.id,
        "message": "Rapport en cours de génération"
    }

@router.get("/rapports/{tache_id}/statut")
async def statut_rapport(tache_id: str):
    from celery.result import AsyncResult
    result = AsyncResult(tache_id)
    
    return {
        "statut": result.status,  # PENDING, STARTED, SUCCESS, FAILURE
        "resultat": result.result if result.ready() else None
    }
```

---

## Chapitre 18 : Envoi d'e-mails et Notifications

### [OBJECTIF] Objectifs du chapitre
- Envoyer des emails HTML asynchrones
- Implémenter des notifications Webhook

---

### [GUIDE] 18.1 — Configuration de l'envoi d'emails

```bash
pip install fastapi-mail
```

```python
# app/core/email.py

from fastapi_mail import FastMail, MessageSchema, ConnectionConfig
from pathlib import Path

config_email = ConnectionConfig(
    MAIL_USERNAME="votre@email.com",
    MAIL_PASSWORD="votre_mot_de_passe",
    MAIL_FROM="votre@email.com",
    MAIL_PORT=587,
    MAIL_SERVER="smtp.gmail.com",
    MAIL_STARTTLS=True,
    MAIL_SSL_TLS=False,
    USE_CREDENTIALS=True,
    TEMPLATE_FOLDER=Path(__file__).parent / "templates/email"
)

mailer = FastMail(config_email)

async def envoyer_email_bienvenue(email: str, nom: str):
    """Envoie un email de bienvenue avec template HTML."""
    message = MessageSchema(
        subject="Bienvenue sur notre plateforme !",
        recipients=[email],
        template_body={
            "nom": nom,
            "lien_confirmation": f"https://monapp.com/confirmer/{email}"
        },
        subtype="html"
    )
    
    await mailer.send_message(message, template_name="bienvenue.html")
```

```html
<!-- app/core/templates/email/bienvenue.html -->
<!DOCTYPE html>
<html>
<body>
    <h1>Bonjour {{ nom }} !</h1>
    <p>Votre compte a été créé avec succès.</p>
    <a href="{{ lien_confirmation }}">Confirmer mon email</a>
</body>
</html>
```

---

# [LOGIQUE] PARTIE 6 — Tests, CI/CD et Déploiement

---

## Chapitre 19 : Tests Unitaires et d'Intégration

### [OBJECTIF] Objectifs du chapitre
- Tester les endpoints FastAPI avec pytest
- Mocker les dépendances (base de données, auth)
- Mesurer la couverture de tests

---

### [GUIDE] 19.1 — Configuration des tests

```bash
pip install pytest pytest-asyncio httpx pytest-cov
```

```python
# tests/conftest.py — Configuration partagée des tests

import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.db.session import get_db, Base

# Base de données de test (SQLite en mémoire)
SQLITE_TEST_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture
async def db_test():
    """Crée une base de données de test fraîche pour chaque test."""
    engine_test = create_async_engine(SQLITE_TEST_URL)
    
    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    AsyncSessionTest = sessionmaker(engine_test, class_=AsyncSession)
    
    async with AsyncSessionTest() as session:
        yield session
    
    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture
async def client(db_test):
    """Client HTTP pour tester l'API."""
    
    # Remplacer la vraie DB par la DB de test
    async def get_db_test():
        yield db_test
    
    app.dependency_overrides[get_db] = get_db_test
    
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
    
    app.dependency_overrides.clear()
```

---

### [GUIDE] 19.2 — Écrire des tests

```python
# tests/test_utilisateurs.py

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_creer_utilisateur(client: AsyncClient):
    """Test la création d'un utilisateur valide."""
    reponse = await client.post(
        "/utilisateurs/",
        json={
            "nom": "Alice Dupont",
            "email": "alice@test.com",
            "mot_de_passe": "MotDePasse123!"
        }
    )
    
    assert reponse.status_code == 201
    donnees = reponse.json()
    assert donnees["email"] == "alice@test.com"
    assert donnees["nom"] == "Alice Dupont"
    assert "mot_de_passe" not in donnees  # Le mdp ne doit JAMAIS être retourné !
    assert "id" in donnees

@pytest.mark.asyncio
async def test_creer_utilisateur_email_duplique(client: AsyncClient):
    """Test que l'email dupliqué est refusé."""
    # Créer une première fois
    await client.post("/utilisateurs/", json={
        "nom": "Alice", "email": "alice@test.com", "mot_de_passe": "Pass123!"
    })
    
    # Essayer de créer avec le même email
    reponse = await client.post("/utilisateurs/", json={
        "nom": "Bob", "email": "alice@test.com", "mot_de_passe": "Pass456!"
    })
    
    assert reponse.status_code == 400
    assert "email" in reponse.json()["detail"].lower()

@pytest.mark.asyncio
async def test_lire_utilisateur_inexistant(client: AsyncClient):
    """Test la récupération d'un utilisateur qui n'existe pas."""
    reponse = await client.get("/utilisateurs/99999")
    assert reponse.status_code == 404

@pytest.mark.asyncio
async def test_login_succes(client: AsyncClient):
    """Test un login réussi."""
    # Créer l'utilisateur
    await client.post("/utilisateurs/", json={
        "nom": "Test User", "email": "test@test.com", "mot_de_passe": "Pass123!"
    })
    
    # Login
    reponse = await client.post("/login", data={
        "username": "test@test.com",
        "password": "Pass123!"
    })
    
    assert reponse.status_code == 200
    donnees = reponse.json()
    assert "access_token" in donnees
    assert donnees["token_type"] == "bearer"

@pytest.mark.asyncio
async def test_route_protegee_sans_token(client: AsyncClient):
    """Test l'accès à une route protégée sans token."""
    reponse = await client.get("/mon-profil")
    assert reponse.status_code == 401

@pytest.mark.asyncio
async def test_pagination(client: AsyncClient):
    """Test que la pagination fonctionne correctement."""
    # Créer 15 utilisateurs
    for i in range(15):
        await client.post("/utilisateurs/", json={
            "nom": f"User {i}",
            "email": f"user{i}@test.com",
            "mot_de_passe": "Pass123!"
        })
    
    # Première page de 10
    reponse = await client.get("/utilisateurs/?page=1&taille=10")
    assert reponse.status_code == 200
    donnees = reponse.json()
    assert len(donnees["items"]) == 10
    assert donnees["total"] == 15
    assert donnees["total_pages"] == 2

# Lancer les tests :
# pytest tests/ -v --cov=app --cov-report=html
```

---

### [GUIDE] 19.3 — Tests avec authentification

```python
@pytest_asyncio.fixture
async def token_utilisateur(client: AsyncClient):
    """Fixture qui crée un utilisateur et retourne son token."""
    await client.post("/utilisateurs/", json={
        "nom": "Test", "email": "authed@test.com", "mot_de_passe": "Pass123!"
    })
    reponse = await client.post("/login", data={
        "username": "authed@test.com", "password": "Pass123!"
    })
    return reponse.json()["access_token"]

@pytest.mark.asyncio
async def test_route_protegee_avec_token(
    client: AsyncClient,
    token_utilisateur: str
):
    reponse = await client.get(
        "/mon-profil",
        headers={"Authorization": f"Bearer {token_utilisateur}"}
    )
    assert reponse.status_code == 200
```

---

## Chapitre 20 : Logging & Monitoring

### [OBJECTIF] Objectifs du chapitre
- Configurer des logs structurés pour la production
- Intégrer Sentry pour la surveillance des erreurs

---

### [GUIDE] 20.1 — Logs JSON pour la production

```bash
pip install python-json-logger
```

```python
# app/core/logging.py

import logging
import sys
from pythonjsonlogger import jsonlogger

def configurer_logging(niveau: str = "INFO", format_json: bool = True):
    handler = logging.StreamHandler(sys.stdout)
    
    if format_json:
        # Format JSON — idéal pour les outils de centralisation (Loki, CloudWatch)
        formatter = jsonlogger.JsonFormatter(
            "%(asctime)s %(name)s %(levelname)s %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )
    else:
        # Format lisible — pour le développement
        formatter = logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
        )
    
    handler.setFormatter(formatter)
    
    logger = logging.getLogger()
    logger.setLevel(getattr(logging, niveau.upper()))
    logger.addHandler(handler)
```

---

### [GUIDE] 20.2 — Intégration Sentry

**Sentry** est une plateforme de surveillance qui capture les erreurs en production et envoie des alertes.

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

```python
# main.py

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

sentry_sdk.init(
    dsn="https://votre_dsn@sentry.io/projet_id",
    integrations=[FastApiIntegration()],
    traces_sample_rate=0.1,  # Tracer 10% des requêtes
    environment=os.getenv("ENVIRONNEMENT", "development"),
    release="1.0.0",
)
```

---

## Chapitre 21 : Déploiement

### [OBJECTIF] Objectifs du chapitre
- Dockeriser une application FastAPI
- Configurer Gunicorn + Uvicorn pour la production
- Mettre en place Nginx comme reverse proxy

---

### [GUIDE] 21.1 — Dockerfile

```dockerfile
# Dockerfile

# Image de base légère
FROM python:3.12-slim

# Variables d'environnement
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PORT=8000

# Dossier de travail
WORKDIR /app

# Installer les dépendances en premier (mise en cache Docker)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copier le code
COPY . .

# Exposer le port
EXPOSE $PORT

# Commande de démarrage
# Gunicorn gère plusieurs workers Uvicorn
CMD ["gunicorn", "main:app",
     "--worker-class", "uvicorn.workers.UvicornWorker",
     "--workers", "4",
     "--bind", "0.0.0.0:8000",
     "--timeout", "120"]
```

```yaml
# docker-compose.yml — Pour le développement

version: "3.9"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - ENVIRONNEMENT=development
      - URL_BASE_DE_DONNEES=postgresql+asyncpg://user:password@db/mabase
      - CLE_SECRETE=dev-secret-key
    volumes:
      - .:/app  # Hot reload en développement
    depends_on:
      - db
      - redis
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
  
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mabase
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:
```

---

### [GUIDE] 21.2 — Configuration Nginx

```nginx
# nginx.conf

upstream api_fastapi {
    server 127.0.0.1:8000;
}

server {
    listen 80;
    server_name monapi.com;
    
    # Rediriger vers HTTPS
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl;
    server_name monapi.com;
    
    ssl_certificate /etc/ssl/certs/monapi.crt;
    ssl_certificate_key /etc/ssl/private/monapi.key;
    
    # Limiter la taille des uploads
    client_max_body_size 10M;
    
    location / {
        proxy_pass http://api_fastapi;
        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;
    }
    
    # WebSocket
    location /ws {
        proxy_pass http://api_fastapi;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}
```

---

### [GUIDE] 21.3 — Nombre optimal de workers

```bash
# Règle générale : (2 × nombre_de_cœurs) + 1
# Pour un serveur 4 cœurs :
gunicorn main:app \
    --worker-class uvicorn.workers.UvicornWorker \
    --workers 9 \
    --bind 0.0.0.0:8000 \
    --access-logfile - \
    --error-logfile - \
    --log-level info
```

---

## Chapitre 22 : CI/CD avec GitHub Actions

### [OBJECTIF] Objectifs du chapitre
- Automatiser les tests à chaque push
- Construire et publier l'image Docker
- Déployer automatiquement sur un serveur

---

### [GUIDE] 22.1 — Pipeline GitHub Actions complet

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

name: CI/CD FastAPI

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

jobs:
  # ── Job 1 : Tests ───────────────────────────────────────────
  tests:
    name: [TEST] Tests
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: testpassword
          POSTGRES_DB: testdb
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      
      redis:
        image: redis:7
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      
      - name: Installer les dépendances
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio httpx pytest-cov
      
      - name: Lancer les tests
        env:
          URL_BASE_DE_DONNEES: "postgresql+asyncpg://postgres:testpassword@localhost/testdb"
          CLE_SECRETE: "test-secret-key-pour-ci"
          REDIS_URL: "redis://localhost:6379"
        run: |
          pytest tests/ -v --cov=app --cov-report=xml
      
      - name: Upload couverture
        uses: codecov/codecov-action@v4
        with:
          file: ./coverage.xml
  
  # ── Job 2 : Build Docker ────────────────────────────────────
  build:
    name: [DOCKER] Build & Push Docker
    runs-on: ubuntu-latest
    needs: tests  # S'exécute uniquement si les tests passent
    if: github.ref == 'refs/heads/main'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Login Docker Hub
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKERHUB_USERNAME }}
          password: ${{ secrets.DOCKERHUB_TOKEN }}
      
      - name: Build et push
        uses: docker/build-push-action@v5
        with:
          push: true
          tags: |
            monuser/monapi:latest
            monuser/monapi:${{ github.sha }}
  
  # ── Job 3 : Déploiement ─────────────────────────────────────
  deployer:
    name: [RAPIDE] Déploiement
    runs-on: ubuntu-latest
    needs: build
    
    steps:
      - name: Déployer sur le serveur
        uses: appleboy/ssh-action@v1
        with:
          host: ${{ secrets.SERVEUR_IP }}
          username: deploy
          key: ${{ secrets.SSH_PRIVATE_KEY }}
          script: |
            docker pull monuser/monapi:latest
            docker-compose -f /app/docker-compose.prod.yml up -d --no-deps api
            docker image prune -f
```

---

### [OK] Bonne pratique

> Ne committez **jamais** de secrets dans votre code. Utilisez les **GitHub Secrets** pour stocker les mots de passe, clés API, tokens. Accessible dans Settings -> Secrets -> Actions.

### [IDEE] Exercice final des Parties 5 & 6

Mettez en place un pipeline complet pour votre API de blog :
1. Tests automatiques avec pytest (viser 80% de couverture)
2. Dockerfile multi-stage (builder + image finale légère)
3. docker-compose avec PostgreSQL, Redis, et l'API
4. GitHub Actions : tests -> build Docker -> déploiement
5. Monitoring Sentry et logs JSON

---

*Fin des Parties 5 & 6 — Communication, Tests, CI/CD & Déploiement*

> **Prochaine étape :** Partie 7 — Niveau Expert (Architecture avancée, Microservices, Observabilité)
