# [RAPIDE] NexaHub — Projet Fil Rouge FastAPI
## *De zéro à Expert : Construire une plateforme SaaS collaborative*

---

## [OBJECTIF] Présentation du Projet Fil Rouge : NexaHub

**NexaHub** est une plateforme collaborative de gestion de projets et de tâches en temps réel — pensez à un croisement entre Jira, Notion et Slack. Ce projet accompagne chaque chapitre de votre formation. À chaque étape, vous ajoutez une brique concrète à cette application réelle.

### Ce que NexaHub permettra de faire (au final) :
- Créer des comptes utilisateurs et s'authentifier
- Créer des projets, des boards (tableaux Kanban), des tâches
- Assigner des tâches à des membres d'équipe
- Collaborer en temps réel (WebSockets)
- Uploader des fichiers attachés aux tâches
- Recevoir des notifications par email
- Être déployé en production via Docker + CI/CD

> **Philosophie pédagogique** : Chaque chapitre introduit une notion théorique, l'explique en profondeur, puis vous demande de l'implémenter dans NexaHub. Les exercices de complétion sont des blocs de code *volontairement incomplets* que vous devez finir.

---

## [DOSSIER] Structure Finale du Projet (aperçu anticipé)

```
nexahub/
├── app/
│   ├── main.py                  # Point d'entrée
│   ├── api/
│   │   ├── routes/
│   │   │   ├── users.py
│   │   │   ├── projects.py
│   │   │   ├── tasks.py
│   │   │   └── websockets.py
│   │   ├── models/              # Modèles ORM (SQLAlchemy)
│   │   └── schemas/             # Schémas Pydantic
│   ├── core/
│   │   ├── config.py            # Variables d'environnement
│   │   ├── security.py          # JWT, hashing
│   │   └── dependencies.py      # DI globales
│   ├── db/
│   │   ├── session.py           # Connexion DB
│   │   └── migrations/
│   ├── services/                # Logique métier
│   ├── tasks/                   # Celery / Background tasks
│   └── tests/
├── .env
├── docker-compose.yml
├── Dockerfile
└── requirements.txt
```

---

# [MODULE] PARTIE 1 — Fondamentaux de FastAPI

---

## [LIVRE] Chapitre 1 : Introduction à FastAPI

### [OBJECTIF] Objectifs du chapitre
À la fin de ce chapitre, vous saurez :
- Ce qu'est FastAPI et pourquoi il est puissant
- Comment installer l'environnement de développement
- Créer et lancer votre première API

---

### 1.1 — Qu'est-ce que FastAPI ?

FastAPI est un **framework web moderne pour Python**, conçu pour créer des APIs REST (et plus) avec une productivité maximale et une performance optimale. Il a été créé par **Sebastián Ramírez** et publié en 2018. Depuis, il est devenu l'un des frameworks Python les plus populaires et les plus adoptés en production.

Voici ce qui rend FastAPI unique :

#### [RAPIDE] Performance de niveau production
FastAPI est construit sur **Starlette** (pour la partie réseau) et **Pydantic** (pour la validation). Starlette utilise ASGI (Asynchronous Server Gateway Interface), qui permet de traiter de nombreuses requêtes en parallèle sans bloquer. C'est ce qui le rend aussi rapide que Node.js ou Go pour des tâches I/O intensives.

#### [INPUT_SYMBOL_FOR_LATIN_SMALL_LETTERS] Typage Python natif
FastAPI exploite pleinement les **type hints** de Python 3.6+. Cela signifie que vous écrivez du Python normal avec des annotations de types, et FastAPI en déduit automatiquement la validation, la documentation, et la sérialisation. Pas de configuration séparée.

#### [FICHIER] Documentation Automatique
Dès que vous écrivez une route, FastAPI génère automatiquement deux interfaces de documentation interactives :
- **Swagger UI** (accessible à `/docs`) : vous pouvez tester vos endpoints directement depuis le navigateur
- **ReDoc** (accessible à `/redoc`) : une documentation plus lisible et élégante

#### [SYNC] Asynchronisme natif
FastAPI supporte nativement `async/await`, ce qui permet de faire des appels à des bases de données, APIs externes, ou fichiers sans bloquer le serveur.

---

### 1.2 — Comparaison avec les autres frameworks

| Critère | FastAPI | Flask | Django | Express.js |
|---|---|---|---|---|
| Langage | Python | Python | Python | JavaScript |
| Performance | ***** | *** | *** | **** |
| Validation auto | [OK] | [X] (manuel) | Partiel | [X] |
| Documentation auto | [OK] | [X] | [X] | [X] |
| Async natif | [OK] | Partiel | Partiel | [OK] |
| Courbe d'apprentissage | Douce | Très douce | Raide | Douce |
| Adapté pour les APIs | [OK][OK] | [OK] | Peut, mais lourd | [OK] |

**Flask** est simple mais n'offre aucune validation ou documentation intégrée. Pour des APIs sérieuses, vous finissez toujours par ajouter des dizaines de bibliothèques tierces.

**Django** est un framework "batteries included" pensé pour les applications web complètes avec rendu HTML, admin, ORM complet, etc. Excellent pour les applis Django REST Framework, mais verbeux pour des APIs pures.

**FastAPI** combine le meilleur des deux : la simplicité de Flask avec la puissance de Django, en y ajoutant l'asynchronisme et la validation automatique.

---

### 1.3 — Installation de l'environnement

#### Créer un environnement virtuel (obligatoire !)

Un environnement virtuel isole vos dépendances Python par projet. C'est une bonne pratique fondamentale.

```bash
# Créer le dossier du projet
mkdir nexahub && cd nexahub

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

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

# Vous devriez voir (venv) devant votre prompt
```

#### Installer FastAPI et Uvicorn

```bash
pip install fastapi uvicorn[standard]
```

- **fastapi** : le framework lui-même
- **uvicorn[standard]** : le serveur ASGI qui va exécuter votre application. Le `[standard]` installe des extras comme le support `watchfiles` pour le rechargement automatique.

---

### 1.4 — Votre première API : Hello NexaHub

Créez un fichier `main.py` :

```python
# main.py

# On importe la classe FastAPI depuis le module fastapi
from fastapi import FastAPI

# On crée une instance de l'application
# C'est l'objet central qui gère toutes les routes et la configuration
app = FastAPI(
    title="NexaHub API",          # Titre affiché dans Swagger
    description="La plateforme collaborative de gestion de projets",
    version="0.1.0"
)

# Le décorateur @app.get("/") définit une route HTTP GET sur "/"
# La fonction en dessous est appelée quand quelqu'un visite "/"
@app.get("/")
def root():
    # On retourne un dictionnaire Python
    # FastAPI le convertit automatiquement en JSON
    return {
        "message": "Bienvenue sur NexaHub API [RAPIDE]",
        "version": "0.1.0",
        "status": "running"
    }

# Route de santé : utile pour les health checks en production
@app.get("/health")
def health_check():
    return {"status": "healthy"}
```

#### Lancer le serveur de développement

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

Décortiquons cette commande :
- `uvicorn` : le serveur ASGI
- `main` : le nom du fichier Python (sans `.py`)
- `:app` : le nom de la variable FastAPI dans ce fichier
- `--reload` : redémarre automatiquement le serveur quand vous modifiez un fichier (à n'utiliser qu'en développement !)

Vous devriez voir :
```
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [...]
```

Ouvrez maintenant :
- `http://localhost:8000` -> votre réponse JSON
- `http://localhost:8000/docs` -> Swagger UI (magique !)
- `http://localhost:8000/redoc` -> ReDoc

---

### [OK] Bonnes Pratiques du Chapitre 1

> **Toujours utiliser `--reload` en développement, jamais en production.**
> En production, vous utiliserez Gunicorn avec des workers Uvicorn pour gérer plusieurs processus simultanément (voir Chapitre 21).

> **Toujours créer un environnement virtuel** par projet. Ne jamais installer des packages globalement avec `pip` en dehors d'un venv.

> **Nommez votre instance `app`** : c'est la convention universelle dans l'écosystème FastAPI.

---

### [EFFORT] Exercice de Complétion — Chapitre 1

Complétez le code suivant pour ajouter deux nouvelles routes à votre API NexaHub :

```python
from fastapi import FastAPI

app = FastAPI(
    title=___,          # [NOTE] À compléter : donnez un titre à votre API
    version="0.1.0"
)

@app.___.("/")          # [NOTE] À compléter : quelle méthode HTTP pour lire ?
def root():
    return {"message": "Bienvenue sur NexaHub"}

# Route qui retourne des informations sur l'API
@app.get(___)           # [NOTE] À compléter : quel chemin pour "/about" ?
def about():
    return {
        "name": ___,    # [NOTE] À compléter : nom de l'app
        "author": ___,  # [NOTE] À compléter : votre nom
        "features": ["gestion de projets", "tâches", "collaboration"]
    }

# Route de health check
# [NOTE] À compléter : écrivez vous-même cette route entièrement
# Elle doit retourner {"status": "ok", "uptime": "running"}
```

---

## [LIVRE] Chapitre 2 : Routes et Méthodes HTTP

### [OBJECTIF] Objectifs du chapitre
- Maîtriser toutes les méthodes HTTP et leurs usages
- Comprendre les paramètres de route, de requête, et le corps
- Structurer les réponses correctement

---

### 2.1 — Les Méthodes HTTP et leur sémantique

HTTP définit des **méthodes** (ou verbes) qui indiquent l'intention de la requête. Chaque méthode a une sémantique précise que vous devez respecter :

| Méthode | Usage sémantique | Exemple NexaHub |
|---|---|---|
| `GET` | Lire/récupérer des données, sans effet de bord | `GET /projects` -> liste des projets |
| `POST` | Créer une nouvelle ressource | `POST /projects` -> créer un projet |
| `PUT` | Remplacer entièrement une ressource | `PUT /tasks/{id}` -> remplacer une tâche |
| `PATCH` | Modifier partiellement une ressource | `PATCH /tasks/{id}` -> mettre à jour le statut |
| `DELETE` | Supprimer une ressource | `DELETE /projects/{id}` -> supprimer un projet |

> **Règle d'or** : `GET` ne doit **jamais** modifier des données. Un `GET` doit être idempotent : appeler la même route 10 fois donne le même résultat sans effet secondaire.

---

### 2.2 — Paramètres de Route (Path Parameters)

Un paramètre de route est une variable **intégrée dans le chemin URL**, entourée d'accolades `{}`.

```python
from fastapi import FastAPI

app = FastAPI()

# Base de données simulée (on utilisera une vraie DB au Chapitre 6)
fake_projects_db = [
    {"id": 1, "name": "Site e-commerce", "owner": "alice"},
    {"id": 2, "name": "App mobile", "owner": "bob"},
    {"id": 3, "name": "Dashboard analytics", "owner": "alice"},
]

@app.get("/projects/{project_id}")
def get_project(project_id: int):
    # project_id est automatiquement :
    # 1. Extrait de l'URL
    # 2. Converti en int (grâce au type hint)
    # 3. Validé (si ce n'est pas un int, FastAPI retourne une erreur 422)
    
    for project in fake_projects_db:
        if project["id"] == project_id:
            return project
    
    # Si on arrive ici, le projet n'existe pas
    # On verra HTTPException en détail au Chapitre 7
    from fastapi import HTTPException
    raise HTTPException(status_code=404, detail="Projet non trouvé")
```

**Ce qui se passe en coulisses** :
- Vous visitez `/projects/2`
- FastAPI extrait `"2"` de l'URL (c'est une chaîne)
- Grâce au type hint `project_id: int`, FastAPI convertit `"2"` -> `2`
- Si vous visitez `/projects/abc`, FastAPI retourne automatiquement :
```json
{
  "detail": [
    {
      "loc": ["path", "project_id"],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ]
}
```

---

### 2.3 — Paramètres de Requête (Query Parameters)

Les paramètres de requête apparaissent **après le `?`** dans l'URL : `/projects?owner=alice&limit=10`

```python
@app.get("/projects")
def list_projects(
    owner: str = None,        # Paramètre optionnel (None par défaut)
    limit: int = 10,          # Avec valeur par défaut
    skip: int = 0,            # Pour la pagination
    active: bool = True       # Boolean : ?active=true ou ?active=false
):
    """
    Liste les projets avec filtres optionnels.
    
    - **owner** : filtrer par propriétaire
    - **limit** : nombre max de résultats (défaut: 10)
    - **skip** : nombre de résultats à sauter (pour la pagination)
    """
    results = fake_projects_db
    
    if owner:
        results = [p for p in results if p["owner"] == owner]
    
    # Pagination simple
    return results[skip : skip + limit]
```

Exemples d'appels :
- `/projects` -> tous les projets (défauts)
- `/projects?owner=alice` -> projets d'alice
- `/projects?limit=2&skip=1` -> 2 projets en sautant le premier
- `/projects?owner=alice&limit=1` -> 1 projet d'alice

> **Distinction fondamentale** :
> - Paramètre de **route** `{id}` : identifie une ressource spécifique. Toujours obligatoire.
> - Paramètre de **requête** `?name=...` : filtre, pagination, options. Souvent optionnel.

---

### 2.4 — Corps de Requête (Request Body)

Pour `POST`, `PUT`, `PATCH`, les données sont envoyées dans le **corps** de la requête, pas dans l'URL. On utilise Pydantic (détaillé au Chapitre 3) :

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

app = FastAPI()

# Définition du schéma de données attendu
class ProjectCreate(BaseModel):
    name: str
    description: Optional[str] = None   # Optionnel
    is_public: bool = False              # Avec valeur par défaut

@app.post("/projects", status_code=201)  # 201 = Created
def create_project(project: ProjectCreate):
    # `project` est automatiquement rempli avec les données du corps JSON
    # project.name -> la valeur du champ "name"
    # project.description -> la valeur de "description" (ou None)
    
    new_project = {
        "id": len(fake_projects_db) + 1,
        "name": project.name,
        "description": project.description,
        "is_public": project.is_public
    }
    fake_projects_db.append(new_project)
    return new_project
```

Pour tester, vous envoyez :
```json
POST /projects
Content-Type: application/json

{
  "name": "NexaHub Frontend",
  "description": "Interface React pour NexaHub",
  "is_public": true
}
```

---

### 2.5 — Combiner paramètres de route, requête et body

FastAPI est intelligent : il sait distinguer les trois types automatiquement :
- Si le paramètre existe dans le chemin `{...}` -> **path parameter**
- Si le type est simple (`int`, `str`, `bool`) et PAS dans le chemin -> **query parameter**
- Si le type est un `BaseModel` Pydantic -> **request body**

```python
@app.put("/projects/{project_id}")
def update_project(
    project_id: int,                    # <- path param (dans l'URL)
    notify_team: bool = False,          # <- query param (?notify_team=true)
    project: ProjectCreate = None       # <- body (JSON envoyé)
):
    # FastAPI sait exactement où chercher chaque valeur !
    return {
        "updated_id": project_id,
        "notify_team": notify_team,
        "new_data": project
    }
```

---

### [OK] Bonnes Pratiques du Chapitre 2

> **Nommez vos endpoints de façon sémantique et RESTful :**
> - `/users` (pluriel) et non `/user`
> - `/projects/{id}/tasks` pour les sous-ressources
> - Ne mettez jamais de verbe dans l'URL : `/create-project` est **faux**. Utilisez `POST /projects`.

> **Retournez toujours le bon code HTTP :**
> - `200 OK` : succès d'un GET ou PUT
> - `201 Created` : ressource créée (POST)
> - `204 No Content` : succès sans contenu (DELETE)
> - `404 Not Found` : ressource inexistante
> - `422 Unprocessable Entity` : erreur de validation (géré automatiquement par FastAPI)

---

### [EFFORT] Exercice de Complétion — Chapitre 2

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

# Base de données simulée de tâches
fake_tasks = [
    {"id": 1, "title": "Créer le wireframe", "project_id": 1, "done": False},
    {"id": 2, "title": "Setup la base de données", "project_id": 1, "done": True},
    {"id": 3, "title": "Écrire les tests", "project_id": 2, "done": False},
]

class TaskUpdate(BaseModel):
    title: Optional[str] = None
    done: Optional[bool] = None

# [NOTE] À compléter : Route GET qui retourne toutes les tâches
# Avec un paramètre de requête optionnel `project_id` pour filtrer
@app.___(___) 
def list_tasks(___):
    pass  # À implémenter

# [NOTE] À compléter : Route GET qui retourne une tâche par son ID
# Si la tâche n'existe pas, lever une HTTPException 404
@app.get("/tasks/{___}")
def get_task(___: int):
    pass  # À implémenter

# [NOTE] À compléter : Route PATCH pour mettre à jour partiellement une tâche
# Elle doit accepter un body de type TaskUpdate
# et mettre à jour uniquement les champs fournis (non-None)
@app.___(___) 
def update_task(task_id: int, ___: ___):
    pass  # À implémenter

# [NOTE] À compléter : Route DELETE pour supprimer une tâche
# Retourner un statut 204 et aucun contenu
```

---

## [LIVRE] Chapitre 3 : Typage et Validation Automatique avec Pydantic

### [OBJECTIF] Objectifs du chapitre
- Comprendre Pydantic en profondeur
- Créer des modèles de données robustes
- Utiliser les validateurs et contraintes

---

### 3.1 — Introduction à Pydantic

**Pydantic** est une bibliothèque de validation de données qui utilise les type hints Python. C'est le moteur de validation de FastAPI. Sans Pydantic, FastAPI ne fonctionnerait pas.

Voici le problème que Pydantic résout :

```python
# Sans Pydantic : vous devez tout valider manuellement [WEARY_FACE]
def create_user(data: dict):
    if "name" not in data:
        raise ValueError("name requis")
    if not isinstance(data["name"], str):
        raise ValueError("name doit être une chaîne")
    if len(data["name"]) < 2:
        raise ValueError("name trop court")
    if "age" not in data:
        raise ValueError("age requis")
    if not isinstance(data["age"], int):
        try:
            data["age"] = int(data["age"])  # tentative de conversion
        except:
            raise ValueError("age invalide")
    # ... et ça continue

# Avec Pydantic : tout ça est automatique [OK]
from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    age: int = Field(ge=0, le=120)
```

---

### 3.2 — BaseModel et types de base

```python
from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from datetime import datetime
from enum import Enum

# Enum pour des valeurs fixes
class TaskStatus(str, Enum):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"
    CANCELLED = "cancelled"

class TaskBase(BaseModel):
    title: str = Field(
        ...,                        # `...` signifie obligatoire (Ellipsis)
        min_length=3,               # Minimum 3 caractères
        max_length=200,             # Maximum 200 caractères
        description="Le titre de la tâche"
    )
    description: Optional[str] = Field(
        default=None,               # Optionnel, None par défaut
        max_length=2000
    )
    status: TaskStatus = TaskStatus.TODO    # Valeur par défaut
    priority: int = Field(
        default=1,
        ge=1,                       # ge = greater or equal (>=)
        le=5,                       # le = less or equal (<=)
        description="Priorité de 1 (basse) à 5 (critique)"
    )
    due_date: Optional[datetime] = None     # Date d'échéance optionnelle
    tags: List[str] = []                    # Liste de tags (vide par défaut)
```

#### Contraintes disponibles dans `Field()`

| Contrainte | Description | Exemple |
|---|---|---|
| `min_length` | Longueur minimale (str) | `min_length=3` |
| `max_length` | Longueur maximale (str) | `max_length=100` |
| `ge` | Greater or Equal (≥) | `ge=0` |
| `gt` | Greater Than (>) | `gt=0` |
| `le` | Less or Equal (≤) | `le=100` |
| `lt` | Less Than (<) | `lt=100` |
| `regex` | Pattern regex | `regex=r"^[A-Z]"` |
| `default` | Valeur par défaut | `default=None` |
| `description` | Doc pour Swagger | `description="..."` |

---

### 3.3 — Schémas de Création, Lecture et Mise à Jour

Une bonne pratique est de définir **plusieurs schémas** pour le même objet selon l'opération :

```python
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

# Schéma de base : champs communs
class TaskBase(BaseModel):
    title: str = Field(..., min_length=3, max_length=200)
    description: Optional[str] = None
    priority: int = Field(default=1, ge=1, le=5)

# Schéma pour la CRÉATION (ce que le client envoie)
# Hérite de TaskBase, peut ajouter des champs spécifiques à la création
class TaskCreate(TaskBase):
    project_id: int     # Obligatoire à la création
    assigned_to: Optional[int] = None  # ID utilisateur assigné

# Schéma pour la MISE À JOUR (tous les champs optionnels !)
class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=3, max_length=200)
    description: Optional[str] = None
    priority: Optional[int] = Field(None, ge=1, le=5)
    status: Optional[str] = None
    # ^ Tous optionnels car on peut ne mettre à jour qu'un seul champ

# Schéma pour la LECTURE (ce que l'API retourne)
# Inclut les champs générés par le serveur (id, dates, etc.)
class TaskResponse(TaskBase):
    id: int
    project_id: int
    status: str
    created_at: datetime
    updated_at: Optional[datetime]
    
    class Config:
        # Permet de lire depuis des objets ORM (pas juste des dicts)
        # Sera utile au Chapitre 6 avec SQLAlchemy
        from_attributes = True  # Anciennement `orm_mode = True`
```

---

### 3.4 — Validateurs Personnalisés

```python
from pydantic import BaseModel, Field, validator, root_validator
from typing import Optional

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=30)
    email: str
    password: str = Field(..., min_length=8)
    password_confirm: str
    
    # Validateur sur un seul champ
    @validator("username")
    def username_must_be_alphanumeric(cls, v):
        # `v` est la valeur du champ
        if not v.replace("_", "").isalnum():
            raise ValueError("Le username ne peut contenir que des lettres, chiffres et underscores")
        return v.lower()  # On peut aussi transformer la valeur !
    
    @validator("email")
    def email_must_be_valid(cls, v):
        if "@" not in v or "." not in v.split("@")[-1]:
            raise ValueError("Email invalide")
        return v.lower()
    
    # Validateur qui vérifie plusieurs champs ensemble
    @root_validator
    def passwords_must_match(cls, values):
        password = values.get("password")
        password_confirm = values.get("password_confirm")
        
        if password and password_confirm and password != password_confirm:
            raise ValueError("Les mots de passe ne correspondent pas")
        
        return values
```

---

### 3.5 — Modèles imbriqués

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

class UserMini(BaseModel):
    id: int
    username: str
    avatar_url: Optional[str]

class CommentResponse(BaseModel):
    id: int
    content: str
    author: UserMini         # Objet imbriqué !

class TaskDetailResponse(BaseModel):
    id: int
    title: str
    assignee: Optional[UserMini]   # Peut être None
    comments: List[CommentResponse] = []  # Liste d'objets imbriqués
    
# Résultat JSON :
# {
#   "id": 1,
#   "title": "Ma tâche",
#   "assignee": {"id": 5, "username": "alice", "avatar_url": null},
#   "comments": [
#     {"id": 1, "content": "Super !", "author": {"id": 5, "username": "alice", ...}}
#   ]
# }
```

---

### [OK] Bonnes Pratiques du Chapitre 3

> **Toujours définir trois schémas distincts par entité :**
> - `XxxCreate` : pour la création (entrée utilisateur)
> - `XxxUpdate` : pour la mise à jour (tous champs optionnels)
> - `XxxResponse` : pour la lecture (inclut id, dates, etc.)

> **Ne jamais exposer le mot de passe dans `XxxResponse`**, même hashé.

> **Utilisez des `Enum` pour les champs à valeurs fixes** (statuts, rôles, catégories). Cela donne une validation automatique et une documentation claire.

---

### [EFFORT] Exercice de Complétion — Chapitre 3

Créez les schémas Pydantic complets pour l'entité `Project` de NexaHub :

```python
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

# [NOTE] À compléter : créer une Enum pour le statut du projet
# Valeurs possibles : "planning", "active", "on_hold", "completed", "archived"
class ProjectStatus(___):
    PLANNING = ___
    ACTIVE = ___
    # ... compléter

# [NOTE] À compléter : schéma de base avec les champs communs
class ProjectBase(BaseModel):
    name: str = Field(___, min_length=___, max_length=___)
    description: Optional[str] = ___
    # [NOTE] Ajouter un champ `deadline` optionnel de type datetime
    # [NOTE] Ajouter un champ `max_members` de type int avec contrainte ge=1, le=100
    
# [NOTE] À compléter : schéma de création
class ProjectCreate(___):  # Hérite de ?
    # [NOTE] Ajouter un validateur pour que `name` ne contienne pas de caractères spéciaux
    @validator("name")
    def name_must_be_clean(cls, v):
        # À implémenter
        pass

# [NOTE] À compléter : schéma de mise à jour (tous les champs optionnels)
class ProjectUpdate(BaseModel):
    # [NOTE] Reprendre tous les champs de ProjectBase mais en Optional
    pass

# [NOTE] À compléter : schéma de réponse
class ProjectResponse(___):  # Hérite de ?
    id: int
    status: ___  # Utiliser l'Enum
    owner_id: int
    created_at: datetime
    member_count: int = 0
    
    class Config:
        from_attributes = ___  # True ou False ?
```

---

## [LIVRE] Chapitre 4 : Documentation Automatique

### [OBJECTIF] Objectifs du chapitre
- Exploiter et personnaliser Swagger UI et ReDoc
- Documenter toutes les routes professionnellement
- Ajouter des exemples et descriptions riches

---

### 4.1 — Swagger UI et ReDoc : Comment ça marche

Quand FastAPI démarre, il génère automatiquement un **schéma OpenAPI** (anciennement Swagger) — un fichier JSON qui décrit toute votre API. Vous pouvez le voir à `/openapi.json`.

Ce schéma est ensuite utilisé par :
- **Swagger UI** (`/docs`) : interface interactive pour tester les endpoints
- **ReDoc** (`/redoc`) : documentation lisible et bien structurée

La beauté du système : **vous n'avez rien à configurer**. Tout ce que vous écrivez dans votre code (types, descriptions, exemples) se reflète automatiquement dans la documentation.

---

### 4.2 — Personnalisation de l'application

```python
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title="NexaHub API",
    description="""
## [RAPIDE] API de NexaHub — Gestion collaborative de projets

NexaHub est une plateforme de gestion de projets et de tâches en équipe.

### Fonctionnalités principales :
- **Projets** : Créer et gérer des projets collaboratifs
- **Tâches** : Assigner, suivre et compléter des tâches
- **Équipes** : Inviter des membres et gérer les permissions
- **Temps réel** : Collaboration en direct via WebSockets

### Authentification :
Toutes les routes (sauf `/` et `/health`) nécessitent un token JWT.
Obtenez votre token via `POST /auth/login`.
""",
    version="1.0.0",
    terms_of_service="https://nexahub.io/terms",
    contact={
        "name": "Équipe NexaHub",
        "url": "https://nexahub.io/support",
        "email": "dev@nexahub.io",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
    # Personnaliser les URLs de documentation
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json"
)
```

---

### 4.3 — Tags : Organiser les routes en groupes

Les **tags** permettent de regrouper les routes dans Swagger UI :

```python
# Définir les tags avec descriptions
tags_metadata = [
    {
        "name": "auth",
        "description": "Authentification et gestion des tokens JWT.",
    },
    {
        "name": "projects",
        "description": "Gestion des projets collaboratifs.",
        "externalDocs": {
            "description": "Documentation complète",
            "url": "https://nexahub.io/docs/projects",
        },
    },
    {
        "name": "tasks",
        "description": "Gestion des tâches au sein des projets.",
    },
    {
        "name": "users",
        "description": "Gestion des profils utilisateurs.",
    },
]

app = FastAPI(openapi_tags=tags_metadata)

# Utiliser les tags sur les routes
@app.get("/projects", tags=["projects"])
def list_projects():
    pass

@app.post("/projects", tags=["projects"])
def create_project():
    pass

@app.get("/users/me", tags=["users"])
def get_current_user():
    pass
```

---

### 4.4 — Documenter chaque route en détail

```python
from fastapi import FastAPI, Query, Path
from pydantic import BaseModel
from typing import Optional, List

@app.get(
    "/projects/{project_id}/tasks",
    tags=["tasks"],
    summary="Lister les tâches d'un projet",
    description="""
Retourne toutes les tâches associées à un projet spécifique.

Vous pouvez filtrer par statut et prioriser par date d'échéance.
Seuls les membres du projet ont accès à cette route.
""",
    response_description="Liste paginée des tâches",
    status_code=200,
    # Déclarer les réponses possibles pour la doc
    responses={
        200: {"description": "Liste des tâches retournée avec succès"},
        403: {"description": "Accès refusé — vous n'êtes pas membre du projet"},
        404: {"description": "Projet non trouvé"},
    }
)
def get_project_tasks(
    project_id: int = Path(
        ...,
        description="L'identifiant unique du projet",
        example=42,         # Exemple affiché dans Swagger
        ge=1
    ),
    status: Optional[str] = Query(
        default=None,
        description="Filtrer par statut",
        enum=["todo", "in_progress", "done"]   # Liste de valeurs autorisées
    ),
    limit: int = Query(
        default=20,
        ge=1,
        le=100,
        description="Nombre maximum de résultats"
    )
):
    """
    Cette docstring est aussi affichée dans Swagger !
    
    Vous pouvez utiliser **Markdown** dans la docstring.
    - Elle supporte les listes
    - Et le *formatage*
    """
    return {"project_id": project_id, "status": status, "limit": limit}
```

---

### 4.5 — Modèles d'exemples dans les schémas

```python
from pydantic import BaseModel, Field

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=3, max_length=200)
    description: Optional[str] = None
    priority: int = Field(1, ge=1, le=5)
    
    class Config:
        # Exemple affiché dans Swagger pour ce modèle
        json_schema_extra = {
            "example": {
                "title": "Implémenter l'authentification JWT",
                "description": "Créer les endpoints /login et /refresh avec validation complète",
                "priority": 4
            }
        }
```

---

### 4.6 — Désactiver la documentation en production

```python
import os

# En production, on peut vouloir cacher la documentation
IS_PRODUCTION = os.getenv("ENVIRONMENT", "development") == "production"

app = FastAPI(
    docs_url=None if IS_PRODUCTION else "/docs",
    redoc_url=None if IS_PRODUCTION else "/redoc",
)
```

---

### [OK] Bonnes Pratiques du Chapitre 4

> **Documentez TOUTES les routes** avec au moins un `summary` et un `description`. La documentation est votre contrat avec les développeurs front-end ou les utilisateurs de l'API.

> **Utilisez des tags cohérents** et définissez-les avec `openapi_tags` pour avoir une documentation organisée.

> **Ajoutez des exemples** dans vos schémas Pydantic. Cela rend Swagger UI immédiatement compréhensible.

---

### [EFFORT] Exercice de Complétion — Chapitre 4

```python
from fastapi import FastAPI, Query, Path
from pydantic import BaseModel, Field
from typing import Optional

# [NOTE] À compléter : créer l'app avec titre, description et metadata de tags
# Tags à créer : "projects", "tasks", "users", "health"
tags_metadata = [
    ___  # À compléter
]

app = FastAPI(
    title=___,
    description=___,   # Écrire une belle description en Markdown
    openapi_tags=___
)

class ProjectCreate(BaseModel):
    name: str
    description: Optional[str] = None
    
    class Config:
        # [NOTE] À compléter : ajouter un exemple JSON pour Swagger
        json_schema_extra = {
            "example": ___  # À compléter avec des données réalistes
        }

# [NOTE] À compléter : documenter cette route en détail
# - Ajouter summary, description, response_description
# - Ajouter des réponses possibles (200, 404, 403)
# - Utiliser Path() pour documenter project_id
# - Utiliser Query() pour documenter les paramètres de filtrage
@app.get("/projects/{project_id}")
def get_project(
    project_id: int,  # [NOTE] Remplacer par Path() documenté
    include_tasks: bool = False  # [NOTE] Remplacer par Query() documenté
):
    return {"id": project_id, "include_tasks": include_tasks}

# [NOTE] À compléter : route POST /projects avec tag "projects"
# Corps de type ProjectCreate, summary et description complets
@app.___(
    ___,
    ___,   # tag
    ___,   # summary
    ___,   # description
)
def create_project(___: ___):
    pass
```

---

## [OBJECTIF] Récapitulatif — Partie 1

À ce stade, vous maîtrisez :

| Notion | Niveau atteint |
|---|---|
| Structure d'une app FastAPI | [OK] Acquis |
| Méthodes HTTP et leurs usages | [OK] Acquis |
| Paramètres de route, requête, body | [OK] Acquis |
| Modèles Pydantic et validation | [OK] Acquis |
| Documentation Swagger/ReDoc | [OK] Acquis |

### Ce qu'on a construit pour NexaHub :
- Routes de base pour les projets et les tâches
- Schémas Pydantic pour la création/lecture/mise à jour
- Documentation interactive fonctionnelle

### Prochain chapitre :
Dans la **Partie 2**, on va structurer sérieusement le projet, le connecter à une vraie base de données, et implémenter les opérations CRUD complètes.

---

*[BOOKMARK] Fichier suivant : `fastapi_nexahub_partie2.md` — Architecture & Base de Données*

# [RAPIDE] NexaHub — Partie 2 : Architecture & Base de Données
## *Structurer, Persister, Gérer les Erreurs*

---

# [CONFIG] PARTIE 2 — Architecture et Structuration

---

## [LIVRE] Chapitre 5 : Structuration du Projet

### [OBJECTIF] Objectifs du chapitre
- Organiser un projet FastAPI professionnel et maintenable
- Comprendre le principe de séparation des responsabilités
- Utiliser `APIRouter` pour modulariser les routes

---

### 5.1 — Pourquoi structurer son projet ?

Quand votre API grandit, tout mettre dans `main.py` devient rapidement un cauchemar : fichier de 2000 lignes illisible, impossible de travailler en équipe, difficile de tester. La solution : **diviser le projet en modules avec des responsabilités claires**.

---

### 5.2 — L'arborescence NexaHub

```
nexahub/
├── app/
│   ├── __init__.py
│   ├── main.py                    # Point d'entrée
│   ├── api/
│   │   ├── routes/
│   │   │   ├── auth.py            # /auth/*
│   │   │   ├── users.py           # /users/*
│   │   │   ├── projects.py        # /projects/*
│   │   │   ├── tasks.py           # /tasks/*
│   │   │   └── websockets.py      # /ws/*
│   │   └── dependencies.py        # DI partagées (auth, pagination)
│   ├── schemas/                   # Modèles Pydantic (validation I/O)
│   │   ├── user.py
│   │   ├── project.py
│   │   └── task.py
│   ├── models/                    # Modèles ORM SQLAlchemy (tables DB)
│   │   ├── user.py
│   │   ├── project.py
│   │   └── task.py
│   ├── services/                  # Logique métier (CRUD réel)
│   │   ├── user_service.py
│   │   ├── project_service.py
│   │   └── task_service.py
│   ├── core/
│   │   ├── config.py              # Variables d'environnement
│   │   ├── security.py            # JWT, hashing
│   │   └── exceptions.py          # Exceptions personnalisées
│   └── db/
│       └── session.py             # Connexion et sessions
├── tests/
├── .env
├── .env.example
├── requirements.txt
├── Dockerfile
└── docker-compose.yml
```

> **Règle fondamentale** : **1 fichier = 1 responsabilité**. Ne mélangez jamais schémas Pydantic, modèles ORM et logique métier dans le même fichier.

---

### 5.3 — APIRouter : Modulariser les routes

`APIRouter` permet de définir des routes dans des fichiers séparés et de les inclure dans l'application principale.

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

from fastapi import APIRouter, Depends, HTTPException, status
from typing import List
from app.schemas.project import ProjectCreate, ProjectUpdate, ProjectResponse

# Créer un router avec un préfixe et un tag
# Toutes les routes de ce fichier auront automatiquement "/projects" comme préfixe
router = APIRouter(
    prefix="/projects",
    tags=["projects"],
    responses={
        401: {"description": "Non authentifié"},
        403: {"description": "Permission refusée"},
    }
)

@router.get("/", response_model=List[ProjectResponse])
async def list_projects():
    """Liste tous les projets accessibles."""
    return []

@router.post("/", response_model=ProjectResponse, status_code=201)
async def create_project(project: ProjectCreate):
    """Crée un nouveau projet."""
    pass

@router.get("/{project_id}", response_model=ProjectResponse)
async def get_project(project_id: int):
    """Récupère un projet par son ID."""
    pass

@router.put("/{project_id}", response_model=ProjectResponse)
async def update_project(project_id: int, project: ProjectUpdate):
    """Met à jour un projet."""
    pass

@router.delete("/{project_id}", status_code=204)
async def delete_project(project_id: int):
    """Supprime un projet."""
    pass
```

---

### 5.4 — Assembler dans main.py

```python
# app/main.py

from fastapi import FastAPI
from app.api.routes import projects, tasks, users, auth

tags_metadata = [
    {"name": "auth", "description": "Authentification JWT"},
    {"name": "users", "description": "Gestion des utilisateurs"},
    {"name": "projects", "description": "Gestion des projets"},
    {"name": "tasks", "description": "Gestion des tâches"},
]

app = FastAPI(
    title="NexaHub API",
    version="1.0.0",
    openapi_tags=tags_metadata
)

# Inclure les routers — chaque router a déjà son préfixe défini
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(projects.router)
app.include_router(tasks.router)

@app.get("/", tags=["health"])
async def root():
    return {"message": "NexaHub API", "docs": "/docs"}
```

---

### 5.5 — Configuration centralisée avec pydantic-settings

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

from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    # Application
    APP_NAME: str = "NexaHub"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False
    ENVIRONMENT: str = "development"

    # Base de données
    DATABASE_URL: str = "sqlite+aiosqlite:///./nexahub.db"

    # JWT
    SECRET_KEY: str = "changeme-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7

    # CORS
    ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"
        case_sensitive = True

# Singleton global — importez `settings` partout
settings = Settings()
```

---

### [OK] Bonnes Pratiques du Chapitre 5

> **1 fichier = 1 responsabilité.** Ne mélangez jamais les schémas Pydantic, les modèles ORM et la logique métier.

> **Jamais de secret dans le code source.** Toujours utiliser des variables d'environnement via `.env`.

> Ajoutez `.env` dans `.gitignore` immédiatement et créez `.env.example` à la place.

---

### [EFFORT] Exercice de Complétion — Chapitre 5

```python
# app/api/routes/users.py
from fastapi import ___  # À compléter : quoi importer ?

router = ___(
    prefix=___,    # [NOTE] Quel préfixe pour les users ?
    tags=___       # [NOTE] Quel tag ?
)

# [NOTE] Ajouter ces 4 routes (corps vides pour l'instant) :
# GET  /users/         -> liste des utilisateurs
# GET  /users/me       -> profil de l'utilisateur connecté (IMPORTANT: avant {user_id} !)
# GET  /users/{user_id} -> profil d'un utilisateur
# PATCH /users/me      -> mettre à jour son profil

# [ATTENTION] Pourquoi /users/me doit être AVANT /users/{user_id} ?
# Réfléchissez et écrivez la réponse en commentaire.

# app/core/config.py
# [NOTE] Ajouter dans Settings :
# - REDIS_URL avec valeur par défaut "redis://localhost:6379"
# - MAX_UPLOAD_SIZE en octets (défaut : 10 * 1024 * 1024)
# - ALLOWED_EXTENSIONS comme liste (défaut : ["jpg", "png", "pdf"])
```

---

## [LIVRE] Chapitre 6 : Gestion des Données — Bases de Données

### [OBJECTIF] Objectifs du chapitre
- Connecter FastAPI à une base de données SQL de façon asynchrone
- Comprendre la différence fondamentale entre modèles ORM et schémas Pydantic
- Utiliser Alembic pour les migrations

---

### 6.1 — La Distinction Cruciale : ORM vs Pydantic

C'est **la confusion numéro 1** chez les débutants FastAPI. Voici ce qu'il faut retenir absolument :

```
┌─────────────────────────────────────────────────────────┐
│  SCHÉMA PYDANTIC (app/schemas/)                         │
│  -> Valide et sérialise les données de l'API             │
│  -> Hérite de BaseModel                                  │
│  -> Utilisé dans les routes (paramètres et retours)      │
│  -> Ce que voit l'utilisateur externe                    │
└─────────────────────────────────────────────────────────┘
              ^v Conversion dans les services

┌─────────────────────────────────────────────────────────┐
│  MODÈLE ORM (app/models/)                               │
│  -> Représente une table dans la base de données         │
│  -> Hérite de Base (SQLAlchemy)                          │
│  -> Utilisé dans les services (opérations DB)            │
│  -> Ce que voit la base de données                       │
└─────────────────────────────────────────────────────────┘
```

**Exemple concret** : le champ `hashed_password` existe dans le modèle ORM (il est stocké en DB) mais **ne doit jamais apparaître** dans le schéma Pydantic de réponse.

---

### 6.2 — Configuration SQLAlchemy Asynchrone

```bash
pip install sqlalchemy[asyncio] asyncpg aiosqlite alembic pydantic-settings
```

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

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings

# Le moteur gère le pool de connexions à la DB
engine = create_async_engine(
    settings.DATABASE_URL,
    echo=settings.DEBUG,     # Affiche les requêtes SQL en mode debug
    pool_size=10,            # Connexions maintenues en permanence
    max_overflow=20,         # Connexions supplémentaires possibles si besoin
    pool_pre_ping=True,      # Vérifie les connexions avant utilisation
)

# Factory de sessions
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False   # Important pour accéder aux objets après commit
)

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

# ─────────────────────────────────────────────────────────────────
# DÉPENDANCE FASTAPI — à injecter dans toutes les routes via Depends()
# ─────────────────────────────────────────────────────────────────
async def get_db():
    """
    Générateur qui crée une session DB, la fournit à la route,
    commit si tout va bien, rollback si une erreur se produit,
    et ferme toujours la session à la fin.

    Utilisation dans une route :
        async def ma_route(db: AsyncSession = Depends(get_db)):
            ...
    """
    async with AsyncSessionLocal() as session:
        try:
            yield session          # La route reçoit la session ici
            await session.commit() # Commit automatique si succès
        except Exception:
            await session.rollback() # Annuler tout si erreur
            raise
        finally:
            await session.close()  # Toujours fermer
```

---

### 6.3 — Modèles ORM

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

from sqlalchemy import String, Boolean, DateTime, Integer, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from app.db.session import Base
from datetime import datetime

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True, nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    full_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), onupdate=func.now()
    )

    # Relation : un User possède plusieurs Projects
    projects: Mapped[list["Project"]] = relationship(
        "Project", back_populates="owner", cascade="all, delete-orphan"
    )

    def __repr__(self):
        return f"<User {self.username}>"
```

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

from sqlalchemy import String, Boolean, DateTime, Integer, Text, ForeignKey, Table, Column
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from app.db.session import Base
from datetime import datetime

# Table d'association many-to-many : Projet <-> Membres
project_members = Table(
    "project_members",
    Base.metadata,
    Column("project_id", Integer, ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True),
    Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
)

class Project(Base):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    status: Mapped[str] = mapped_column(String(50), default="active")
    is_public: Mapped[bool] = mapped_column(Boolean, default=False)
    owner_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False)

    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), onupdate=func.now())

    # Relations
    owner: Mapped["User"] = relationship("User", back_populates="projects")
    members: Mapped[list["User"]] = relationship("User", secondary=project_members)
    tasks: Mapped[list["Task"]] = relationship(
        "Task", back_populates="project", cascade="all, delete-orphan"
    )
```

---

### 6.4 — Migrations avec Alembic

Alembic gère l'évolution du schéma de votre base de données dans le temps, comme Git pour votre DB.

```bash
# Initialiser Alembic
alembic init alembic
```

Modifier `alembic/env.py` :

```python
# alembic/env.py — sections à modifier

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

# IMPORTANT : Importer tous vos modèles ici pour qu'Alembic les détecte
from app.db.session import Base
from app.models import user, project, task  # noqa : importer pour les side effects
from app.core.config import settings

config = context.config
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)

target_metadata = Base.metadata  # Alembic compare ça avec la DB réelle
```

```bash
# Créer une migration automatique (Alembic détecte les changements)
alembic revision --autogenerate -m "initial tables"

# Appliquer les migrations
alembic upgrade head

# Voir l'historique des migrations
alembic history

# Annuler la dernière migration
alembic downgrade -1
```

---

### [OK] Bonnes Pratiques du Chapitre 6

> **Ne jamais utiliser `Base.metadata.create_all()` en production.** C'est pour les tests uniquement. En production, utilisez toujours Alembic.

> **Nommez les migrations de façon descriptive** : `add_user_avatar_field`, `create_task_table`, etc.

> **Toujours inclure `ondelete="CASCADE"` dans les ForeignKey** quand la suppression doit se propager.

---

### [EFFORT] Exercice de Complétion — Chapitre 6

```python
# app/models/task.py
# [NOTE] Créer le modèle ORM complet pour une tâche NexaHub avec :
# - id (PK)
# - title (String 200, non null)
# - description (Text, nullable)
# - status (String 50, défaut "todo")
# - priority (Integer, défaut 1)
# - due_date (DateTime, nullable)
# - project_id (FK vers projects.id, cascade delete)
# - assignee_id (FK vers users.id, nullable — tâche non assignée possible)
# - created_at, updated_at (timestamps automatiques)
# - Relations : project, assignee

from sqlalchemy import ___
from sqlalchemy.orm import ___
from app.db.session import Base

class Task(Base):
    __tablename__ = ___

    # [NOTE] Compléter tous les champs
    id: Mapped[___] = mapped_column(___, primary_key=True, index=True)
    title: Mapped[___] = mapped_column(___, nullable=False)
    # ... continuer

    # [NOTE] Définir les relations
    project: Mapped[___] = relationship(___, back_populates=___)
    assignee: Mapped[___] = relationship(___, foreign_keys=[___])
```

---

## [LIVRE] Chapitre 7 : CRUD Complet

### [OBJECTIF] Objectifs du chapitre
- Implémenter toutes les opérations CRUD avec SQLAlchemy async
- Structurer la logique dans des services séparés
- Gérer les erreurs avec HTTPException

---

### 7.1 — Pattern Service : Séparer la logique métier

Un **service** est une classe (ou un module) qui contient toute la logique d'accès aux données. Les routes ne font qu'appeler les services. Cela permet de :
- Tester la logique sans FastAPI
- Réutiliser la logique dans différentes routes
- Garder les routes fines et lisibles

```
Route (HTTP) -> Service (logique métier) -> DB (SQLAlchemy)
```

---

### 7.2 — Service Projet

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

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from typing import Optional, List
from app.models.project import Project
from app.schemas.project import ProjectCreate, ProjectUpdate
from fastapi import HTTPException, status

class ProjectService:

    @staticmethod
    async def get_all(
        db: AsyncSession,
        skip: int = 0,
        limit: int = 20,
        owner_id: Optional[int] = None
    ) -> List[Project]:
        """Récupère tous les projets avec pagination optionnelle."""
        query = select(Project).options(
            selectinload(Project.owner),    # Charge l'owner en même temps
            selectinload(Project.members)   # Charge les membres
        )

        if owner_id:
            query = query.where(Project.owner_id == owner_id)

        query = query.offset(skip).limit(limit)
        result = await db.execute(query)
        return result.scalars().all()

    @staticmethod
    async def get_by_id(db: AsyncSession, project_id: int) -> Project:
        """Récupère un projet par son ID. Lève 404 s'il n'existe pas."""
        query = select(Project).where(Project.id == project_id).options(
            selectinload(Project.owner),
            selectinload(Project.tasks)
        )
        result = await db.execute(query)
        project = result.scalar_one_or_none()

        if not project:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Projet avec l'ID {project_id} introuvable"
            )
        return project

    @staticmethod
    async def create(
        db: AsyncSession,
        project_data: ProjectCreate,
        owner_id: int
    ) -> Project:
        """Crée un nouveau projet."""
        # Vérifier si un projet avec ce nom existe déjà pour cet owner
        existing = await db.execute(
            select(Project).where(
                Project.name == project_data.name,
                Project.owner_id == owner_id
            )
        )
        if existing.scalar_one_or_none():
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Vous avez déjà un projet avec ce nom"
            )

        # Créer l'objet ORM
        new_project = Project(
            **project_data.model_dump(),  # Convertit le schéma Pydantic en dict
            owner_id=owner_id
        )

        db.add(new_project)
        await db.flush()  # Attribuer l'ID sans commit définitif
        await db.refresh(new_project)  # Recharger depuis la DB (pour avoir created_at, etc.)
        return new_project

    @staticmethod
    async def update(
        db: AsyncSession,
        project_id: int,
        update_data: ProjectUpdate,
        current_user_id: int
    ) -> Project:
        """Met à jour un projet. Seul l'owner peut modifier."""
        project = await ProjectService.get_by_id(db, project_id)

        # Vérification des permissions
        if project.owner_id != current_user_id:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Seul le propriétaire peut modifier ce projet"
            )

        # Mettre à jour uniquement les champs fournis (non-None)
        update_dict = update_data.model_dump(exclude_unset=True)
        for field, value in update_dict.items():
            setattr(project, field, value)

        await db.flush()
        await db.refresh(project)
        return project

    @staticmethod
    async def delete(
        db: AsyncSession,
        project_id: int,
        current_user_id: int
    ) -> None:
        """Supprime un projet. Seul l'owner peut supprimer."""
        project = await ProjectService.get_by_id(db, project_id)

        if project.owner_id != current_user_id:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Seul le propriétaire peut supprimer ce projet"
            )

        await db.delete(project)
        # Le commit est fait par get_db() dans session.py
```

---

### 7.3 — Routes CRUD utilisant le service

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

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from app.db.session import get_db
from app.services.project_service import ProjectService
from app.schemas.project import ProjectCreate, ProjectUpdate, ProjectResponse

router = APIRouter(prefix="/projects", tags=["projects"])

@router.get("/", response_model=List[ProjectResponse])
async def list_projects(
    skip: int = 0,
    limit: int = 20,
    db: AsyncSession = Depends(get_db)
    # current_user = Depends(get_current_user)  <- on ajoutera ça au Chap 11
):
    return await ProjectService.get_all(db, skip=skip, limit=limit)

@router.post("/", response_model=ProjectResponse, status_code=201)
async def create_project(
    project: ProjectCreate,
    db: AsyncSession = Depends(get_db)
):
    # Pour l'instant on met owner_id=1, on le remplacera avec l'auth (Chap 11)
    return await ProjectService.create(db, project, owner_id=1)

@router.get("/{project_id}", response_model=ProjectResponse)
async def get_project(
    project_id: int,
    db: AsyncSession = Depends(get_db)
):
    return await ProjectService.get_by_id(db, project_id)

@router.put("/{project_id}", response_model=ProjectResponse)
async def update_project(
    project_id: int,
    project: ProjectUpdate,
    db: AsyncSession = Depends(get_db)
):
    return await ProjectService.update(db, project_id, project, current_user_id=1)

@router.delete("/{project_id}", status_code=204)
async def delete_project(
    project_id: int,
    db: AsyncSession = Depends(get_db)
):
    await ProjectService.delete(db, project_id, current_user_id=1)
    # Pas de return — 204 No Content
```

---

### [OK] Bonnes Pratiques du Chapitre 7

> **Toujours retourner le bon code HTTP** : 201 pour une création, 204 pour une suppression réussie, 404 pour une ressource introuvable.

> **Utiliser `exclude_unset=True`** dans `.model_dump()` pour une mise à jour partielle (PATCH) : ne modifier que ce qui a été explicitement envoyé.

> **Toujours vérifier les permissions** dans le service, pas dans la route.

---

### [EFFORT] Exercice de Complétion — Chapitre 7

```python
# app/services/task_service.py
# [NOTE] Créer le TaskService complet avec les méthodes suivantes :

class TaskService:

    @staticmethod
    async def get_by_project(
        db: AsyncSession,
        project_id: int,
        status_filter: Optional[str] = None,   # Filtre optionnel
        skip: int = 0,
        limit: int = 50
    ) -> List[Task]:
        # [NOTE] Récupérer les tâches d'un projet
        # Si status_filter est fourni, filtrer par statut
        # Charger l'assignee avec selectinload
        pass

    @staticmethod
    async def assign_task(
        db: AsyncSession,
        task_id: int,
        user_id: int,           # Utilisateur à qui assigner
        current_user_id: int    # Celui qui fait l'action
    ) -> Task:
        # [NOTE] Assigner une tâche à un utilisateur
        # Vérifier que la tâche existe (404 sinon)
        # Vérifier que l'utilisateur à assigner est membre du projet
        # Mettre à jour assignee_id
        pass

    @staticmethod
    async def change_status(
        db: AsyncSession,
        task_id: int,
        new_status: str,
        current_user_id: int
    ) -> Task:
        # [NOTE] Changer le statut d'une tâche
        # Valider que new_status est dans ["todo", "in_progress", "done", "cancelled"]
        # Seul l'assignee ou l'owner du projet peut changer le statut
        pass
```

---

## [LIVRE] Chapitre 8 : Gestion des Erreurs & Exceptions

### [OBJECTIF] Objectifs du chapitre
- Gérer proprement toutes les erreurs de l'API
- Créer des handlers globaux personnalisés
- Ne jamais exposer les erreurs internes

---

### 8.1 — HTTPException : La base

```python
from fastapi import HTTPException, status

# Lever une erreur avec HTTPException
raise HTTPException(
    status_code=status.HTTP_404_NOT_FOUND,  # Utiliser les constantes !
    detail="Utilisateur introuvable",
    headers={"X-Error": "user_not_found"}   # Headers optionnels
)
```

Utilisez les **constantes de `status`** plutôt que les codes numériques bruts : c'est plus lisible et moins sujet aux fautes de frappe.

---

### 8.2 — Exceptions personnalisées

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

from fastapi import HTTPException, status
from typing import Any, Dict, Optional

class NexaHubException(HTTPException):
    """Exception de base pour NexaHub."""
    def __init__(
        self,
        status_code: int,
        message: str,
        error_code: str,            # Code interne pour les clients API
        details: Optional[Any] = None
    ):
        super().__init__(
            status_code=status_code,
            detail={
                "message": message,
                "error_code": error_code,
                "details": details
            }
        )

# Exceptions spécifiques
class ResourceNotFoundException(NexaHubException):
    def __init__(self, resource: str, resource_id: Any):
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            message=f"{resource} avec l'ID {resource_id} introuvable",
            error_code="RESOURCE_NOT_FOUND",
            details={"resource": resource, "id": resource_id}
        )

class PermissionDeniedException(NexaHubException):
    def __init__(self, action: str):
        super().__init__(
            status_code=status.HTTP_403_FORBIDDEN,
            message=f"Vous n'avez pas la permission d'effectuer : {action}",
            error_code="PERMISSION_DENIED",
            details={"action": action}
        )

class DuplicateResourceException(NexaHubException):
    def __init__(self, resource: str, field: str):
        super().__init__(
            status_code=status.HTTP_409_CONFLICT,
            message=f"Un(e) {resource} avec ce {field} existe déjà",
            error_code="DUPLICATE_RESOURCE",
            details={"resource": resource, "conflicting_field": field}
        )
```

---

### 8.3 — Handler global d'exceptions

```python
# app/main.py

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import IntegrityError
import logging

logger = logging.getLogger(__name__)

app = FastAPI()

# Handler pour les erreurs de validation Pydantic
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """
    Formater les erreurs de validation de façon lisible.
    Par défaut, FastAPI retourne un format un peu verbeux.
    """
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " -> ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

    logger.warning(f"Erreur de validation sur {request.url}: {errors}")

    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "VALIDATION_ERROR",
            "message": "Les données envoyées sont invalides",
            "errors": errors
        }
    )

# Handler pour les erreurs d'intégrité DB (unicité, FK, etc.)
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
    """Évite d'exposer les détails SQL internes."""
    logger.error(f"Erreur d'intégrité DB: {exc}")
    return JSONResponse(
        status_code=status.HTTP_409_CONFLICT,
        content={
            "error": "DATABASE_INTEGRITY_ERROR",
            "message": "Cette opération viole une contrainte de base de données"
        }
    )

# Handler global pour les erreurs non prévues
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Ne jamais exposer les détails d'une erreur interne."""
    logger.error(f"Erreur inattendue sur {request.url}: {exc}", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": "INTERNAL_SERVER_ERROR",
            "message": "Une erreur interne s'est produite. Notre équipe a été notifiée."
        }
    )
```

---

### [OK] Bonnes Pratiques du Chapitre 8

> **Ne jamais exposer les erreurs internes** (stack trace, détails SQL). Loggez en interne, répondez avec un message générique.

> **Créez un système de codes d'erreur internes** (`RESOURCE_NOT_FOUND`, `PERMISSION_DENIED`) pour que les clients front-end puissent réagir programmatiquement.

> **Loggez toujours les erreurs serveur** avec le niveau `ERROR` ou `CRITICAL` pour pouvoir les retrouver en production.

---

### [EFFORT] Exercice de Complétion — Chapitre 8

```python
# [NOTE] Créer une exception NexaHub pour chaque cas suivant
# et l'utiliser dans un handler global :

# 1. TokenExpiredException : le JWT est expiré (401)
class TokenExpiredException(NexaHubException):
    def __init__(self):
        super().__init__(
            status_code=___,
            message=___,
            error_code=___
        )

# 2. RateLimitExceededException : trop de requêtes (429)
class RateLimitExceededException(___):
    # [NOTE] À compléter
    pass

# 3. Dans main.py : ajouter un handler pour RequestValidationError
# qui retourne un message d'erreur en français
# avec la liste des champs invalides et les messages d'erreur
@app.exception_handler(___)
async def ___(request: Request, exc: ___):
    # [NOTE] Formater les erreurs Pydantic de façon lisible
    # Retourner un JSONResponse avec status 422
    pass
```

---

## [LIVRE] Chapitre 9 : Middleware & Hooks

### [OBJECTIF] Objectifs du chapitre
- Comprendre ce qu'est un middleware et son utilité
- Implémenter CORS, logging, timing et autres middlewares courants
- Créer un middleware personnalisé

---

### 9.1 — Qu'est-ce qu'un Middleware ?

Un **middleware** est une couche qui s'intercale entre la requête HTTP et votre handler de route. Il peut :
- Lire et modifier la requête entrante
- Lire et modifier la réponse sortante
- Court-circuiter le traitement (ex: rejeter une requête non authentifiée)
- Mesurer le temps de traitement

```
Requête -> [Middleware 1] -> [Middleware 2] -> Route Handler -> [Middleware 2] -> [Middleware 1] -> Réponse
           (CORS)           (Logging)                         (Logging)         (CORS)
```

Les middlewares s'exécutent en pile (LIFO pour la réponse).

---

### 9.2 — CORS Middleware

**CORS** (Cross-Origin Resource Sharing) est un mécanisme de sécurité du navigateur qui bloque les requêtes vers une API venant d'un domaine différent (par ex. votre frontend React sur `localhost:3000` qui appelle votre API sur `localhost:8000`).

```python
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,  # Origines autorisées (liste)
    allow_credentials=True,      # Autoriser les cookies et auth headers
    allow_methods=["*"],         # Méthodes HTTP autorisées (GET, POST, etc.)
    allow_headers=["*"],         # Headers autorisés
)

# [ATTENTION] En production, ne JAMAIS utiliser allow_origins=["*"]
# Spécifiez exactement les domaines autorisés :
# allow_origins=["https://app.nexahub.io", "https://admin.nexahub.io"]
```

---

### 9.3 — Middleware de Logging et Timing

```python
# app/main.py

import time
import uuid
import logging
from fastapi import Request

logger = logging.getLogger("nexahub.access")

@app.middleware("http")
async def logging_and_timing_middleware(request: Request, call_next):
    """
    Middleware custom qui :
    1. Assigne un ID unique à chaque requête (pour le traçage)
    2. Mesure le temps de traitement
    3. Logge toutes les requêtes avec leur statut et durée
    """
    # Générer un ID de requête unique
    request_id = str(uuid.uuid4())[:8]

    # Stocker dans l'état de la requête pour y accéder depuis les routes
    request.state.request_id = request_id

    start_time = time.time()

    # Appeler le prochain handler (route ou middleware suivant)
    response = await call_next(request)

    # Calculer la durée après la réponse
    process_time = (time.time() - start_time) * 1000  # en millisecondes

    # Ajouter des headers de debug dans la réponse
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{process_time:.2f}ms"

    # Logger la requête
    logger.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"-> {response.status_code} ({process_time:.2f}ms)"
    )

    return response
```

---

### 9.4 — Events Startup et Shutdown

```python
# app/main.py

from contextlib import asynccontextmanager
from app.db.session import engine, Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    Code exécuté au démarrage et à l'arrêt de l'application.
    C'est la méthode moderne (remplace @app.on_event("startup")).
    """
    # ── STARTUP ──
    print("[RAPIDE] NexaHub démarre...")

    # Créer les tables (seulement pour le développement)
    # En production : utiliser alembic upgrade head
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    print("[OK] Base de données initialisée")
    print(f"[GUIDE] Documentation : http://localhost:8000/docs")

    yield  # L'application tourne ici

    # ── SHUTDOWN ──
    print("[ROUGE] NexaHub s'arrête...")
    await engine.dispose()  # Fermer toutes les connexions DB
    print("[OK] Connexions fermées proprement")

# Passer lifespan à l'app
app = FastAPI(lifespan=lifespan, title="NexaHub API")
```

---

### [OK] Bonnes Pratiques du Chapitre 9

> **Limitez le nombre de middlewares** : chaque middleware ajoute une latence. N'ajoutez que ceux vraiment nécessaires.

> **Toujours utiliser `lifespan` (contextmanager)** plutôt que `@app.on_event("startup")` — cette dernière est dépréciée.

> **En production, configurez CORS de façon stricte** : listez exactement les domaines autorisés. `allow_origins=["*"]` est un risque de sécurité.

---

### [EFFORT] Exercice de Complétion — Chapitre 9

```python
# [NOTE] Créer un middleware de sécurité pour NexaHub
# Ce middleware doit :
# 1. Bloquer les requêtes dont le User-Agent est vide (bots malveillants basiques)
# 2. Ajouter des headers de sécurité à chaque réponse :
#    - X-Content-Type-Options: nosniff
#    - X-Frame-Options: DENY
#    - X-XSS-Protection: 1; mode=block

@app.middleware("http")
async def security_headers_middleware(request: Request, call_next):
    # [NOTE] Vérifier que User-Agent est présent
    user_agent = request.headers.get(___)
    if not user_agent:
        # [NOTE] Retourner directement une JSONResponse 400
        # sans appeler call_next
        return ___

    # [NOTE] Appeler le handler normal
    response = await ___

    # [NOTE] Ajouter les 3 headers de sécurité à la réponse
    response.headers[___] = ___
    response.headers[___] = ___
    response.headers[___] = ___

    return response

# [NOTE] Créer un middleware de rate limiting BASIQUE (sans Redis pour l'instant)
# Utiliser un dictionnaire en mémoire pour compter les requêtes par IP
# Si une IP dépasse 100 requêtes par minute, retourner 429

from collections import defaultdict
from datetime import datetime, timedelta

request_counts: dict = defaultdict(list)  # IP -> liste de timestamps

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # [NOTE] Extraire l'IP du client
    client_ip = request.client.host

    # [NOTE] Nettoyer les entrées de plus d'1 minute
    # [NOTE] Compter les requêtes récentes
    # [NOTE] Si > 100, retourner 429
    # [NOTE] Sinon, ajouter le timestamp actuel et continuer

    pass
```

---

## [OBJECTIF] Récapitulatif — Partie 2

À ce stade, NexaHub a :

| Composant | Statut |
|---|---|
| Structure modulaire avec APIRouter | [OK] En place |
| Configuration centralisée (.env) | [OK] En place |
| Base de données connectée (async) | [OK] En place |
| Modèles ORM (User, Project, Task) | [OK] En place |
| Migrations Alembic | [OK] En place |
| CRUD complet avec services | [OK] En place |
| Gestion d'erreurs globale | [OK] En place |
| Middlewares (CORS, logging, timing) | [OK] En place |

### Ce qu'on a construit pour NexaHub :
Une vraie API structurée professionnellement, connectée à une base de données, avec gestion des erreurs robuste et des middlewares essentiels.

### Prochain chapitre :
La **Partie 3** couvre la sécurité : authentification JWT, hashing de mots de passe, rôles et permissions.

---

*[BOOKMARK] Fichier suivant : `fastapi_nexahub_partie3.md` — Sécurité & Authentification*

# [RAPIDE] NexaHub — Partie 3 : Sécurité & Authentification
## *JWT, Permissions, CORS, Protection*

---

# [VERROUILLE] PARTIE 3 — Sécurité & Authentification

---

## [LIVRE] Chapitre 10 : Authentification Basique avec JWT

### [OBJECTIF] Objectifs du chapitre
- Comprendre le fonctionnement de JWT
- Implémenter l'inscription et la connexion
- Hasher les mots de passe correctement
- Protéger les premières routes

---

### 10.1 — Comment fonctionne l'authentification JWT ?

**JWT** (JSON Web Token) est un standard ouvert pour transmettre des informations de façon sécurisée sous forme de token signé.

```
1. L'utilisateur envoie ses identifiants :
   POST /auth/login { "email": "alice@nexahub.io", "password": "secret" }

2. Le serveur vérifie les identifiants et génère un token :
   "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNj..."

3. L'utilisateur stocke ce token et l'envoie dans chaque requête :
   Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

4. Le serveur vérifie le token, extrait l'ID utilisateur, et répond
```

#### Structure d'un JWT

Un JWT est composé de 3 parties séparées par des points `.` :

```
HEADER.PAYLOAD.SIGNATURE

eyJhbGciOiJIUzI1NiJ9   <- Header (algo, type) — encodé en Base64
.eyJzdWIiOiIxIiwiZXhwIjoxNjk5fQ   <- Payload (données) — encodé en Base64
.SflKxwRJSMeKKF2QT4fwpMeJf36POk6   <- Signature — HMAC-SHA256 du header+payload
```

**Le payload peut contenir** :
- `sub` : subject (ID de l'utilisateur)
- `exp` : expiration (timestamp Unix)
- `iat` : issued at (création)
- `type` : "access" ou "refresh"

> **Important** : Le payload JWT est **encodé** (Base64), pas **chiffré**. N'y mettez jamais d'informations sensibles (mot de passe, numéro de carte...). N'importe qui peut le décoder.

---

### 10.2 — Installation des dépendances

```bash
pip install python-jose[cryptography] passlib[bcrypt] python-multipart
# python-jose : génération et vérification de JWT
# passlib[bcrypt] : hashing de mots de passe
# python-multipart : nécessaire pour OAuth2PasswordRequestForm
```

---

### 10.3 — Module de sécurité

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

from datetime import datetime, timedelta, timezone
from typing import Optional, Literal
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

# ─────────────────────────────────────────────────────────────
# HASHING DES MOTS DE PASSE
# ─────────────────────────────────────────────────────────────

# bcrypt est l'algorithme de référence pour les mots de passe
# Il est intentionnellement LENT pour résister aux attaques brute-force
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hash un mot de passe en clair avec bcrypt."""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """
    Vérifie si un mot de passe en clair correspond au hash.
    Utilise une comparaison en temps constant pour éviter
    les timing attacks.
    """
    return pwd_context.verify(plain_password, hashed_password)


# ─────────────────────────────────────────────────────────────
# GESTION DES TOKENS JWT
# ─────────────────────────────────────────────────────────────

def create_token(
    subject: str,                           # Généralement l'ID ou l'email de l'user
    token_type: Literal["access", "refresh"] = "access",
    expires_delta: Optional[timedelta] = None
) -> str:
    """
    Crée un JWT signé.

    Le token contient :
    - sub : identifiant de l'utilisateur
    - type : "access" ou "refresh"
    - exp : timestamp d'expiration
    - iat : timestamp de création
    """
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    elif token_type == "access":
        expire = datetime.now(timezone.utc) + timedelta(
            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
        )
    else:
        expire = datetime.now(timezone.utc) + timedelta(
            days=settings.REFRESH_TOKEN_EXPIRE_DAYS
        )

    payload = {
        "sub": str(subject),
        "type": token_type,
        "exp": expire,
        "iat": datetime.now(timezone.utc)
    }

    return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)


def decode_token(token: str) -> dict:
    """
    Décode et vérifie un JWT.

    Lève JWTError si :
    - Le token est malformé
    - La signature est invalide
    - Le token est expiré

    Retourne le payload si tout est valide.
    """
    try:
        payload = jwt.decode(
            token,
            settings.SECRET_KEY,
            algorithms=[settings.ALGORITHM]
        )
        return payload
    except JWTError as e:
        raise JWTError(f"Token invalide : {e}")
```

---

### 10.4 — Schémas d'authentification

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

from pydantic import BaseModel, EmailStr, Field

class LoginRequest(BaseModel):
    email: str = Field(..., description="Email ou username")
    password: str = Field(..., min_length=1)

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"
    expires_in: int   # Durée en secondes

class RefreshRequest(BaseModel):
    refresh_token: str

class RegisterRequest(BaseModel):
    username: str = Field(..., min_length=3, max_length=30)
    email: str
    password: str = Field(..., min_length=8, max_length=100)
    full_name: str = Field(..., min_length=2, max_length=100)

    class Config:
        json_schema_extra = {
            "example": {
                "username": "alice",
                "email": "alice@nexahub.io",
                "password": "MonMotDePasse123!",
                "full_name": "Alice Dupont"
            }
        }
```

---

### 10.5 — Routes d'authentification

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

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from datetime import timedelta
from app.db.session import get_db
from app.models.user import User
from app.schemas.auth import LoginRequest, TokenResponse, RegisterRequest
from app.core.security import (
    hash_password, verify_password, create_token
)
from app.core.config import settings

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register", status_code=201)
async def register(data: RegisterRequest, db: AsyncSession = Depends(get_db)):
    """Créer un nouveau compte NexaHub."""

    # Vérifier que l'email n'est pas déjà utilisé
    existing_email = await db.execute(
        select(User).where(User.email == data.email.lower())
    )
    if existing_email.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Cet email est déjà utilisé"
        )

    # Vérifier que le username n'est pas déjà pris
    existing_username = await db.execute(
        select(User).where(User.username == data.username.lower())
    )
    if existing_username.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Ce nom d'utilisateur est déjà pris"
        )

    # Créer le nouvel utilisateur
    new_user = User(
        username=data.username.lower(),
        email=data.email.lower(),
        hashed_password=hash_password(data.password),  # <- JAMAIS stocker en clair
        full_name=data.full_name,
    )
    db.add(new_user)
    await db.flush()
    await db.refresh(new_user)

    return {
        "message": "Compte créé avec succès ! Vous pouvez maintenant vous connecter.",
        "user_id": new_user.id,
        "username": new_user.username
    }


@router.post("/login", response_model=TokenResponse)
async def login(data: LoginRequest, db: AsyncSession = Depends(get_db)):
    """
    Se connecter et obtenir des tokens JWT.

    Accepte email ou username dans le champ 'email'.
    """

    # Chercher l'utilisateur par email ou username
    result = await db.execute(
        select(User).where(
            (User.email == data.email.lower()) |
            (User.username == data.email.lower())
        )
    )
    user = result.scalar_one_or_none()

    # [ATTENTION] Message d'erreur VOLONTAIREMENT vague pour éviter l'énumération
    # Ne jamais dire "email correct mais mauvais mot de passe"
    if not user or not verify_password(data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Identifiants incorrects",
            headers={"WWW-Authenticate": "Bearer"}
        )

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Votre compte a été désactivé"
        )

    # Générer les tokens
    access_token = create_token(user.id, token_type="access")
    refresh_token = create_token(user.id, token_type="refresh")

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token,
        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )
```

---

### [OK] Bonnes Pratiques du Chapitre 10

> **Ne jamais stocker un mot de passe en clair.** Toujours utiliser bcrypt ou argon2.

> **Message d'erreur d'authentification volontairement vague** : ne dites pas "email correct mais mauvais mot de passe". Dites toujours "Identifiants incorrects". Sinon vous permettez l'énumération d'emails.

> **Gardez la SECRET_KEY longue et aléatoire** : au moins 32 caractères. Utilisez `openssl rand -hex 32` pour en générer une.

---

### [EFFORT] Exercice de Complétion — Chapitre 10

```python
# [NOTE] Implémenter la route /auth/refresh
# Elle doit :
# 1. Accepter un refresh_token dans le body
# 2. Décoder le token et vérifier que son type est "refresh"
# 3. Récupérer l'utilisateur depuis la DB
# 4. Générer et retourner un nouveau access_token

@router.post("/refresh", response_model=___)
async def refresh_token(data: RefreshRequest, db: AsyncSession = Depends(get_db)):
    from jose import JWTError
    from app.core.security import decode_token

    # [NOTE] Décoder le token
    try:
        payload = decode_token(___)
    except JWTError:
        raise HTTPException(status_code=___, detail=___)

    # [NOTE] Vérifier que c'est bien un refresh token
    if payload.get("type") != ___:
        raise HTTPException(status_code=___, detail="Token invalide")

    # [NOTE] Récupérer l'utilisateur
    user_id = payload.get("sub")
    user = await ___

    # [NOTE] Retourner un nouveau access_token
    new_token = create_token(___, token_type=___)
    return {"access_token": new_token, "token_type": "bearer"}

# [NOTE] Bonus : implémenter /auth/logout
# Dans une vraie app, on invalide le refresh token (via une blacklist Redis)
# Pour l'instant, retourner simplement {"message": "Déconnecté avec succès"}
```

---

## [LIVRE] Chapitre 11 : Authentification Avancée & Permissions

### [OBJECTIF] Objectifs du chapitre
- Créer la dépendance `get_current_user` réutilisable
- Implémenter un système de rôles et de permissions
- Protéger toutes les routes qui le nécessitent

---

### 11.1 — La Dependency Injection de FastAPI

**Dependency Injection** (DI) est le mécanisme le plus puissant de FastAPI. Il permet de déclarer des dépendances qui sont automatiquement résolues et injectées dans vos routes.

```python
# Sans DI : copier-coller dans chaque route [WEARY_FACE]
@router.get("/projects")
async def list_projects(request: Request):
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    if not token:
        raise HTTPException(401, "Non authentifié")
    payload = decode_token(token)
    user = await get_user_by_id(payload["sub"])
    # ... enfin la vraie logique

# Avec DI : propre et réutilisable [OK]
@router.get("/projects")
async def list_projects(current_user: User = Depends(get_current_user)):
    # current_user est automatiquement résolu par FastAPI
    # ... directement la vraie logique
```

---

### 11.2 — Dépendance get_current_user

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

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from jose import JWTError
from app.db.session import get_db
from app.models.user import User
from app.core.security import decode_token

# Extracteur du token Bearer depuis le header Authorization
http_bearer = HTTPBearer(auto_error=False)


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(http_bearer),
    db: AsyncSession = Depends(get_db)
) -> User:
    """
    Dépendance principale d'authentification.

    1. Extrait le token Bearer du header Authorization
    2. Décode et vérifie le JWT
    3. Récupère l'utilisateur depuis la DB
    4. Vérifie que le compte est actif

    Utilisation :
        current_user: User = Depends(get_current_user)
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Non authentifié ou token invalide",
        headers={"WWW-Authenticate": "Bearer"},
    )

    # Vérifier que le header Authorization existe
    if not credentials:
        raise credentials_exception

    # Décoder le token
    try:
        payload = decode_token(credentials.credentials)
        user_id = payload.get("sub")
        token_type = payload.get("type")

        if user_id is None:
            raise credentials_exception

        # Vérifier que c'est un access token (pas un refresh token)
        if token_type != "access":
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Utilisez un access token, pas un refresh token"
            )

    except JWTError:
        raise credentials_exception

    # Récupérer l'utilisateur depuis la DB
    result = await db.execute(select(User).where(User.id == int(user_id)))
    user = result.scalar_one_or_none()

    if user is None:
        raise credentials_exception

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Compte désactivé"
        )

    return user


# ─────────────────────────────────────────────────────────────
# DÉPENDANCES DE RÔLES
# ─────────────────────────────────────────────────────────────

def require_superuser(current_user: User = Depends(get_current_user)) -> User:
    """Dépendance qui vérifie que l'utilisateur est admin/superuser."""
    if not current_user.is_superuser:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cette action nécessite des droits administrateur"
        )
    return current_user


# ─────────────────────────────────────────────────────────────
# DÉPENDANCE DE PAGINATION
# ─────────────────────────────────────────────────────────────

class PaginationParams:
    """
    Dépendance réutilisable pour la pagination.

    Utilisation :
        async def list_items(pagination: PaginationParams = Depends()):
            items = await service.get_all(skip=pagination.skip, limit=pagination.limit)
    """
    def __init__(
        self,
        page: int = 1,              # Numéro de page (commence à 1)
        per_page: int = 20          # Éléments par page
    ):
        if page < 1:
            raise HTTPException(status_code=400, detail="page doit être >= 1")
        if per_page < 1 or per_page > 100:
            raise HTTPException(status_code=400, detail="per_page doit être entre 1 et 100")

        self.skip = (page - 1) * per_page
        self.limit = per_page
        self.page = page
        self.per_page = per_page
```

---

### 11.3 — Protéger les routes

```python
# app/api/routes/projects.py — mise à jour

from app.api.dependencies import get_current_user, PaginationParams
from app.models.user import User

@router.get("/", response_model=List[ProjectResponse])
async def list_projects(
    pagination: PaginationParams = Depends(),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)  # <- PROTECTION
):
    """Liste les projets de l'utilisateur connecté."""
    return await ProjectService.get_all(
        db,
        skip=pagination.skip,
        limit=pagination.limit,
        owner_id=current_user.id  # <- Filtrer par l'utilisateur courant
    )

@router.post("/", response_model=ProjectResponse, status_code=201)
async def create_project(
    project: ProjectCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)  # <- PROTECTION
):
    """Crée un projet pour l'utilisateur connecté."""
    return await ProjectService.create(db, project, owner_id=current_user.id)
```

---

### 11.4 — Système de Permissions basé sur le projet

Pour NexaHub, les permissions sont basées sur la relation avec le projet :

```python
# app/api/dependencies.py — ajouter

async def get_project_or_403(
    project_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
) -> "Project":
    """
    Récupère un projet ET vérifie que l'utilisateur y a accès.
    L'utilisateur doit être propriétaire ou membre du projet.
    """
    from app.models.project import Project
    from sqlalchemy.orm import selectinload

    result = await db.execute(
        select(Project)
        .options(selectinload(Project.members))
        .where(Project.id == project_id)
    )
    project = result.scalar_one_or_none()

    if not project:
        raise HTTPException(status_code=404, detail="Projet introuvable")

    # Vérifier l'accès : owner OU membre OU projet public
    is_owner = project.owner_id == current_user.id
    is_member = any(m.id == current_user.id for m in project.members)
    is_public = project.is_public

    if not (is_owner or is_member or is_public):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Vous n'avez pas accès à ce projet"
        )

    return project


async def require_project_owner(
    project: "Project" = Depends(get_project_or_403),
    current_user: User = Depends(get_current_user)
) -> "Project":
    """
    Vérifie que l'utilisateur est propriétaire du projet.
    À utiliser pour les actions sensibles (supprimer, modifier les paramètres).
    """
    if project.owner_id != current_user.id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Seul le propriétaire peut effectuer cette action"
        )
    return project
```

---

### [OK] Bonnes Pratiques du Chapitre 11

> **Centralisez toutes les dépendances dans `dependencies.py`**. N'écrivez jamais la logique de vérification JWT directement dans une route.

> **Composez les dépendances** : `require_project_owner` dépend de `get_project_or_403` qui dépend de `get_current_user`. FastAPI résout automatiquement toute la chaîne.

> **Séparez clairement "authentifié" et "autorisé"** : "authentifié" = je sais qui vous êtes ; "autorisé" = vous avez le droit de faire cette action.

---

### [EFFORT] Exercice de Complétion — Chapitre 11

```python
# [NOTE] Créer un décorateur de permission flexible pour NexaHub

from enum import Enum
from typing import Callable

class ProjectRole(str, Enum):
    OWNER = "owner"
    MEMBER = "member"
    VIEWER = "viewer"    # Peut seulement lire

def require_project_role(required_role: ProjectRole):
    """
    Factory de dépendance pour les permissions par rôle.

    Utilisation :
        @router.delete("/{project_id}")
        async def delete_project(
            project = Depends(require_project_role(ProjectRole.OWNER))
        ):
    """
    async def _check_role(
        project_id: int,
        db: AsyncSession = Depends(get_db),
        current_user: User = Depends(get_current_user)
    ):
        # [NOTE] Récupérer le projet
        project = await ___

        # [NOTE] Déterminer le rôle de l'utilisateur dans ce projet
        # OWNER : project.owner_id == current_user.id
        # MEMBER : current_user.id dans project.members
        # VIEWER : projet public ou membre

        user_role = ___  # À calculer

        # [NOTE] Hiérarchie des rôles : OWNER > MEMBER > VIEWER
        # Un OWNER peut tout faire, un MEMBER peut faire les actions MEMBER et VIEWER
        role_hierarchy = {
            ProjectRole.VIEWER: 0,
            ProjectRole.MEMBER: 1,
            ProjectRole.OWNER: 2
        }

        if role_hierarchy.get(user_role, -1) < role_hierarchy[required_role]:
            raise HTTPException(
                status_code=403,
                detail=f"Rôle requis : {required_role.value}"
            )

        return project

    return _check_role

# [NOTE] Utiliser require_project_role dans ces routes :
# - GET /projects/{id}/tasks -> VIEWER minimum
# - POST /projects/{id}/tasks -> MEMBER minimum
# - DELETE /projects/{id} -> OWNER uniquement
# - POST /projects/{id}/invite -> OWNER uniquement
```

---

## [LIVRE] Chapitre 12 : Sécurité Applicative

### [OBJECTIF] Objectifs du chapitre
- Comprendre et configurer CORS correctement
- Protéger contre les attaques courantes (XSS, injection)
- Valider strictement toutes les entrées

---

### 12.1 — CORS en production

Nous avons vu CORS au chapitre 9. Voici une configuration production complète :

```python
# app/main.py

from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings

# Configuration CORS différenciée selon l'environnement
if settings.ENVIRONMENT == "development":
    origins = ["http://localhost:3000", "http://localhost:5173", "http://127.0.0.1:3000"]
else:
    # En production : UNIQUEMENT vos domaines connus
    origins = settings.ALLOWED_ORIGINS  # Défini dans .env

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,      # Nécessaire pour les cookies de session
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=[
        "Authorization",
        "Content-Type",
        "X-Requested-With",
        "Accept",
        "Origin",
    ],
    max_age=600,  # Durée de cache de la réponse preflight (en secondes)
)
```

---

### 12.2 — Protection contre les injections avec Pydantic

Pydantic est votre première ligne de défense. Si vous définissez des types stricts, les tentatives d'injection sont automatiquement rejetées.

```python
from pydantic import BaseModel, Field, validator
import re

class TaskCreate(BaseModel):
    # Le regex rejette les caractères dangereux pour la DB et le HTML
    title: str = Field(
        ...,
        min_length=3,
        max_length=200,
    )
    description: str | None = Field(None, max_length=5000)

    @validator("title")
    def sanitize_title(cls, v):
        # Supprimer les balises HTML potentielles
        clean = re.sub(r"<[^>]*>", "", v)
        # Supprimer les caractères de contrôle
        clean = re.sub(r"[\x00-\x1F\x7F]", "", clean)
        # Vérifier qu'il reste du contenu
        if not clean.strip():
            raise ValueError("Le titre ne peut pas être vide après nettoyage")
        return clean.strip()

    @validator("description")
    def sanitize_description(cls, v):
        if v is None:
            return v
        # Même nettoyage pour la description
        clean = re.sub(r"<script[^>]*>.*?</script>", "", v, flags=re.DOTALL | re.IGNORECASE)
        return clean
```

---

### 12.3 — Headers de sécurité complets

```python
# app/main.py — Middleware de sécurité complet

from fastapi import Request
from fastapi.responses import Response

@app.middleware("http")
async def security_headers_middleware(request: Request, call_next):
    response: Response = await call_next(request)

    # Empêche le browser de "deviner" le Content-Type
    response.headers["X-Content-Type-Options"] = "nosniff"

    # Empêche l'intégration dans des iframes (clickjacking)
    response.headers["X-Frame-Options"] = "DENY"

    # Active le filtre XSS du browser (legacy mais utile)
    response.headers["X-XSS-Protection"] = "1; mode=block"

    # Force HTTPS pour 1 an (HSTS)
    if settings.ENVIRONMENT == "production":
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"

    # Content Security Policy : contrôle les ressources autorisées
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "script-src 'self' 'unsafe-inline'; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "font-src 'self';"
    )

    # Referrer Policy : contrôle les informations de référent
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

    # Permissions Policy : désactive les APIs sensibles du navigateur
    response.headers["Permissions-Policy"] = (
        "camera=(), microphone=(), geolocation=(), payment=()"
    )

    return response
```

---

### 12.4 — Validation des entrées fichiers

```python
# Nous verrons les uploads en détail au Chapitre 15
# Mais voici les validations de sécurité critiques

from fastapi import UploadFile, HTTPException
import magic  # pip install python-magic

ALLOWED_MIME_TYPES = {
    "image/jpeg", "image/png", "image/gif", "image/webp",
    "application/pdf",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
}
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

async def validate_upload(file: UploadFile) -> bytes:
    """
    Valide un fichier uploadé.
    NE PAS faire confiance au Content-Type envoyé par le client !
    """
    # Lire le contenu
    contents = await file.read()

    # Vérifier la taille
    if len(contents) > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,
            detail=f"Fichier trop volumineux (max {MAX_FILE_SIZE // 1024 // 1024}MB)"
        )

    # Vérifier le VRAI type MIME (pas celui déclaré par le client)
    # python-magic lit les magic bytes du fichier
    real_mime_type = magic.from_buffer(contents, mime=True)
    if real_mime_type not in ALLOWED_MIME_TYPES:
        raise HTTPException(
            status_code=415,
            detail=f"Type de fichier non autorisé : {real_mime_type}"
        )

    return contents
```

---

### 12.5 — Limiter le débit (Rate Limiting) avec Redis

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

import redis.asyncio as redis
from fastapi import Request, HTTPException, status
from app.core.config import settings

redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True)

async def rate_limiter(
    request: Request,
    max_requests: int = 100,
    window_seconds: int = 60
) -> None:
    """
    Rate limiter basé sur Redis avec algorithme sliding window.

    Utilisation :
        @router.post("/auth/login")
        async def login(_: None = Depends(lambda: rate_limiter(request, max_requests=5))):
    """
    client_ip = request.client.host
    key = f"rate_limit:{client_ip}:{request.url.path}"

    # Incrémenter le compteur
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, window_seconds)
    results = await pipe.execute()

    request_count = results[0]

    if request_count > max_requests:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=f"Trop de requêtes. Réessayez dans {window_seconds} secondes.",
            headers={"Retry-After": str(window_seconds)}
        )
```

---

### [OK] Bonnes Pratiques du Chapitre 12

> **Ne faites jamais confiance au Content-Type déclaré par le client** pour les uploads. Utilisez les magic bytes pour détecter le vrai type.

> **Configurez CORS avec des origines explicites** en production. `allow_origins=["*"]` est interdit.

> **Validez toutes les entrées** côté serveur, même si le frontend valide déjà. Le frontend peut être contourné.

---

### [EFFORT] Exercice de Complétion — Chapitre 12

```python
# [NOTE] Implémenter un système complet de sécurité pour NexaHub

# 1. Créer un validateur Pydantic qui vérifie la force d'un mot de passe
from pydantic import validator

class PasswordMixin:
    """Mixin de validation de mot de passe. Héritez-en dans vos schémas."""

    @validator("password")
    def password_must_be_strong(cls, v):
        # [NOTE] Vérifier que le mot de passe :
        # - Contient au moins 1 majuscule
        # - Contient au moins 1 minuscule
        # - Contient au moins 1 chiffre
        # - Contient au moins 1 caractère spécial (@$!%*?&)
        # - Fait au moins 8 caractères
        # Utiliser re.search() pour chaque condition
        import re

        if not re.search(___, v):  # majuscule
            raise ValueError(___)
        if not re.search(___, v):  # minuscule
            raise ValueError(___)
        # ... compléter

        return v

# 2. Implémenter la blacklist de tokens JWT avec Redis
# Quand un utilisateur se déconnecte, son access_token est ajouté à la blacklist
# La dépendance get_current_user doit vérifier la blacklist avant de valider

async def is_token_blacklisted(token: str) -> bool:
    # [NOTE] Vérifier dans Redis si le token est blacklisté
    # Clé Redis : "blacklist:{token}"
    pass

async def blacklist_token(token: str, expires_in: int) -> None:
    # [NOTE] Ajouter le token à la blacklist avec une expiration
    # (inutile de garder le token après son expiration naturelle)
    pass

# 3. Route /auth/logout
@router.post("/logout")
async def logout(
    credentials: HTTPAuthorizationCredentials = Depends(http_bearer),
    current_user: User = Depends(get_current_user)
):
    # [NOTE] Blacklister le token actuel
    # [NOTE] Retourner {"message": "Déconnexion réussie"}
    pass
```

---

## [OBJECTIF] Récapitulatif — Partie 3

À ce stade, NexaHub est **sécurisé** :

| Composant | Statut |
|---|---|
| Inscription/Connexion avec JWT | [OK] En place |
| Hashing bcrypt des mots de passe | [OK] En place |
| Dépendance `get_current_user` | [OK] En place |
| Système de permissions par projet | [OK] En place |
| Headers de sécurité HTTP | [OK] En place |
| CORS configuré | [OK] En place |
| Validation stricte des entrées | [OK] En place |
| Rate limiting basique | [OK] En place |

### Ce qu'on a construit pour NexaHub :
Un système d'authentification complet de niveau production : inscription, connexion, tokens JWT, refresh tokens, permissions granulaires par projet.

### Prochain chapitre :
La **Partie 4** couvre l'asynchronisme et l'optimisation des performances.

---

*[BOOKMARK] Fichier suivant : `fastapi_nexahub_partie4.md` — Performance, Communication & Intégrations*

# [RAPIDE] NexaHub — Partie 4 : Performance & Communication
## *Async, Cache, WebSockets, Fichiers, Emails*

---

# [RAPIDE] PARTIE 4 — Performance & Asynchronisme

---

## [LIVRE] Chapitre 13 : Programmation Asynchrone

### [OBJECTIF] Objectifs du chapitre
- Comprendre en profondeur la différence sync vs async
- Savoir quand utiliser async et quand ne pas l'utiliser
- Maîtriser BackgroundTasks pour les tâches non-bloquantes

---

### 13.1 — Sync vs Async : Le modèle mental

Imaginez un restaurant :

**Mode synchrone (sync)** : Un seul serveur. Il prend la commande de la table 1, va en cuisine, attend que le plat soit prêt, revient servir la table 1, puis s'occupe de la table 2. Les tables 2, 3, 4 attendent... même si leur commande n'est pas encore prête.

**Mode asynchrone (async)** : Un seul serveur intelligent. Il prend la commande de la table 1, la donne en cuisine, prend la commande de la table 2 pendant que la table 1 attend, donne en cuisine, prend la commande de la table 3... Quand un plat est prêt, il va le servir. Il fait tout en parallèle *sans jamais vraiment attendre*.

C'est exactement ce qu'async/await fait dans votre code :

```python
import asyncio
import httpx

# VERSION SYNCHRONE : chaque requête bloque la suivante
def get_user_data_sync(user_id: int) -> dict:
    # Appel DB : on ATTEND la réponse (0-50ms bloqués)
    user = db.query(User).get(user_id)
    # Appel API externe : on ATTEND encore (100-500ms bloqués)
    profile = requests.get(f"https://api.github.com/user/{user_id}")
    return {"user": user, "github": profile.json()}


# VERSION ASYNCHRONE : on n'attend pas, on fait autre chose pendant
async def get_user_data_async(user_id: int) -> dict:
    # Ces deux opérations s'exécutent SIMULTANÉMENT
    user, github_profile = await asyncio.gather(
        db_get_user(user_id),                     # Async DB query
        httpx.AsyncClient().get(f"...{user_id}")  # Async HTTP call
    )
    # asyncio.gather exécute les deux en parallèle !
    return {"user": user, "github": github_profile.json()}
```

---

### 13.2 — Règle d'or : Quand utiliser async ?

```
Si votre fonction fait une opération I/O -> utilisez async def
    - Requête DB (SQLAlchemy async, TortoiseORM, Motor)
    - Requête HTTP externe (httpx, aiohttp)
    - Lecture/écriture de fichier (aiofiles)
    - Redis, cache, message queue

Si votre fonction ne fait que du calcul (CPU-bound) -> utilisez def
    - Traitement d'images
    - Calculs mathématiques
    - Parsing de documents
    - Compression
```

> [ATTENTION] **Piège courant** : appeler une fonction `sync` bloquante depuis un contexte `async` bloque TOUT le serveur FastAPI ! Utilisez `run_in_executor` pour les fonctions CPU-bound.

```python
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def process_image_endpoint(file_path: str):
    # [X] MAUVAIS : bloque le serveur pendant le traitement d'image
    # result = blocking_image_processing(file_path)

    # [OK] BON : exécute la fonction sync dans un thread séparé
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        executor,
        blocking_image_processing,  # Fonction sync
        file_path                   # Ses arguments
    )
    return result
```

---

### 13.3 — BackgroundTasks : Tâches en arrière-plan légères

`BackgroundTasks` permet d'exécuter du code **après** avoir envoyé la réponse HTTP. Idéal pour des tâches courtes qui ne doivent pas bloquer la réponse :

```python
from fastapi import BackgroundTasks, Depends
from app.services.email_service import send_welcome_email
from app.services.analytics_service import track_event

@router.post("/auth/register", status_code=201)
async def register(
    data: RegisterRequest,
    background_tasks: BackgroundTasks,  # <- Injecté automatiquement
    db: AsyncSession = Depends(get_db)
):
    # Créer l'utilisateur (opération principale)
    new_user = await create_user(db, data)

    # Ajouter des tâches en arrière-plan
    # Ces tâches s'exécutent APRÈS que la réponse 201 a été envoyée
    background_tasks.add_task(
        send_welcome_email,          # Fonction à appeler
        user_email=new_user.email,   # Arguments nommés
        username=new_user.username
    )
    background_tasks.add_task(
        track_event,
        event="user_registered",
        user_id=new_user.id
    )

    # Cette réponse est envoyée IMMÉDIATEMENT
    # L'email et le tracking se font après
    return {"message": "Compte créé !", "user_id": new_user.id}
```

> **Attention** : `BackgroundTasks` est léger et **non garanti**. Si le serveur plante après la réponse, la tâche n'est pas exécutée. Pour des tâches critiques (emails transactionnels, paiements), utilisez Celery (Chapitre 17).

---

### 13.4 — asyncio.gather : Paralléliser les requêtes

```python
# Exemple concret NexaHub : récupérer le dashboard d'un utilisateur
# qui nécessite plusieurs requêtes DB distinctes

@router.get("/dashboard")
async def get_dashboard(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """
    Retourne toutes les données du dashboard en UN seul appel.
    Optimisé avec asyncio.gather pour paralléliser les requêtes DB.
    """

    # Lancer toutes les requêtes EN PARALLÈLE
    my_projects, assigned_tasks, recent_activity, team_members = await asyncio.gather(
        ProjectService.get_all(db, owner_id=current_user.id, limit=5),
        TaskService.get_assigned_to(db, user_id=current_user.id, status="in_progress"),
        ActivityService.get_recent(db, user_id=current_user.id, limit=10),
        TeamService.get_teammates(db, user_id=current_user.id)
    )
    # Sans gather : 4 requêtes séquentielles (ex: 4x30ms = 120ms)
    # Avec gather : 4 requêtes parallèles (ex: max(30ms) = 30ms)

    return {
        "projects": my_projects,
        "assigned_tasks": assigned_tasks,
        "recent_activity": recent_activity,
        "team_members": team_members
    }
```

---

### [OK] Bonnes Pratiques du Chapitre 13

> **Toujours `async def` pour les routes qui font du I/O** (DB, HTTP, fichiers). Utilisez `def` uniquement pour les routes purement calculatoires.

> **`asyncio.gather` pour les requêtes indépendantes** : si deux requêtes ne dépendent pas l'une de l'autre, lancez-les en parallèle.

> **`BackgroundTasks` pour les effets de bord non-critiques** : emails de notification, analytics, logs externes. Utilisez Celery pour les tâches critiques.

---

### [EFFORT] Exercice de Complétion — Chapitre 13

```python
# [NOTE] Optimiser cet endpoint qui charge un projet avec toutes ses données

@router.get("/projects/{project_id}/full")
async def get_project_full(
    project_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """
    Retourne un projet avec :
    - Les détails du projet
    - Les membres de l'équipe
    - Les tâches en cours (status="in_progress")
    - Les 5 dernières activités
    - Les statistiques (total tâches, tâches complétées, %)
    """

    # [X] VERSION SÉQUENTIELLE À TRANSFORMER (trop lente)
    project = await ProjectService.get_by_id(db, project_id)
    members = await ProjectService.get_members(db, project_id)
    tasks = await TaskService.get_by_project(db, project_id, status="in_progress")
    activities = await ActivityService.get_by_project(db, project_id, limit=5)
    stats = await TaskService.get_stats(db, project_id)

    # [NOTE] VOTRE MISSION : Réécrire en utilisant asyncio.gather
    # pour paralléliser les requêtes indépendantes
    # Attention : les stats dépendent-elles des tâches ?
    # Réfléchissez aux dépendances avant de paralléliser

    return {
        "project": project,
        "members": members,
        "active_tasks": tasks,
        "recent_activity": activities,
        "stats": stats
    }

# [NOTE] Ajouter des BackgroundTasks à la route de création de tâche :
# - Notifier par email l'utilisateur assigné
# - Mettre à jour les statistiques du projet dans Redis
# - Logger l'événement dans un service d'analytics
@router.post("/tasks/")
async def create_task(
    task: TaskCreate,
    background_tasks: ___,  # [NOTE] Comment l'injecter ?
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    new_task = await TaskService.create(db, task, creator_id=current_user.id)

    # [NOTE] Ajouter 3 background tasks ici
    if new_task.assignee_id:
        background_tasks.add_task(___, ___)

    return new_task
```

---

## [LIVRE] Chapitre 14 : Optimisation & Caching

### [OBJECTIF] Objectifs du chapitre
- Implémenter un cache Redis pour les données fréquemment lues
- Paginer les résultats efficacement
- Compresser les réponses volumineuses

---

### 14.1 — Caching avec Redis

Redis est une base de données clé-valeur en mémoire. Elle est extrêmement rapide (~1ms) et idéale pour cacher des données qui changent peu.

```bash
pip install redis[asyncio] hiredis
```

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

import redis.asyncio as redis
import json
from typing import Any, Optional, Callable
from functools import wraps
from app.core.config import settings

# Client Redis global
redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True)


async def cache_get(key: str) -> Optional[Any]:
    """Lire depuis le cache."""
    value = await redis_client.get(key)
    if value:
        return json.loads(value)
    return None


async def cache_set(key: str, value: Any, expire_seconds: int = 300) -> None:
    """Écrire dans le cache avec expiration."""
    await redis_client.setex(
        key,
        expire_seconds,
        json.dumps(value, default=str)  # default=str pour les datetime
    )


async def cache_delete(key: str) -> None:
    """Invalider une entrée du cache."""
    await redis_client.delete(key)


async def cache_delete_pattern(pattern: str) -> None:
    """Invalider toutes les entrées matchant un pattern. Ex: 'projects:*'"""
    keys = await redis_client.keys(pattern)
    if keys:
        await redis_client.delete(*keys)
```

---

### 14.2 — Utiliser le cache dans les services

```python
# app/services/project_service.py — version avec cache

from app.core.cache import cache_get, cache_set, cache_delete_pattern
import json

class ProjectService:

    @staticmethod
    async def get_by_id(db: AsyncSession, project_id: int) -> Project:
        """Récupère un projet avec mise en cache."""
        cache_key = f"project:{project_id}"

        # 1. Chercher dans le cache
        cached = await cache_get(cache_key)
        if cached:
            # Cache hit ! Retourner sans aller en DB
            print(f"[CACHE HIT] {cache_key}")
            return cached  # Note: dans la vraie vie, reconstruire l'objet ORM

        # 2. Cache miss -> aller en DB
        print(f"[CACHE MISS] {cache_key}")
        result = await db.execute(
            select(Project).where(Project.id == project_id)
        )
        project = result.scalar_one_or_none()

        if not project:
            raise HTTPException(status_code=404, detail="Projet introuvable")

        # 3. Mettre en cache pour 5 minutes
        project_dict = {
            "id": project.id,
            "name": project.name,
            "description": project.description,
            "status": project.status,
            "owner_id": project.owner_id,
        }
        await cache_set(cache_key, project_dict, expire_seconds=300)

        return project

    @staticmethod
    async def update(db: AsyncSession, project_id: int, ...) -> Project:
        """Met à jour un projet ET invalide le cache."""
        # ... logique de mise à jour

        # Invalider le cache du projet modifié
        await cache_delete(f"project:{project_id}")
        # Invalider aussi la liste des projets (elle contient ce projet)
        await cache_delete_pattern("projects:list:*")

        return updated_project
```

---

### 14.3 — Pagination avec métadonnées

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

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

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    """
    Réponse paginée générique.

    Utilisation :
        response_model=PaginatedResponse[ProjectResponse]
    """
    items: List[T]
    total: int          # Nombre total d'éléments (sans pagination)
    page: int           # Page actuelle
    per_page: int       # Éléments par page
    pages: int          # Nombre total de pages
    has_next: bool      # Il y a une page suivante ?
    has_prev: bool      # Il y a une page précédente ?

# app/services/project_service.py

from sqlalchemy import func, select

async def get_paginated(
    db: AsyncSession,
    page: int = 1,
    per_page: int = 20
) -> PaginatedResponse:
    """Récupère des projets avec pagination et métadonnées."""

    skip = (page - 1) * per_page

    # Compter le total (requête séparée)
    count_query = select(func.count(Project.id))
    total_result = await db.execute(count_query)
    total = total_result.scalar()

    # Récupérer les items de la page
    items_query = select(Project).offset(skip).limit(per_page)
    items_result = await db.execute(items_query)
    items = items_result.scalars().all()

    pages = (total + per_page - 1) // per_page  # Arrondi supérieur

    return {
        "items": items,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": pages,
        "has_next": page < pages,
        "has_prev": page > 1
    }
```

---

### 14.4 — Compression des réponses

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

# Activer la compression GZIP pour les réponses > 1KB
# Réduit la taille des réponses JSON de 60-80%
app.add_middleware(GZipMiddleware, minimum_size=1000)
```

---

### [OK] Bonnes Pratiques du Chapitre 14

> **Profiler avant d'optimiser** : utilisez des outils comme Postman ou Locust pour identifier les vraies lenteurs avant d'ajouter un cache.

> **Stratégie d'invalidation du cache** : le problème dur du cache est de savoir quand invalider. Adoptez une stratégie claire : TTL fixe, invalidation explicite à la mise à jour, ou combinaison des deux.

> **Ne cachez pas les données sensibles** ou personnelles dans un cache partagé sans précautions.

---

### [EFFORT] Exercice de Complétion — Chapitre 14

```python
# [NOTE] Implémenter un décorateur de cache générique

from functools import wraps

def cached(key_prefix: str, expire: int = 300):
    """
    Décorateur qui cache automatiquement le résultat d'une fonction async.

    Utilisation :
        @cached("user", expire=600)
        async def get_user(user_id: int) -> dict:
            ...
    # Cache key sera : "user:{user_id}"
    """
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # [NOTE] Construire la cache key à partir du key_prefix et des arguments
            # Exemple : pour get_user(user_id=42), la key doit être "user:42"
            cache_key = f"{key_prefix}:{___}"

            # [NOTE] Chercher dans le cache
            cached_value = await cache_get(___)
            if cached_value is not None:
                return ___

            # [NOTE] Appeler la vraie fonction
            result = await func(*args, **kwargs)

            # [NOTE] Mettre en cache
            await cache_set(___, ___, expire_seconds=___)

            return result
        return wrapper
    return decorator

# [NOTE] Utiliser ce décorateur sur une route qui retourne souvent les mêmes données
# Exemple : les statistiques d'un projet (recalculées toutes les 5 minutes max)
@cached("project_stats", expire=300)
async def get_project_stats(project_id: int) -> dict:
    # [NOTE] Retourner : total_tasks, completed_tasks, completion_percentage,
    #               overdue_tasks, active_members_count
    pass
```

---

# [RESEAU] PARTIE 5 — Communication & Intégrations

---

## [LIVRE] Chapitre 15 : Uploads & Downloads

### [OBJECTIF] Objectifs du chapitre
- Gérer les uploads de fichiers de façon sécurisée
- Servir des fichiers statiques et téléchargeables
- Générer des fichiers dynamiques (PDF, CSV)

---

### 15.1 — Upload de fichiers

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

from fastapi import APIRouter, UploadFile, File, Depends, HTTPException
from fastapi.responses import FileResponse, StreamingResponse
import aiofiles  # pip install aiofiles
import os
import uuid
from pathlib import Path

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

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".pdf", ".docx", ".xlsx"}

@router.post("/upload")
async def upload_file(
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_user)
):
    """Upload un fichier et retourne son URL."""

    # Vérifier l'extension
    ext = Path(file.filename).suffix.lower()
    if ext not in ALLOWED_EXTENSIONS:
        raise HTTPException(
            status_code=415,
            detail=f"Extension non autorisée. Permises : {ALLOWED_EXTENSIONS}"
        )

    # Lire le fichier par chunks pour ne pas saturer la mémoire
    contents = b""
    async for chunk in file:
        contents += chunk
        if len(contents) > MAX_FILE_SIZE:
            raise HTTPException(status_code=413, detail="Fichier trop volumineux (max 10MB)")

    # Générer un nom unique pour éviter les collisions et les path traversal
    unique_filename = f"{uuid.uuid4()}{ext}"
    user_dir = UPLOAD_DIR / str(current_user.id)
    user_dir.mkdir(exist_ok=True)
    file_path = user_dir / unique_filename

    # Écrire le fichier de façon asynchrone
    async with aiofiles.open(file_path, "wb") as f:
        await f.write(contents)

    return {
        "filename": unique_filename,
        "original_name": file.filename,
        "size": len(contents),
        "url": f"/files/{current_user.id}/{unique_filename}"
    }


@router.post("/upload/multiple")
async def upload_multiple_files(
    files: list[UploadFile] = File(...),
    current_user: User = Depends(get_current_user)
):
    """Upload plusieurs fichiers en une seule requête."""
    if len(files) > 10:
        raise HTTPException(status_code=400, detail="Maximum 10 fichiers à la fois")

    results = []
    for file in files:
        # Réutiliser la logique de l'endpoint précédent
        result = await upload_file(file, current_user)
        results.append(result)

    return {"uploaded": results, "count": len(results)}
```

---

### 15.2 — Téléchargement et génération de fichiers

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

from fastapi.responses import StreamingResponse
import csv
import io

@router.get("/projects/{project_id}/export/csv")
async def export_tasks_csv(
    project_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """Exporte les tâches d'un projet en CSV."""

    tasks = await TaskService.get_by_project(db, project_id)

    # Créer le CSV en mémoire
    output = io.StringIO()
    writer = csv.DictWriter(
        output,
        fieldnames=["id", "title", "status", "priority", "assignee", "due_date"]
    )
    writer.writeheader()

    for task in tasks:
        writer.writerow({
            "id": task.id,
            "title": task.title,
            "status": task.status,
            "priority": task.priority,
            "assignee": task.assignee.username if task.assignee else "",
            "due_date": str(task.due_date) if task.due_date else ""
        })

    output.seek(0)

    # Retourner comme fichier téléchargeable
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename=project_{project_id}_tasks.csv"
        }
    )
```

---

### [OK] Bonnes Pratiques du Chapitre 15

> **Ne jamais stocker le nom original** du fichier uploadé sur le disque. Utilisez un UUID pour éviter path traversal et collisions.

> **Toujours limiter la taille** des uploads et vérifier le type MIME réel.

> **Pour les gros fichiers**, utilisez le streaming plutôt que de tout charger en mémoire.

---

### [EFFORT] Exercice de Complétion — Chapitre 15

```python
# [NOTE] Implémenter l'export PDF d'un rapport de projet
# Utiliser la bibliothèque reportlab : pip install reportlab

from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
import io

@router.get("/projects/{project_id}/export/pdf")
async def export_project_report_pdf(
    project_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    """Génère un rapport PDF du projet."""

    # [NOTE] Récupérer le projet et ses données
    project = await ___
    tasks = await ___
    members = await ___

    # [NOTE] Générer le PDF
    buffer = io.BytesIO()
    pdf = canvas.Canvas(buffer, pagesize=A4)

    # [NOTE] Ajouter ces éléments au PDF :
    # - Titre : "Rapport NexaHub — [nom du projet]"
    # - Date de génération
    # - Informations du projet (description, status, owner)
    # - Liste des membres
    # - Tableau des tâches avec colonnes : titre, statut, priorité, assigné

    pdf.save()
    buffer.seek(0)

    return StreamingResponse(
        buffer,
        media_type="application/pdf",
        headers={
            "Content-Disposition": f"attachment; filename=rapport_{project_id}.pdf"
        }
    )
```

---

## [LIVRE] Chapitre 16 : WebSockets & Streaming Temps Réel

### [OBJECTIF] Objectifs du chapitre
- Comprendre les WebSockets et leur différence avec HTTP
- Implémenter un système de notifications temps réel
- Gérer plusieurs clients connectés simultanément

---

### 16.1 — HTTP vs WebSocket

| Aspect | HTTP | WebSocket |
|---|---|---|
| Type de connexion | Request/Response (fermée après) | Connexion persistante bidirectionnelle |
| Initiateur | Client uniquement | Client et serveur peuvent envoyer |
| Overhead | Headers HTTP à chaque requête | Minimal après le handshake |
| Cas d'usage | CRUD, API REST | Chat, notifications, collaboration live |

Pour NexaHub, on utilise les WebSockets pour :
- Notifier en temps réel quand une tâche est modifiée
- Afficher qui est en ligne sur un projet
- Chat d'équipe en temps réel

---

### 16.2 — WebSocket Manager

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

from fastapi import WebSocket
from typing import Dict, List, Set
import json
import asyncio

class ConnectionManager:
    """
    Gère toutes les connexions WebSocket actives.

    Structure :
    - connections: {project_id: [WebSocket, WebSocket, ...]}
    Permet de broadcaster à tous les membres d'un projet.
    """

    def __init__(self):
        # project_id -> ensemble de WebSockets connectés
        self.project_connections: Dict[int, Set[WebSocket]] = {}
        # user_id -> WebSocket (connexion personnelle)
        self.user_connections: Dict[int, WebSocket] = {}

    async def connect(
        self,
        websocket: WebSocket,
        user_id: int,
        project_id: int
    ) -> None:
        """Accepter et enregistrer une nouvelle connexion."""
        await websocket.accept()

        # Enregistrer pour le projet
        if project_id not in self.project_connections:
            self.project_connections[project_id] = set()
        self.project_connections[project_id].add(websocket)

        # Enregistrer pour l'utilisateur
        self.user_connections[user_id] = websocket

        # Notifier les autres membres du projet
        await self.broadcast_to_project(
            project_id,
            {
                "type": "user_joined",
                "user_id": user_id,
                "message": f"L'utilisateur {user_id} s'est connecté"
            },
            exclude=websocket
        )

    def disconnect(self, websocket: WebSocket, user_id: int, project_id: int) -> None:
        """Désenregistrer une connexion fermée."""
        if project_id in self.project_connections:
            self.project_connections[project_id].discard(websocket)
            if not self.project_connections[project_id]:
                del self.project_connections[project_id]

        if user_id in self.user_connections:
            del self.user_connections[user_id]

    async def send_to_user(self, user_id: int, message: dict) -> bool:
        """Envoyer un message à un utilisateur spécifique."""
        websocket = self.user_connections.get(user_id)
        if websocket:
            try:
                await websocket.send_json(message)
                return True
            except Exception:
                # La connexion est morte, la nettoyer
                del self.user_connections[user_id]
        return False  # Utilisateur hors ligne

    async def broadcast_to_project(
        self,
        project_id: int,
        message: dict,
        exclude: WebSocket = None
    ) -> None:
        """Broadcaster un message à tous les membres d'un projet."""
        connections = self.project_connections.get(project_id, set()).copy()
        dead_connections = set()

        for websocket in connections:
            if websocket == exclude:
                continue
            try:
                await websocket.send_json(message)
            except Exception:
                # Connexion morte -> marquer pour suppression
                dead_connections.add(websocket)

        # Nettoyer les connexions mortes
        for dead in dead_connections:
            self.project_connections[project_id].discard(dead)

    def get_online_users(self, project_id: int) -> List[int]:
        """Retourne les IDs des utilisateurs en ligne sur un projet."""
        # Dans une vraie app, on stockerait aussi le mapping websocket->user_id
        return list(self.user_connections.keys())


# Instance globale partagée
manager = ConnectionManager()
```

---

### 16.3 — Route WebSocket

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

from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends, Query
from app.core.websocket_manager import manager
from app.core.security import decode_token
from jose import JWTError

router = APIRouter(tags=["websockets"])

@router.websocket("/ws/projects/{project_id}")
async def project_websocket(
    websocket: WebSocket,
    project_id: int,
    token: str = Query(..., description="JWT token pour l'authentification")
):
    """
    WebSocket pour un projet spécifique.
    Le client doit envoyer son JWT token en query parameter.

    URL : ws://localhost:8000/ws/projects/42?token=eyJ...
    """

    # Authentifier via le token JWT
    # (Les WebSockets ne supportent pas les headers Authorization facilement)
    try:
        payload = decode_token(token)
        user_id = int(payload["sub"])
    except (JWTError, KeyError, ValueError):
        await websocket.close(code=4001, reason="Token invalide")
        return

    # Connecter le client
    await manager.connect(websocket, user_id=user_id, project_id=project_id)

    # Envoyer un message de bienvenue
    await websocket.send_json({
        "type": "connected",
        "project_id": project_id,
        "user_id": user_id,
        "online_users": manager.get_online_users(project_id)
    })

    try:
        # Boucle principale : écouter les messages du client
        while True:
            # Attendre un message (bloquant de façon async)
            data = await websocket.receive_json()

            # Types de messages gérés
            if data.get("type") == "task_update":
                # Broadcaster la mise à jour à tous les membres du projet
                await manager.broadcast_to_project(
                    project_id,
                    {
                        "type": "task_updated",
                        "task_id": data.get("task_id"),
                        "changes": data.get("changes"),
                        "updated_by": user_id
                    }
                )

            elif data.get("type") == "cursor_move":
                # Diffuser la position du curseur (pour l'édition collaborative)
                await manager.broadcast_to_project(
                    project_id,
                    {
                        "type": "cursor_moved",
                        "user_id": user_id,
                        "position": data.get("position")
                    },
                    exclude=websocket  # Pas besoin de se l'envoyer à soi-même
                )

            elif data.get("type") == "ping":
                # Keepalive : répondre avec un pong
                await websocket.send_json({"type": "pong"})

    except WebSocketDisconnect:
        # Le client s'est déconnecté
        manager.disconnect(websocket, user_id=user_id, project_id=project_id)

        # Notifier les autres
        await manager.broadcast_to_project(
            project_id,
            {
                "type": "user_left",
                "user_id": user_id,
                "message": f"L'utilisateur {user_id} s'est déconnecté"
            }
        )
```

---

### [OK] Bonnes Pratiques du Chapitre 16

> **Authentifiez toujours les WebSockets** via un token en query parameter (les headers Authorization ne sont pas facilement accessibles dans les WS).

> **Gérez toujours `WebSocketDisconnect`** pour nettoyer les connexions mortes.

> **Pour une application à grande échelle**, utilisez Redis Pub/Sub pour synchroniser les messages entre plusieurs instances de serveur.

---

### [EFFORT] Exercice de Complétion — Chapitre 16

```python
# [NOTE] Implémenter un système de notifications WebSocket global
# Chaque utilisateur a sa propre connexion WS pour les notifications personnelles

@router.websocket("/ws/notifications")
async def notifications_websocket(
    websocket: WebSocket,
    token: str = Query(...)
):
    """
    WebSocket de notifications pour l'utilisateur connecté.
    Reçoit toutes ses notifications en temps réel.
    """
    # [NOTE] Authentifier l'utilisateur
    try:
        user_id = ___
    except ___:
        await websocket.close(code=4001)
        return

    # [NOTE] Connecter l'utilisateur au manager
    # (sans project_id puisque c'est une connexion personnelle)
    await ___

    await websocket.send_json({
        "type": "connected",
        "message": "Connexion aux notifications établie"
    })

    try:
        while True:
            # [NOTE] Les notifications WS n'ont pas besoin d'écouter les messages clients
            # On peut juste attendre et rester connecté avec un keepalive
            data = await websocket.receive_json()
            if data.get("type") == "ping":
                await websocket.send_json({"type": "pong"})

    except WebSocketDisconnect:
        # [NOTE] Déconnecter proprement
        ___

# [NOTE] Modifier la route create_task pour notifier via WS l'utilisateur assigné
@router.post("/tasks/")
async def create_task(task: TaskCreate, background_tasks: BackgroundTasks, ...):
    new_task = await TaskService.create(...)

    # [NOTE] Si une tâche est assignée, notifier l'assigné via WebSocket
    if new_task.assignee_id:
        # Essayer d'abord en temps réel via WS
        sent = await manager.send_to_user(
            new_task.assignee_id,
            {
                "type": "task_assigned",
                "task_id": new_task.id,
                "task_title": new_task.title,
                "project_id": new_task.project_id
            }
        )

        # Si l'utilisateur est hors ligne, mettre en file d'attente
        if not sent:
            # [NOTE] Enregistrer la notification dans la DB pour la récupérer plus tard
            background_tasks.add_task(___, ___)
```

---

## [LIVRE] Chapitre 17 : Background Tasks & Scheduling avec Celery

### [OBJECTIF] Objectifs du chapitre
- Comprendre quand Celery est nécessaire vs BackgroundTasks
- Configurer Celery avec Redis comme broker
- Créer des tâches planifiées (scheduled tasks)

---

### 17.1 — Pourquoi Celery ?

| Critère | BackgroundTasks | Celery |
|---|---|---|
| Fiabilité | Non garanti (crash = perte) | Garanti (persisté dans Redis/RabbitMQ) |
| Monitoring | Aucun | Dashboard Flower complet |
| Retry automatique | Non | Oui, configurable |
| Tâches planifiées | Non | Oui (Celery Beat) |
| Workers séparés | Non (même process) | Oui (workers indépendants) |
| Cas d'usage | Logs, analytics, notifs non-critiques | Emails transactionnels, paiements, reports |

---

### 17.2 — Configuration Celery

```bash
pip install celery[redis] flower
```

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

from celery import Celery
from app.core.config import settings

# Créer l'application Celery
celery_app = Celery(
    "nexahub",
    broker=settings.REDIS_URL,    # Redis comme broker de messages
    backend=settings.REDIS_URL,   # Redis pour stocker les résultats
    include=[
        "app.tasks.email_tasks",
        "app.tasks.report_tasks",
        "app.tasks.cleanup_tasks",
    ]
)

# Configuration
celery_app.conf.update(
    # Sérialisation JSON (plus sécurisé que pickle)
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Europe/Paris",
    enable_utc=True,

    # Retry automatique en cas d'échec
    task_max_retries=3,
    task_default_retry_delay=60,  # Réessayer après 60 secondes

    # Expire les résultats après 1 heure
    result_expires=3600,

    # Planification des tâches (Celery Beat)
    beat_schedule={
        "cleanup-expired-tokens": {
            "task": "app.tasks.cleanup_tasks.cleanup_expired_tokens",
            "schedule": 3600.0,  # Toutes les heures
        },
        "send-daily-digest": {
            "task": "app.tasks.email_tasks.send_daily_digest",
            "schedule": {
                "hour": 8,        # Tous les jours à 8h
                "minute": 0,
            }
        },
    }
)
```

---

### 17.3 — Définir des tâches Celery

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

from app.tasks.celery_app import celery_app
import smtplib
from email.mime.text import MIMEText
from app.core.config import settings

@celery_app.task(
    bind=True,           # Accès à `self` pour les retries
    max_retries=3,
    default_retry_delay=120  # Réessayer après 2 minutes
)
def send_task_assignment_email(self, user_email: str, task_title: str, project_name: str):
    """
    Envoyer un email quand une tâche est assignée.
    Cette tâche est garanti d'être exécutée même si le serveur redémarre.
    """
    try:
        msg = MIMEText(f"""
Bonjour,

Une nouvelle tâche vous a été assignée sur NexaHub :
- Projet : {project_name}
- Tâche : {task_title}

Connectez-vous sur https://nexahub.io pour voir les détails.

L'équipe NexaHub
        """)
        msg["Subject"] = f"[NexaHub] Nouvelle tâche : {task_title}"
        msg["From"] = "noreply@nexahub.io"
        msg["To"] = user_email

        with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) as server:
            server.starttls()
            server.login(settings.SMTP_USER, settings.SMTP_PASSWORD)
            server.send_message(msg)

    except Exception as exc:
        # Retry automatique avec délai exponentiel
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)
```

---

### 17.4 — Appeler Celery depuis FastAPI

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

from app.tasks.email_tasks import send_task_assignment_email

@router.post("/tasks/")
async def create_task(task: TaskCreate, db: AsyncSession = Depends(get_db), ...):
    new_task = await TaskService.create(db, task, ...)

    if new_task.assignee_id:
        assignee = await UserService.get_by_id(db, new_task.assignee_id)
        project = await ProjectService.get_by_id(db, new_task.project_id)

        # Déléguer l'envoi d'email à Celery (async, non-bloquant, garanti)
        send_task_assignment_email.delay(
            user_email=assignee.email,
            task_title=new_task.title,
            project_name=project.name
        )

    return new_task
```

```bash
# Lancer les workers Celery (dans un terminal séparé)
celery -A app.tasks.celery_app worker --loglevel=info

# Lancer le planificateur Celery Beat
celery -A app.tasks.celery_app beat --loglevel=info

# Dashboard de monitoring Flower
celery -A app.tasks.celery_app flower --port=5555
```

---

### [OK] Bonnes Pratiques du Chapitre 17

> **Utilisez Celery pour tout ce qui est critique** : emails transactionnels, paiements, rapports. Utilisez BackgroundTasks pour les analytics et logs.

> **Toujours configurer des retries** avec un délai exponentiel pour gérer les pannes temporaires.

> **Utilisez Flower** en développement pour surveiller vos tâches et debugger.

---

### [EFFORT] Exercice de Complétion — Chapitre 17

```python
# [NOTE] Créer une tâche Celery pour générer un rapport hebdomadaire

@celery_app.task(bind=True, max_retries=2)
def generate_weekly_report(self, user_id: int):
    """
    Génère un rapport hebdomadaire pour un utilisateur.
    Planifié chaque lundi à 9h.

    Le rapport doit contenir :
    - Nombre de tâches complétées la semaine dernière
    - Tâches en retard
    - Activité de l'équipe
    - Prévision pour la semaine suivante
    """
    try:
        # [NOTE] Créer une session DB synchrone pour Celery
        # (Celery ne supporte pas les sessions async SQLAlchemy directement)
        # Astuce : utiliser une session synchrone séparée

        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker
        from app.core.config import settings

        sync_url = settings.DATABASE_URL.replace("postgresql+asyncpg", "postgresql")
        engine = create_engine(sync_url)
        Session = sessionmaker(engine)

        with Session() as db:
            # [NOTE] Récupérer les données de la semaine passée
            # [NOTE] Générer le rapport (PDF ou email HTML)
            # [NOTE] Envoyer l'email
            pass

    except Exception as exc:
        raise self.retry(exc=exc, countdown=___)

# [NOTE] Ajouter cette tâche au planning Celery Beat
# Elle doit s'exécuter chaque lundi à 9h00
# Utiliser la syntaxe crontab de Celery
```

---

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

### [OBJECTIF] Objectifs du chapitre
- Envoyer des emails HTML avec templates
- Implémenter les notifications multi-canaux (email, WS, webhook)

---

### 18.1 — Emails avec fastapi-mail

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

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

from fastapi_mail import FastMail, MessageSchema, ConnectionConfig, MessageType
from jinja2 import Environment, FileSystemLoader
from app.core.config import settings

# Configuration SMTP
mail_config = ConnectionConfig(
    MAIL_USERNAME=settings.SMTP_USER,
    MAIL_PASSWORD=settings.SMTP_PASSWORD,
    MAIL_FROM="noreply@nexahub.io",
    MAIL_FROM_NAME="NexaHub",
    MAIL_PORT=settings.SMTP_PORT,
    MAIL_SERVER=settings.SMTP_HOST,
    MAIL_STARTTLS=True,
    MAIL_SSL_TLS=False,
    TEMPLATE_FOLDER="app/templates/emails"  # Dossier de templates Jinja2
)

mail = FastMail(mail_config)

async def send_email(
    to: list[str],
    subject: str,
    template_name: str,
    template_data: dict
) -> None:
    """Envoie un email HTML basé sur un template Jinja2."""
    message = MessageSchema(
        subject=subject,
        recipients=to,
        template_body=template_data,
        subtype=MessageType.html
    )
    await mail.send_message(message, template_name=template_name)


# Templates dans app/templates/emails/
# welcome.html, task_assigned.html, project_invite.html, etc.
```

```html
<!-- app/templates/emails/task_assigned.html -->
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <style>
        body { font-family: Arial, sans-serif; background-color: #f5f5f5; }
        .container { max-width: 600px; margin: 20px auto; background: white; padding: 30px; border-radius: 8px; }
        .header { background: #6366f1; color: white; padding: 20px; border-radius: 6px 6px 0 0; }
        .task-card { border: 1px solid #e5e7eb; border-radius: 6px; padding: 15px; margin: 15px 0; }
        .btn { background: #6366f1; color: white; padding: 12px 24px; text-decoration: none; border-radius: 6px; }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>[RAPIDE] NexaHub</h1>
        </div>
        <p>Bonjour <strong>{{ username }}</strong>,</p>
        <p>Une nouvelle tâche vous a été assignée :</p>
        <div class="task-card">
            <h3>{{ task_title }}</h3>
            <p>Projet : {{ project_name }}</p>
            <p>Priorité : {{ priority }}/5</p>
            {% if due_date %}
            <p>Échéance : {{ due_date }}</p>
            {% endif %}
        </div>
        <p><a href="{{ app_url }}/tasks/{{ task_id }}" class="btn">Voir la tâche</a></p>
    </div>
</body>
</html>
```

---

### [OK] Bonnes Pratiques du Chapitre 18

> **Utilisez toujours un service d'envoi asynchrone** pour les emails. Un email synchrone peut bloquer votre API pendant plusieurs secondes si le serveur SMTP est lent.

> **Testez vos emails avec Mailhog ou Mailtrap** en développement. Ne jamais envoyer d'emails réels en dev.

---

### [EFFORT] Exercice de Complétion — Chapitre 18

```python
# [NOTE] Créer un système de notifications multi-canal pour NexaHub

class NotificationService:
    """
    Service centralisé de notifications.
    Choisit automatiquement le canal selon les préférences de l'utilisateur.
    """

    @staticmethod
    async def notify_task_assigned(
        assignee_id: int,
        task_id: int,
        task_title: str,
        project_name: str,
        db: AsyncSession
    ) -> None:
        # [NOTE] Récupérer les préférences de notification de l'utilisateur
        user = await UserService.get_by_id(db, assignee_id)

        # [NOTE] 1. Toujours enregistrer en DB pour l'historique
        await NotificationRepository.create(db, {
            "user_id": assignee_id,
            "type": "task_assigned",
            "title": "Nouvelle tâche assignée",
            "body": f"Vous avez été assigné(e) à : {task_title}",
            "data": {"task_id": task_id, "project_name": project_name}
        })

        # [NOTE] 2. Notification WebSocket si l'utilisateur est en ligne
        sent_ws = await manager.send_to_user(assignee_id, {
            "type": "notification",
            "title": "Nouvelle tâche",
            "body": f"Vous avez été assigné(e) à : {task_title}"
        })

        # [NOTE] 3. Email si WS échoue (utilisateur hors ligne)
        # et si l'utilisateur a activé les emails de notification
        if not sent_ws and user.email_notifications:
            await send_email(
                to=[user.email],
                subject=f"[NexaHub] Nouvelle tâche : {task_title}",
                template_name=___,
                template_data={
                    "username": user.username,
                    "task_title": ___,
                    "project_name": ___,
                    "task_id": ___,
                    "app_url": settings.APP_URL
                }
            )
```

---

## [OBJECTIF] Récapitulatif — Parties 4 & 5

À ce stade, NexaHub est **performant et communicatif** :

| Composant | Statut |
|---|---|
| Async/Await dans toutes les routes | [OK] En place |
| Parallélisation avec asyncio.gather | [OK] En place |
| Cache Redis pour les données fréquentes | [OK] En place |
| Pagination avec métadonnées | [OK] En place |
| Upload de fichiers sécurisé | [OK] En place |
| Export CSV et PDF | [OK] En place |
| WebSockets temps réel | [OK] En place |
| Celery pour les tâches critiques | [OK] En place |
| Emails HTML avec templates | [OK] En place |
| Notifications multi-canal | [OK] En place |

### Prochain chapitre :
La **Partie 5** couvre les tests, le CI/CD et le déploiement en production.

---

*[BOOKMARK] Fichier suivant : `fastapi_nexahub_partie5.md` — Tests, CI/CD, Déploiement & Architecture Expert*

# [RAPIDE] NexaHub — Partie 5 : Tests, Déploiement & Architecture Expert
## *Tester, Déployer, Scaler*

---

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

---

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

### [OBJECTIF] Objectifs du chapitre
- Écrire des tests robustes avec pytest et httpx
- Mocker les dépendances et la base de données
- Tester les routes authentifiées

---

### 19.1 — Philosophie des tests pour FastAPI

Une API bien testée couvre :
- **Tests unitaires** : tester une fonction isolée (ex: `hash_password`, un validateur Pydantic)
- **Tests d'intégration** : tester un endpoint de bout en bout (HTTP -> Service -> DB)
- **Tests de charge** : vérifier que l'API tient sous une forte charge (Locust, Chapitre 25)

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

---

### 19.2 — Configuration de base (conftest.py)

```python
# tests/conftest.py

import pytest
import asyncio
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.pool import StaticPool

from app.main import app
from app.db.session import Base, get_db
from app.core.security import hash_password
from app.models.user import User

# ─────────────────────────────────────────────────────────────
# BASE DE DONNÉES DE TEST
# ─────────────────────────────────────────────────────────────

# Utiliser SQLite en mémoire pour les tests (rapide et isolé)
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

engine_test = create_async_engine(
    TEST_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,  # Pool simple pour les tests
)

TestSessionLocal = async_sessionmaker(
    engine_test,
    class_=AsyncSession,
    expire_on_commit=False
)


# Override de la dépendance get_db pour utiliser la DB de test
async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
    async with TestSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()


# Appliquer l'override AVANT les tests
app.dependency_overrides[get_db] = override_get_db


# ─────────────────────────────────────────────────────────────
# FIXTURES
# ─────────────────────────────────────────────────────────────

@pytest.fixture(scope="session")
def event_loop():
    """Créer une boucle asyncio pour toute la session de tests."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(autouse=True, scope="function")
async def setup_database():
    """Créer et supprimer les tables pour chaque test (isolation complète)."""
    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    yield  # Test s'exécute ici

    async with engine_test.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
    """Client HTTP async pour les tests."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac


@pytest.fixture
async def db_session() -> AsyncGenerator[AsyncSession, None]:
    """Session DB pour créer des données de test directement."""
    async with TestSessionLocal() as session:
        yield session


@pytest.fixture
async def test_user(db_session: AsyncSession) -> User:
    """Créer un utilisateur de test."""
    user = User(
        username="testuser",
        email="test@nexahub.io",
        hashed_password=hash_password("TestPassword123!"),
        full_name="Test User",
        is_active=True
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user


@pytest.fixture
async def test_user_token(client: AsyncClient, test_user: User) -> str:
    """Obtenir un token JWT valide pour l'utilisateur de test."""
    response = await client.post("/auth/login", json={
        "email": "test@nexahub.io",
        "password": "TestPassword123!"
    })
    assert response.status_code == 200
    return response.json()["access_token"]


@pytest.fixture
def auth_headers(test_user_token: str) -> dict:
    """Headers avec token JWT pour les requêtes authentifiées."""
    return {"Authorization": f"Bearer {test_user_token}"}
```

---

### 19.3 — Écrire les tests

```python
# tests/test_auth.py

import pytest
from httpx import AsyncClient

class TestRegister:
    """Tests de l'endpoint POST /auth/register"""

    async def test_register_success(self, client: AsyncClient):
        """Test d'inscription réussie."""
        response = await client.post("/auth/register", json={
            "username": "alice",
            "email": "alice@nexahub.io",
            "password": "MonMotDePasse123!",
            "full_name": "Alice Dupont"
        })
        assert response.status_code == 201
        data = response.json()
        assert data["username"] == "alice"
        assert "password" not in str(data)  # Le mot de passe ne doit jamais apparaître !

    async def test_register_duplicate_email(self, client: AsyncClient, test_user):
        """Test : l'email est déjà utilisé."""
        response = await client.post("/auth/register", json={
            "username": "newuser",
            "email": "test@nexahub.io",  # Email déjà pris par test_user
            "password": "Password123!",
            "full_name": "New User"
        })
        assert response.status_code == 409
        assert "email" in response.json()["detail"].lower()

    async def test_register_invalid_email(self, client: AsyncClient):
        """Test : email invalide."""
        response = await client.post("/auth/register", json={
            "username": "bob",
            "email": "pas-un-email",
            "password": "Password123!",
            "full_name": "Bob"
        })
        assert response.status_code == 422

    async def test_register_password_too_short(self, client: AsyncClient):
        """Test : mot de passe trop court."""
        response = await client.post("/auth/register", json={
            "username": "charlie",
            "email": "charlie@test.io",
            "password": "abc",   # Trop court (< 8 chars)
            "full_name": "Charlie"
        })
        assert response.status_code == 422


class TestLogin:

    async def test_login_success(self, client: AsyncClient, test_user):
        """Test de connexion réussie."""
        response = await client.post("/auth/login", json={
            "email": "test@nexahub.io",
            "password": "TestPassword123!"
        })
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert "refresh_token" in data
        assert data["token_type"] == "bearer"

    async def test_login_wrong_password(self, client: AsyncClient, test_user):
        """Test : mauvais mot de passe."""
        response = await client.post("/auth/login", json={
            "email": "test@nexahub.io",
            "password": "MauvaisMotDePasse"
        })
        assert response.status_code == 401
        # Vérifier que le message d'erreur est vague (sécurité)
        assert "incorrect" in response.json()["detail"].lower()

    async def test_login_nonexistent_email(self, client: AsyncClient):
        """Test : email inexistant."""
        response = await client.post("/auth/login", json={
            "email": "nobody@nexahub.io",
            "password": "Password123!"
        })
        assert response.status_code == 401
        # Le même message que mauvais mot de passe (anti-énumération)


# tests/test_projects.py

class TestProjects:

    async def test_create_project_authenticated(
        self,
        client: AsyncClient,
        auth_headers: dict
    ):
        """Test : créer un projet en étant authentifié."""
        response = await client.post(
            "/projects/",
            json={"name": "Mon Projet", "description": "Test"},
            headers=auth_headers
        )
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Mon Projet"
        assert "id" in data
        assert "created_at" in data

    async def test_create_project_unauthenticated(self, client: AsyncClient):
        """Test : impossible de créer un projet sans token."""
        response = await client.post(
            "/projects/",
            json={"name": "Mon Projet"}
            # Pas de auth_headers !
        )
        assert response.status_code == 401

    async def test_list_projects_only_own(
        self,
        client: AsyncClient,
        auth_headers: dict,
        db_session
    ):
        """Test : l'utilisateur ne voit que ses propres projets."""
        # Créer 2 projets pour l'utilisateur test
        for name in ["Projet A", "Projet B"]:
            await client.post("/projects/", json={"name": name}, headers=auth_headers)

        # Créer un projet pour un AUTRE utilisateur directement en DB
        from app.models.project import Project
        other_project = Project(name="Projet Autre", owner_id=9999)
        db_session.add(other_project)
        await db_session.commit()

        # L'utilisateur ne doit voir que SES projets
        response = await client.get("/projects/", headers=auth_headers)
        assert response.status_code == 200
        projects = response.json()["items"]
        assert len(projects) == 2
        assert all(p["name"] in ["Projet A", "Projet B"] for p in projects)

    async def test_delete_project_only_owner(
        self,
        client: AsyncClient,
        auth_headers: dict,
        db_session
    ):
        """Test : seul le propriétaire peut supprimer un projet."""
        # Créer un projet
        create_res = await client.post(
            "/projects/",
            json={"name": "À Supprimer"},
            headers=auth_headers
        )
        project_id = create_res.json()["id"]

        # Créer un AUTRE utilisateur et tenter la suppression
        other_user = User(
            username="intruder",
            email="intruder@test.io",
            hashed_password=hash_password("Password123!"),
            full_name="Intruder",
            is_active=True
        )
        db_session.add(other_user)
        await db_session.commit()

        intruder_login = await client.post("/auth/login", json={
            "email": "intruder@test.io",
            "password": "Password123!"
        })
        intruder_headers = {"Authorization": f"Bearer {intruder_login.json()['access_token']}"}

        # L'intrus ne peut pas supprimer le projet
        delete_res = await client.delete(f"/projects/{project_id}", headers=intruder_headers)
        assert delete_res.status_code == 403

        # Le vrai propriétaire peut supprimer
        delete_res = await client.delete(f"/projects/{project_id}", headers=auth_headers)
        assert delete_res.status_code == 204
```

---

### [OK] Bonnes Pratiques du Chapitre 19

> **Testez les routes critiques en priorité** : authentification, permissions, CRUD principal.

> **Isolation complète** : chaque test repart d'une base de données vide. N'utilisez jamais l'état d'un test précédent.

> **Testez les cas d'erreur autant que les cas de succès** : 401, 403, 404, 422 sont aussi importants que 200.

---

### [EFFORT] Exercice de Complétion — Chapitre 19

```python
# tests/test_tasks.py
# [NOTE] Écrire les tests complets pour les tâches NexaHub

class TestTasks:

    # [NOTE] Test 1 : Créer une tâche dans un projet (authentifié, membre du projet)
    async def test_create_task_as_member(self, client, auth_headers, ...):
        pass

    # [NOTE] Test 2 : Assigner une tâche à un autre membre
    async def test_assign_task_to_member(self, ...):
        pass

    # [NOTE] Test 3 : Un non-membre ne peut pas créer de tâche dans un projet privé
    async def test_create_task_unauthorized(self, ...):
        # Expected status : 403
        pass

    # [NOTE] Test 4 : Changer le statut d'une tâche
    async def test_change_task_status(self, ...):
        # Todo -> In Progress -> Done
        pass

    # [NOTE] Test 5 : Supprimer une tâche supprime aussi ses commentaires (cascade)
    async def test_delete_task_cascades_comments(self, ...):
        pass

    # [NOTE] Test 6 : La liste des tâches supporte le filtrage par statut
    async def test_filter_tasks_by_status(self, ...):
        # Créer 3 tâches : 1 todo, 1 in_progress, 1 done
        # Vérifier que le filtre ?status=todo ne retourne que 1 tâche
        pass
```

---

## [LIVRE] Chapitre 20 : Logging & Monitoring

### [OBJECTIF] Objectifs du chapitre
- Configurer un système de logging structuré pour la production
- Intégrer Sentry pour les alertes d'erreur
- Monitorer avec Prometheus et Grafana

---

### 20.1 — Logging structuré

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

import logging
import sys
from app.core.config import settings

def setup_logging():
    """Configure le système de logging selon l'environnement."""

    # Format JSON pour la production (facilement parseable par Elasticsearch, Datadog, etc.)
    if settings.ENVIRONMENT == "production":
        import json

        class JsonFormatter(logging.Formatter):
            def format(self, record):
                log_data = {
                    "timestamp": self.formatTime(record),
                    "level": record.levelname,
                    "logger": record.name,
                    "message": record.getMessage(),
                    "module": record.module,
                    "function": record.funcName,
                    "line": record.lineno,
                }
                if record.exc_info:
                    log_data["exception"] = self.formatException(record.exc_info)
                return json.dumps(log_data)

        formatter = JsonFormatter()
    else:
        # Format lisible pour le développement
        formatter = logging.Formatter(
            "[%(asctime)s] %(levelname)s %(name)s — %(message)s",
            datefmt="%H:%M:%S"
        )

    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)

    # Configurer le logger root
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(
        logging.DEBUG if settings.DEBUG else logging.INFO
    )

    # Réduire le bruit des librairies tierces
    logging.getLogger("sqlalchemy.engine").setLevel(
        logging.INFO if settings.DEBUG else logging.WARNING
    )
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("httpx").setLevel(logging.WARNING)
```

---

### 20.2 — 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
from app.core.config import settings

if settings.SENTRY_DSN:
    sentry_sdk.init(
        dsn=settings.SENTRY_DSN,
        environment=settings.ENVIRONMENT,
        traces_sample_rate=0.1,  # 10% des requêtes tracées (performance)
        profiles_sample_rate=0.05,  # 5% de profiling
        integrations=[
            FastApiIntegration(transaction_style="endpoint"),
            SqlalchemyIntegration(),
        ],
        # Ne pas envoyer les infos personnelles à Sentry
        send_default_pii=False,
    )
```

---

### 20.3 — Métriques Prometheus

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

```python
# app/main.py

from prometheus_fastapi_instrumentator import Instrumentator

# Ajouter les métriques Prometheus automatiquement
Instrumentator().instrument(app).expose(app, endpoint="/metrics")
# Les métriques sont disponibles à GET /metrics
# Elles incluent : latence, taux d'erreurs, requêtes/seconde, etc.
```

---

### [OK] Bonnes Pratiques du Chapitre 20

> **Séparation des logs applicatifs et système.** Les logs de votre application doivent être distincts des logs Uvicorn/Nginx.

> **Structured logging (JSON) en production.** Facile à ingérer par Elasticsearch, Datadog, CloudWatch.

> **Sentry capture uniquement les exceptions non gérées.** Assurez-vous que vos handler globaux (Chapitre 8) loggent avant de swallower l'exception.

---

### [EFFORT] Exercice de Complétion — Chapitre 20

```python
# [NOTE] Créer un middleware qui log chaque requête avec des informations enrichies

@app.middleware("http")
async def rich_logging_middleware(request: Request, call_next):
    """
    Log chaque requête avec :
    - Request ID unique
    - Méthode HTTP et URL
    - User ID (si authentifié)
    - Status code
    - Durée en ms
    - Taille de la réponse en bytes
    """
    import time
    import uuid

    request_id = str(uuid.uuid4())[:8]
    start_time = time.time()

    # [NOTE] Extraire le user_id depuis le JWT si présent
    # (sans casser si le token est absent ou invalide)
    user_id = None
    auth_header = request.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        try:
            token = auth_header.replace("Bearer ", "")
            payload = decode_token(token)
            user_id = payload.get("sub")
        except Exception:
            pass  # Token invalide, pas grave pour le logging

    # [NOTE] Appeler le handler
    response = await call_next(request)

    # [NOTE] Logger avec toutes les informations
    duration_ms = (time.time() - start_time) * 1000
    logger.info(
        "HTTP Request",
        extra={
            "request_id": ___,
            "method": ___,
            "path": ___,
            "user_id": ___,
            "status_code": ___,
            "duration_ms": ___,
        }
    )

    return response
```

---

## [LIVRE] Chapitre 21 : Déploiement en Production

### [OBJECTIF] Objectifs du chapitre
- Dockeriser l'application FastAPI
- Configurer Nginx comme reverse proxy
- Utiliser Gunicorn + Uvicorn pour la production

---

### 21.1 — Dockerfile optimisé

```dockerfile
# Dockerfile

# ── ÉTAPE 1 : Builder ──────────────────────────────────────────
FROM python:3.11-slim AS builder

WORKDIR /app

# Copier seulement les requirements d'abord (optimisation du cache Docker)
COPY requirements.txt .

# Installer les dépendances dans un dossier isolé
RUN pip install --user --no-cache-dir -r requirements.txt


# ── ÉTAPE 2 : Production ──────────────────────────────────────
FROM python:3.11-slim AS production

# Créer un utilisateur non-root pour la sécurité
RUN groupadd --gid 1001 appgroup && \
    useradd --uid 1001 --gid appgroup --no-create-home appuser

WORKDIR /app

# Copier uniquement les packages installés depuis le builder
COPY --from=builder /root/.local /home/appuser/.local

# Copier le code source
COPY --chown=appuser:appgroup . .

# Basculer vers l'utilisateur non-root
USER appuser

# Variables d'environnement de production
ENV PATH=/home/appuser/.local/bin:$PATH \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    ENVIRONMENT=production

# Port exposé
EXPOSE 8000

# Healthcheck pour Docker/Kubernetes
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

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

---

### 21.2 — docker-compose.yml

```yaml
# docker-compose.yml

version: "3.9"

services:
  # L'API FastAPI
  api:
    build: .
    restart: unless-stopped
    env_file: .env
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    volumes:
      - uploads:/app/uploads
    networks:
      - nexahub-network

  # Base de données PostgreSQL
  db:
    image: postgres:15-alpine
    restart: unless-stopped
    environment:
      POSTGRES_DB: nexahub
      POSTGRES_USER: nexahub_user
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U nexahub_user -d nexahub"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - nexahub-network

  # Redis pour le cache et Celery
  redis:
    image: redis:7-alpine
    restart: unless-stopped
    command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD}
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - nexahub-network

  # Worker Celery
  celery_worker:
    build: .
    command: celery -A app.tasks.celery_app worker --loglevel=info --concurrency=4
    env_file: .env
    depends_on:
      - redis
      - db
    restart: unless-stopped
    networks:
      - nexahub-network

  # Planificateur Celery Beat
  celery_beat:
    build: .
    command: celery -A app.tasks.celery_app beat --loglevel=info
    env_file: .env
    depends_on:
      - redis
    restart: unless-stopped
    networks:
      - nexahub-network

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

volumes:
  postgres_data:
  redis_data:
  uploads:

networks:
  nexahub-network:
    driver: bridge
```

---

### 21.3 — Configuration Nginx

```nginx
# nginx.conf

events {
    worker_connections 1024;
}

http {
    upstream nexahub_api {
        server api:8000;
        keepalive 32;
    }

    # Redirection HTTP -> HTTPS
    server {
        listen 80;
        server_name api.nexahub.io;

        location /.well-known/acme-challenge/ {
            root /var/www/certbot;
        }

        location / {
            return 301 https://$host$request_uri;
        }
    }

    # Serveur HTTPS
    server {
        listen 443 ssl http2;
        server_name api.nexahub.io;

        ssl_certificate /etc/letsencrypt/live/api.nexahub.io/fullchain.pem;
        ssl_certificate_key /etc/letsencrypt/live/api.nexahub.io/privkey.pem;
        ssl_protocols TLSv1.2 TLSv1.3;

        # Limiter la taille des uploads
        client_max_body_size 15M;

        # Rate limiting au niveau Nginx
        limit_req_zone $binary_remote_addr zone=api_limit:10m rate=30r/m;

        location / {
            limit_req zone=api_limit burst=10 nodelay;

            proxy_pass http://nexahub_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";
            proxy_read_timeout 3600;  # 1h pour les WebSockets
        }
    }
}
```

---

### [OK] Bonnes Pratiques du Chapitre 21

> **Multi-stage Docker build** : l'image finale ne contient que le nécessaire (pas pip, pas les sources de compilation).

> **Jamais root dans le container.** Créez toujours un utilisateur dédié.

> **`--reload` ne sert qu'en développement.** En production : Gunicorn + Uvicorn workers.

> **Nombre de workers Gunicorn** : `(2 × CPU_cores) + 1`. Sur un serveur 2 cœurs -> 5 workers.

---

### [EFFORT] Exercice de Complétion — Chapitre 21

```bash
# [NOTE] Compléter les scripts de déploiement

# deploy.sh — Script de déploiement sur le serveur
#!/bin/bash
set -e  # Arrêter si une commande échoue

echo "[RAPIDE] Déploiement NexaHub..."

# [NOTE] 1. Pull les dernières images
docker compose pull

# [NOTE] 2. Appliquer les migrations Alembic AVANT de redémarrer l'app
# (pour éviter des downtime si la migration échoue)
docker compose run --rm api alembic ___

# [NOTE] 3. Redémarrer les services avec zero-downtime
# Utiliser --no-deps et redémarrer un service à la fois
docker compose up -d --no-deps ___

# [NOTE] 4. Vérifier que l'app est healthy
echo "[HOURGLASS_WITH_FLOWING_SAND] Attente du health check..."
sleep 10
curl -f http://localhost/health || (echo "[X] Health check échoué!" && exit 1)

echo "[OK] Déploiement réussi!"
```

---

## [LIVRE] Chapitre 22 : CI/CD avec GitHub Actions

### [OBJECTIF] Objectifs du chapitre
- Automatiser les tests à chaque push
- Builder et pusher l'image Docker automatiquement
- Déployer en production de façon sécurisée

---

### 22.1 — Pipeline CI/CD GitHub Actions

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

name: NexaHub CI/CD

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

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

jobs:
  # ── JOB 1 : Tests ──────────────────────────────────────────
  test:
    runs-on: ubuntu-latest

    services:
      # Service PostgreSQL pour les tests d'intégration
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: testpassword
          POSTGRES_DB: nexahub_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

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

      - name: Setup Python 3.11
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
          cache: "pip"  # Cache des dépendances pip

      - name: Install dependencies
        run: pip install -r requirements.txt -r requirements-dev.txt

      - name: Run linter (ruff)
        run: ruff check app/ tests/

      - name: Run type checker (mypy)
        run: mypy app/ --ignore-missing-imports

      - name: Run tests
        env:
          DATABASE_URL: postgresql+asyncpg://postgres:testpassword@localhost:5432/nexahub_test
          SECRET_KEY: test-secret-key-for-ci
          ENVIRONMENT: testing
        run: |
          pytest tests/ \
            --cov=app \
            --cov-report=xml \
            --cov-report=term-missing \
            -v \
            --timeout=60

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

  # ── JOB 2 : Build Docker ───────────────────────────────────
  build:
    needs: test  # Seulement si les tests passent
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    permissions:
      contents: read
      packages: write

    steps:
      - uses: actions/checkout@v4

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

      - name: Extract metadata for Docker
        id: meta
        uses: docker/metadata-action@v5
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
          tags: |
            type=sha,prefix=sha-
            type=raw,value=latest,enable={{is_default_branch}}

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

  # ── JOB 3 : Déploiement ───────────────────────────────────
  deploy:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment: production  # Environnement GitHub avec protection

    steps:
      - name: Deploy to production
        uses: appleboy/ssh-action@v1
        with:
          host: ${{ secrets.PRODUCTION_HOST }}
          username: ${{ secrets.PRODUCTION_USER }}
          key: ${{ secrets.PRODUCTION_SSH_KEY }}
          script: |
            cd /opt/nexahub
            docker compose pull
            docker compose run --rm api alembic upgrade head
            docker compose up -d
            docker system prune -f
```

---

### [OK] Bonnes Pratiques du Chapitre 22

> **Automatiser les tests avant tout déploiement.** Un déploiement sans tests automatiques est une bombe à retardement.

> **Utiliser les "environments" GitHub** pour les déploiements production avec validation manuelle optionnelle.

> **Jamais de secrets dans le code ou les fichiers de config.** Utilisez les GitHub Secrets.

---

# [SCIENCE] PARTIE 7 — Niveau Expert

---

## [LIVRE] Chapitre 23 & 24 : Architecture Avancée

### [OBJECTIF] Repository Pattern

```python
# app/repositories/base.py

from typing import TypeVar, Generic, Type, Optional, List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from app.db.session import Base

ModelType = TypeVar("ModelType", bound=Base)

class BaseRepository(Generic[ModelType]):
    """
    Repository de base avec les opérations CRUD génériques.
    Héritez pour chaque entité.
    """

    def __init__(self, model: Type[ModelType]):
        self.model = model

    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
        result = await db.execute(select(self.model).where(self.model.id == id))
        return result.scalar_one_or_none()

    async def get_multi(
        self,
        db: AsyncSession,
        skip: int = 0,
        limit: int = 20
    ) -> List[ModelType]:
        result = await db.execute(select(self.model).offset(skip).limit(limit))
        return result.scalars().all()

    async def create(self, db: AsyncSession, obj_in: dict) -> ModelType:
        db_obj = self.model(**obj_in)
        db.add(db_obj)
        await db.flush()
        await db.refresh(db_obj)
        return db_obj

    async def update(
        self,
        db: AsyncSession,
        db_obj: ModelType,
        obj_in: dict
    ) -> ModelType:
        for field, value in obj_in.items():
            setattr(db_obj, field, value)
        await db.flush()
        await db.refresh(db_obj)
        return db_obj

    async def delete(self, db: AsyncSession, id: int) -> bool:
        obj = await self.get(db, id)
        if obj:
            await db.delete(obj)
            return True
        return False

    async def count(self, db: AsyncSession) -> int:
        result = await db.execute(select(func.count(self.model.id)))
        return result.scalar()


# app/repositories/project_repository.py

from app.models.project import Project
from app.repositories.base import BaseRepository

class ProjectRepository(BaseRepository[Project]):
    """Repository spécialisé pour les projets."""

    async def get_by_owner(
        self,
        db: AsyncSession,
        owner_id: int,
        skip: int = 0,
        limit: int = 20
    ) -> List[Project]:
        result = await db.execute(
            select(Project)
            .where(Project.owner_id == owner_id)
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all()

# Instance singleton
project_repository = ProjectRepository(Project)
```

---

## [LIVRE] Chapitre 25 : Observabilité & Performance

### [OBJECTIF] Stress testing avec Locust

```python
# locustfile.py

from locust import HttpUser, task, between
import random

class NexaHubUser(HttpUser):
    wait_time = between(1, 3)  # Pause entre 1 et 3 secondes entre les tâches
    token = None

    def on_start(self):
        """Login au démarrage de chaque utilisateur virtuel."""
        response = self.client.post("/auth/login", json={
            "email": f"user{random.randint(1, 100)}@test.io",
            "password": "TestPassword123!"
        })
        if response.status_code == 200:
            self.token = response.json()["access_token"]

    def get_headers(self):
        return {"Authorization": f"Bearer {self.token}"}

    @task(3)  # Poids 3 : 3x plus probable que les tâches de poids 1
    def list_projects(self):
        self.client.get("/projects/", headers=self.get_headers())

    @task(2)
    def list_tasks(self):
        project_id = random.randint(1, 10)
        self.client.get(f"/projects/{project_id}/tasks", headers=self.get_headers())

    @task(1)
    def create_task(self):
        self.client.post("/tasks/", json={
            "title": f"Task {random.randint(1, 1000)}",
            "project_id": random.randint(1, 10),
            "priority": random.randint(1, 5)
        }, headers=self.get_headers())
```

```bash
# Lancer le stress test
locust -f locustfile.py --host=http://localhost:8000 --headless \
    -u 100 \    # 100 utilisateurs simultanés
    -r 10 \     # 10 nouveaux utilisateurs par seconde
    --run-time 60s \
    --html=report.html
```

---

## [LIVRE] Chapitre 26 : Bonnes Pratiques Finales

### [OBJECTIF] Checklist Expert NexaHub

Avant de déclarer votre API "production-ready", vérifiez chaque point :

#### [VERROUILLE] Sécurité
- [ ] Toutes les routes sensibles protégées par JWT
- [ ] Mots de passe hashés avec bcrypt
- [ ] CORS configuré avec des origines explicites
- [ ] Rate limiting activé
- [ ] Headers de sécurité HTTP en place
- [ ] Variables d'environnement pour tous les secrets
- [ ] `.env` dans `.gitignore`
- [ ] Logs ne contiennent aucun mot de passe ni token

#### [RAPIDE] Performance
- [ ] Toutes les routes I/O utilisent `async def`
- [ ] Requêtes indépendantes parallélisées avec `asyncio.gather`
- [ ] Cache Redis sur les données fréquemment lues
- [ ] Pagination sur toutes les listes
- [ ] Compression GZip activée
- [ ] Connexions DB avec pool configuré

#### [TEST] Qualité
- [ ] Couverture de tests > 80% sur les routes critiques
- [ ] Lint (ruff) et types (mypy) dans le CI
- [ ] Migrations Alembic pour tous les changements de schéma
- [ ] Documentation Swagger complète (summary + description sur chaque route)

#### [RAPIDE] Déploiement
- [ ] Dockerfile multi-stage
- [ ] Utilisateur non-root dans le container
- [ ] Health check configuré
- [ ] Gunicorn + Uvicorn workers
- [ ] Nginx comme reverse proxy
- [ ] CI/CD automatisé (tests -> build -> deploy)
- [ ] Sentry pour les alertes d'erreur

---

### [EFFORT] Exercice Final : Le Grand Projet

Vous avez maintenant toutes les clés. Voici votre défi final pour NexaHub :

```
[LISTE] MISSION : Implémenter ces 5 fonctionnalités de bout en bout

1. Système d'invitation à un projet
   - POST /projects/{id}/invite -> envoie un email avec un lien d'invitation
   - GET /invitations/{token}/accept -> rejoint le projet
   - L'invitation expire après 48h (JWT avec expiration courte)

2. Fil d'activité (Activity Feed)
   - Chaque action (créer tâche, changer statut, assigner) crée une entrée
   - GET /projects/{id}/activity -> fil d'activité paginé
   - Temps réel via WebSocket : nouveau événement -> broadcast

3. Recherche globale
   - GET /search?q=... -> cherche dans projets, tâches, utilisateurs
   - Résultats paginés et triés par pertinence
   - Cache Redis pour les requêtes fréquentes

4. Notifications avec préférences
   - Modèle UserNotificationPreferences : email_on_assign, email_daily_digest, etc.
   - PATCH /users/me/notifications -> modifier ses préférences
   - Le NotificationService respecte ces préférences

5. Tableau de bord analytics
   - GET /analytics/overview -> métriques globales de l'utilisateur
   - GET /analytics/projects/{id} -> burndown chart data
   - Cache Redis avec TTL 5 minutes
   - Export CSV et PDF du rapport

Pour chaque fonctionnalité :
[OK] Schémas Pydantic complets
[OK] Modèle ORM si nécessaire + migration Alembic
[OK] Service avec logique métier
[OK] Route documentée avec tags, summary, responses
[OK] Tests : cas nominal + 2 cas d'erreur minimum
[OK] Cache Redis si les données sont fréquemment lues
```

---

## [TROPHEE] Récapitulatif Complet du Parcours NexaHub

Félicitations ! Si vous avez suivi