FastAPI

Programme complet de formation ��� De d��butant ��

expert

Python asynchrone �� REST �� PostgreSQL �� JWT �� Architecture �� Docker

Programme complet de formation FastAPI — De Débutant à Expert

Sommaire

Partie 1 — Fondations (niveau débutant)

1.1 Python utile pour FastAPI

1.2 HTTP, REST, JSON

1.3 Async / await en Python

1.4 Installation et structure d’un projet FastAPI

Exercices — Partie 1

Partie 4 — Base de données

4.1 SQLAlchemy (ORM)

4.2 PostgreSQL

4.3 Alembic (migrations)

4.4 CRUD complet avec base de données

Exercices — Partie 4

Partie 7 — Architecture avancée

7.1 Clean Architecture adaptée à FastAPI

7.2 Séparation des responsabilités

7.3 Pattern Repository

7.4 Pattern Service

Exercices — Partie 7

Projet 2 (Intermédiaire) — Plateforme de blog avec authentification

Projet 3 (Avancé, type production) — Système de chat en temps réel (mini clone Slack/Discord)
 Programme complet de formation FastAPI
— De Débutant à Expert

Ce programme est conçu pour t’emmener de zéro à la capacité de concevoir, sécuriser, tester et
déployer des APIs professionnelles avec FastAPI. Chaque partie contient : explications, exemples de
code concrets, et exercices pratiques.

Sommaire
1. Fondations
2. FastAPI de base
3. Niveau intermédiaire
4. Base de données
5. Niveau avancé (Auth JWT, sécurité)
6. Niveau professionnel (WebSockets, perf, tests)
7. Architecture avancée (Clean Architecture)
8. Déploiement (Docker, production)
9. Projets pratiques (3 projets progressifs)
10. Annexes : checklist, ressources

Prérequis : bases de programmation (variables, fonctions, boucles). Aucune expérience FastAPI

requise.

Durée indicative : 8 à 12 semaines à raison de 5-8h/semaine.

Partie 1 — Fondations (niveau débutant)

1.1 Python utile pour FastAPI

Tu n’as pas besoin de tout Python, mais ces notions sont indispensables :

Type hints (annotations de type)

FastAPI repose entièrement sur les type hints pour valider les données et générer la documentation
automatique.

def add(a: int, b: int) -> int:

return a + b

def greet(name: str, age: int | None = None) -> str:

if age:
return f"Bonjour {name}, tu as {age} ans"
return f"Bonjour {name}"

Dataclasses et classes

class User:
def __init__(self, name: str, email: str):
[Link] = name
[Link] = email

Décorateurs
FastAPI utilise massivement les décorateurs ( @[Link](...) , @[Link](...) ).

def log_call(func):
def wrapper(*args, **kwargs):
print(f"Appel de {func.__name__}")
return func(*args, **kwargs)
return wrapper

@log_call
def hello():
print("Hello")

Context managers ( with )

Utilisés pour gérer les ressources (connexions DB, fichiers).

with open("[Link]") as f:
contenu = [Link]()

Generators et yield

Essentiels pour les dépendances FastAPI (sessions de base de données).

 def get_numbers():
yield 1
yield 2
yield 3

Modules utiles
typing : Optional , List , Dict , Union
datetime : gestion des dates

enum : Enum pour les valeurs fixes (rôles, statuts)

pathlib : gestion des chemins de fichiers

1.2 HTTP, REST, JSON

HTTP en bref
HTTP est un protocole requête/réponse. Chaque requête a : - une méthode (GET, POST, PUT,
PATCH, DELETE) - une URL - des headers (métadonnées : Content-Type, Authorization…) - un
body (optionnel, souvent en JSON)

Chaque réponse a : - un status code (200, 201, 404, 500…) - des headers - un body

Codes de statut essentiels

Code Signification Usage

200 OK Succès général

201 Created Ressource créée

204 No Content Succès sans contenu (ex: DELETE)

400 Bad Request Données invalides

401 Unauthorized Non authentifié

403 Forbidden Authentifié mais pas autorisé

404 Not Found Ressource introuvable

422 Unprocessable Entity Erreur de validation (Pydantic)

500 Internal Server Error Erreur serveur

REST : principes clés

REST organise une API autour de ressources identifiées par des URLs, manipulées avec les
méthodes HTTP :

GET /users → Lister les utilisateurs

GET /users/{id} → Récupérer un utilisateur
POST /users → Créer un utilisateur
PUT /users/{id} → Remplacer un utilisateur
PATCH /users/{id} → Modifier partiellement
DELETE /users/{id} → Supprimer un utilisateur

Règles de bon sens : - Les URLs contiennent des noms (ressources), pas des verbes ( /users , pas
/getUsers ) - Utilise le pluriel ( /users , pas /user ) - Les relations s’expriment par imbrication
( /users/{id}/posts )
JSON
Format d’échange standard. FastAPI sérialise/désérialise automatiquement entre JSON et objets
Python via Pydantic.

{
"id": 1,
"name": "Alice",
"is_active": true,
"tags": ["admin", "beta"]
}

1.3 Async / await en Python

FastAPI est construit sur ASGI (Asynchronous Server Gateway Interface), contrairement à
Flask/Django classique (WSGI, synchrone).

Pourquoi l’async ?
Quand ton serveur attend une réponse I/O (base de données, appel réseau, fichier), le mode async
permet de traiter d’autres requêtes pendant l’attente, au lieu de bloquer le thread.

import asyncio

async def fetch_data():

print("Début")
await [Link](2) # simule un appel I/O
print("Fin")
return {"data": "ok"}

async def main():

result = await fetch_data()
print(result)

[Link](main())

Ce qui se passe réellement sous le capot

Python exécute un seul thread principal, piloté par une event loop (boucle d’événements). Quand
une coroutine rencontre un await sur une opération I/O, elle rend la main à la boucle, qui peut
alors exécuter une autre tâche en attendant. Ce n’est pas du parallélisme (pas de vrai multi-cœur),
c’est de la concurrence coopérative : très efficace pour des I/O (réseau, DB, fichiers), inutile pour
du calcul pur (CPU-bound), où un thread/process séparé reste nécessaire ( multiprocessing ,
[Link] ).

import asyncio, time

async def task(name: str, delay: int):

print(f"{name} démarre")
await [Link](delay) # rend la main pendant l'attente
print(f"{name} termine")

async def main():

start = [Link]()
# Exécution concurrente : les 3 tâches s'entrelacent
await [Link](task("A", 2), task("B", 1), task("C", 3))
print(f"Total: {[Link]() - start:.1f}s") # ~3s, pas 6s

[Link](main())
Sans [Link] (donc avec des await séquentiels), le total aurait été de 6 secondes
(2+1+3). C’est exactement ce mécanisme qui permet à FastAPI de gérer des milliers de requêtes
concurrentes avec peu de ressources : pendant qu’une requête attend une réponse de la base de
données, le serveur en traite d’autres.

Règles importantes
async def définit une coroutine
await ne peut être utilisé que dans une fonction async def

N’utilise jamais de code bloquant (ex: [Link] , requêtes requests synchrones) dans une
fonction async def — cela bloque toute la boucle d’événements, donc toutes les requêtes
en cours, pas seulement la tienne. C’est l’erreur la plus fréquente chez les débutants avec
FastAPI. Utilise [Link] , asyncpg , aiofiles , etc.
FastAPI accepte aussi bien les routes def (synchrones, exécutées automatiquement dans un
threadpool séparé, donc sans bloquer la boucle) que async def . Règle pratique : si ta fonction
ne fait aucun await à l’intérieur (ex: calcul pur, appel à une librairie synchrone), déclare-la en
def simple — FastAPI s’occupe de l’exécuter dans un thread à part. Si elle fait des appels I/O
asynchrones (DB async, HTTP async), déclare-la async def .
Erreur classique à éviter : mélanger une librairie synchrone bloquante ( requests , [Link] ,
un driver DB non-async) à l’intérieur d’une route async def — cela annule tout le bénéfice de
l’asynchrone.

Exemple concret dans FastAPI

@[Link]("/items")
async def get_items():
items = await db.fetch_all("SELECT * FROM items")
return items

1.4 Installation et structure d’un projet FastAPI

Installation

python -m venv venv

source venv/bin/activate # Windows: venv\Scripts\activate

pip install fastapi uvicorn[standard]

fastapi : le framework
uvicorn : le serveur ASGI qui exécute l’application

Premier fichier

# [Link]
from fastapi import FastAPI

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

@[Link]("/")
async def root():
return {"message": "Hello World"}

Lancer le serveur
 uvicorn main:app --reload

main = nom du fichier [Link]

app = nom de l’instance FastAPI

--reload = redémarre automatiquement à chaque modification (dev uniquement)

Accède à : - [Link] → ton API - [Link] → documentation

Swagger interactive (générée automatiquement !) - [Link] → documentation
ReDoc

Structure minimale d’un projet débutant

mon_projet/
├── venv/
├── [Link]
├── [Link]
└── .gitignore

pip freeze > [Link]

Exercices — Partie 1
1. Installe Python, crée un environnement virtuel, installe FastAPI et Uvicorn.
2. Crée une route GET /hello/{name} qui retourne {"message": f"Bonjour {name}"} .
3. Écris une fonction asynchrone wait_and_return(seconds: int) qui attend seconds secondes puis
retourne "terminé" . Teste-la avec [Link] .
4. Explique en 5 lignes la différence entre PUT et PATCH.
5. Liste 5 codes de statut HTTP et leur signification, sans les recopier depuis ce document.

# Partie 2 — FastAPI de base

# Partie 3 — Niveau intermédiaire

## 3.1 Organisation d’un projet propre

(architecture folder)

Dès que le projet grandit, un seul [Link]

devient ingérable. Structure recommandée :

mon_projet/ ├── app/ │ ├── __init__.py

│ ├── [Link] # point
d'entrée, création de l'app │ ├── core/
│ │ ├── [Link] # settings
(variables d'env) │ │ └── [Link]
# hashing, JWT │ ├── api/ │ │ ├──
__init__.py │ │ ├── [Link]
# dépendances communes (get_db,
get_current_user) │ │ └── routes/ │
│ ├── [Link] │ │ └──
[Link] │ ├── models/ #
modèles SQLAlchemy │ │ └── [Link] │
├── schemas/ # modèles
Pydantic │ │ └── [Link] │ ├── crud/
# logique d'accès aux données │ │ └──
[Link] │ └── db/ │ ├── [Link] │
└── [Link] ├── tests/ ├── alembic/ ├──
[Link] ├── .env └── docker-
[Link]

### Utiliser un APIRouter pour séparer les

routes ```python # app/api/routes/[Link]
from fastapi import APIRouter

router = APIRouter(prefix=“/users”, tags=

[“users”])

@[Link](“/”) async def list_users(): return

[]

@[Link](“/{user_id}”) async def

get_user(user_id: int): return {“id”: user_id}
```

```python # app/[Link] from fastapi import

FastAPI from [Link] import users,
items

app = FastAPI(title=“Mon API”)

app.include_router([Link])
app.include_router([Link]) ```

## 3.2 Dependency Injection

Le système de dépendances de FastAPI

( Depends ) permet de réutiliser de la logique
(auth, DB, pagination) proprement.

```python from fastapi import Depends

def get_query_params(skip: int = 0, limit: int

= 10): return {“skip”: skip, “limit”: limit}

@[Link](“/items”) async def

list_items(params: dict =
Depends(get_query_params)): return params
```

### Dépendance avec yield (typique pour

la DB) ```python def get_db(): db =
SessionLocal() try: yield db finally: [Link]()

@[Link](“/items”) async def list_items(db:

Session = Depends(get_db)): return
[Link](Item).all() ```

### Dépendances imbriquées ```python def

get_current_user(token: str =
Depends(oauth2_scheme)): … return user

def get_current_active_user(user: User =

Depends(get_current_user)): if not
user.is_active: raise HTTPException(400,
“Utilisateur inactif”) return user ```

## 3.3 Middleware

Un middleware intercepte toutes les

requêtes/réponses (logging, temps de
traitement, headers globaux).

```python import time from fastapi import

Request

@[Link](“http”) async def

add_process_time_header(request: Request,
call_next): start_time = [Link]() response
= await call_next(request) process_time =
[Link]() - start_time [Link][“X-
Process-Time”] = str(process_time) return
response ```

## 3.4 CORS

CORS (Cross-Origin Resource Sharing)

contrôle quels domaines front-end peuvent
appeler ton API depuis un navigateur.

```python from [Link]

import CORSMiddleware

app.add_middleware( CORSMiddleware,
allow_origins=[“[Link]
“[Link]
allow_credentials=True, allow_methods=[“*”],
allow_headers=[“*”], ) ```

⚠ Ne mets jamais allow_origins=["*"] avec

allow_credentials=True en production :
c’est une faille de sécurité.

## 3.5 Authentication basique

Avant le JWT complet (partie 5), voici

l’authentification HTTP Basic, utile pour
comprendre le principe.

```python from [Link] import

HTTPBasic, HTTPBasicCredentials import
secrets

security = HTTPBasic()

@[Link](“/admin”) async def

admin_area(credentials: HTTPBasicCredentials
= Depends(security)): correct_username =
secrets.compare_digest([Link],
“admin”) correct_password =
secrets.compare_digest([Link],
“secret”) if not (correct_username and
correct_password): raise
HTTPException(status_code=401,
detail=“Identifiants incorrects”, headers=
{“WWW-Authenticate”: “Basic”}) return
{“message”: “Bienvenue admin”} ```

secrets.compare_digest évite les attaques

par timing.

## Exercices — Partie 3
1. Réorganise ton projet de la Partie 2 selon
l’architecture folder proposée, avec un
APIRouter . 2. Crée une dépendance
get_pagination réutilisable ( skip , limit )
et applique-la à 2 routes différentes. 3. Ajoute
un middleware qui logge la méthode, l’URL et
le temps de réponse de chaque requête. 4.
Configure CORS pour n’autoriser que
[Link] . 5. Implémente une
route protégée par HTTP Basic Auth.
Partie 4 — Base de données

4.1 SQLAlchemy (ORM)

SQLAlchemy permet de manipuler la base de données avec des objets Python plutôt que du SQL
brut.

Installation

pip install sqlalchemy psycopg2-binary

Configuration de la connexion

# app/db/[Link]
from sqlalchemy import create_engine
from [Link] import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base déclarative et modèle

# app/db/[Link]
from [Link] import declarative_base

Base = declarative_base()
 # app/models/[Link]
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from [Link] import relationship
from [Link] import func
from [Link] import Base

class User(Base):
__tablename__ = "users"

id = Column(Integer, primary_key=True, index=True)

username = Column(String(50), unique=True, index=True, nullable=False)
email = Column(String(100), unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=[Link]())

posts = relationship("Post", back_populates="owner")

class Post(Base):
__tablename__ = "posts"

id = Column(Integer, primary_key=True, index=True)

title = Column(String(200), nullable=False)
content = Column(String)
owner_id = Column(Integer, ForeignKey("[Link]"))

owner = relationship("User", back_populates="posts")

4.2 PostgreSQL

Lancer PostgreSQL localement avec Docker (le plus simple)

docker run --name pg-dev -e POSTGRES_PASSWORD=password -e POSTGRES_DB=mydb -p 5432:5432 -d

postgres:16

Bonnes pratiques
Toujours indexer les colonnes utilisées dans les WHERE fréquents ( index=True )
Utiliser des contraintes unique=True pour email/username
Ne jamais stocker de mot de passe en clair

4.3 Alembic (migrations)

Alembic gère les évolutions du schéma de base de données de façon versionnée (comme Git pour
ta DB).

Installation et init

pip install alembic

alembic init alembic

Configuration ( [Link] et [Link] )

Dans alembic/[Link] , importe tes modèles :
 from [Link] import Base
from [Link] import User
from [Link] import Post

target_metadata = [Link]

Et configure l’URL de connexion via une variable d’environnement plutôt qu’en dur.

Créer et appliquer une migration

alembic revision --autogenerate -m "create users and posts tables"

alembic upgrade head

Revenir en arrière

alembic downgrade -1

4.4 CRUD complet avec base de données

Schémas Pydantic

# app/schemas/[Link]
from pydantic import BaseModel, EmailStr, ConfigDict

class UserBase(BaseModel):
username: str
email: EmailStr

class UserCreate(UserBase):
password: str

class UserOut(UserBase):
id: int
is_active: bool

model_config = ConfigDict(from_attributes=True) # permet de lire depuis un objet SQLAlchemy

Couche CRUD
 # app/crud/[Link]
from [Link] import Session
from [Link] import User
from [Link] import UserCreate
from [Link] import hash_password

def get_user(db: Session, user_id: int) -> User | None:

return [Link](User).filter([Link] == user_id).first()

def get_user_by_email(db: Session, email: str) -> User | None:

return [Link](User).filter([Link] == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100) -> list[User]:
return [Link](User).offset(skip).limit(limit).all()

def create_user(db: Session, user_in: UserCreate) -> User:

db_user = User(
username=user_in.username,
email=user_in.email,
hashed_password=hash_password(user_in.password),
)
[Link](db_user)
[Link]()
[Link](db_user)
return db_user

def delete_user(db: Session, user_id: int) -> bool:

user = get_user(db, user_id)
if not user:
return False
[Link](user)
[Link]()
return True

Routes utilisant la DB
 # app/api/routes/[Link]
from fastapi import APIRouter, Depends, HTTPException
from [Link] import Session
from [Link] import get_db
from [Link] import UserCreate, UserOut
from [Link] import user as crud_user

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

@[Link]("/", response_model=UserOut, status_code=201)

def create_user(user_in: UserCreate, db: Session = Depends(get_db)):
if crud_user.get_user_by_email(db, user_in.email):
raise HTTPException(400, "Email déjà utilisé")
return crud_user.create_user(db, user_in)

@[Link]("/{user_id}", response_model=UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
user = crud_user.get_user(db, user_id)
if not user:
raise HTTPException(404, "Utilisateur non trouvé")
return user

@[Link]("/", response_model=list[UserOut])
def list_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
return crud_user.get_users(db, skip, limit)

@[Link]("/{user_id}", status_code=204)
def remove_user(user_id: int, db: Session = Depends(get_db)):
if not crud_user.delete_user(db, user_id):
raise HTTPException(404, "Utilisateur non trouvé")

Exercices — Partie 4
1. Installe PostgreSQL via Docker et connecte-le à un projet FastAPI.
2. Crée les modèles User et Post avec une relation one-to-many.
3. Initialise Alembic et génère la première migration.
4. Implémente un CRUD complet pour Post (create, read, update, delete, list avec pagination).
5. Ajoute une contrainte : un utilisateur ne peut pas créer deux posts avec le même titre
(contrainte unique composite ou validation applicative).

# Partie 5 — Niveau avancé (Sécurité & Auth

JWT)

# Partie 6 — Niveau professionnel

## 6.1 WebSockets

Pour des fonctionnalités temps réel (chat,

notifications).

### Différence avec HTTP classique En HTTP

classique, chaque échange suit le cycle requête →
réponse → connexion fermée. Pour du temps réel, il
faudrait que le client interroge sans arrêt le serveur
(“polling”), ce qui gaspille des ressources et introduit
de la latence.

Un WebSocket ouvre une connexion

bidirectionnelle persistante entre client et
serveur : une fois la connexion établie, les deux
parties peuvent s’envoyer des messages à tout
moment, sans réouvrir de connexion. C’est ce qui
rend possible le chat en temps réel, les notifications
live, ou les tableaux de bord qui se mettent à jour
automatiquement.

### Cycle de vie d’une connexion WebSocket 1. Le

client initie une requête HTTP spéciale ( Upgrade:
websocket ) 2. Le serveur accepte ( await
[Link]() ) et la connexion bascule en
mode WebSocket 3. Les deux parties peuvent
envoyer/recevoir des messages via
send_text / receive_text (ou
send_json / receive_json ) dans une boucle 4. La
connexion reste ouverte jusqu’à ce qu’une des
parties la ferme, ou jusqu’à une déconnexion réseau
( WebSocketDisconnect )

Contrairement aux routes HTTP classiques, une route

WebSocket doit gérer elle-même sa boucle de vie
(généralement un while True qui écoute les
messages entrants), et il faut explicitement nettoyer
les ressources (retirer le client des connexions
actives) à la déconnexion.

```python from fastapi import WebSocket,

WebSocketDisconnect

class ConnectionManager: def init(self):

self.active_connections: listWebSocket = []

async def connect(self, websocket: WebSocket):

await [Link]()
self.active_connections.append(websocket)

def disconnect(self, websocket: WebSocket):

self.active_connections.remove(websocket)

async def broadcast(self, message: str): for

connection in self.active_connections: await
connection.send_text(message)

manager = ConnectionManager()

@[Link](“/ws/{client_id}”) async def

websocket_endpoint(websocket: WebSocket,
client_id: str): await [Link](websocket)
try: while True: data = await
websocket.receive_text() await
[Link](f”{client_id}: {data}“) except
WebSocketDisconnect:
[Link](websocket) ```

## 6.2 Background tasks

Pour exécuter du code après avoir renvoyé la

réponse (ex: envoi d’email) sans bloquer le client.

```python from fastapi import BackgroundTasks

def send_welcome_email(email: str): # logique

d’envoi d’email print(f”Email envoyé à {email}“)
@[Link](“/register”) def register(user_in:
UserCreate, background_tasks: BackgroundTasks,
db: Session = Depends(get_db)): user =
crud_user.create_user(db, user_in)
background_tasks.add_task(send_welcome_email,
[Link]) return user ```

Pour des tâches lourdes/longues ou qui doivent

survivre à un redémarrage du serveur, utilise plutôt
Celery ou ARQ avec Redis comme broker.

## 6.3 Performance et optimisation

- Connexions DB : utilise un pool de connexions

( pool_size , max_overflow dans SQLAlchemy) -
N+1 queries : utilise joinedload / selectinload
pour charger les relations en une requête ```python
from [Link] import selectinload

[Link](User).options(selectinload([Link])).all()
`` - **Requêtes async** : utilise asyncpg +
SQLAlchemy async pour un vrai gain de perf sous
forte charge - **Profiling** : py-spy ,
middleware de timing (partie 3) - **Uvicorn
workers** : en production, lance plusieurs
workers avec Gunicorn+Uvicorn ( gunicorn -k
[Link]`)

## 6.4 Caching

Avec Redis pour éviter de recalculer/requêter

inutilement.

bash pip install redis

```python import redis import json

r = [Link](host=“localhost”, port=6379, db=0)

@[Link](“/items/{item_id}”) def
get_item(item_id: int, db: Session =
Depends(get_db)): cache_key = f”item:{item_id}”
cached = [Link](cache_key) if cached: return
[Link](cached)

item = crud_item.get_item(db, item_id)

[Link](cache_key, 300, [Link](item)) # cache 5
minutes return item ```

Pense à invalider le cache lors des updates/deletes.

## 6.5 Pagination et filtres avancés

```python from pydantic import BaseModel

class PaginatedResponse(BaseModel): total: int skip:

int limit: int items: list[UserOut]

@[Link](“/users”,
response_model=PaginatedResponse) def list_users(
skip: int = 0, limit: int = 20, search: str | None =
None, is_active: bool | None = None, db: Session =
Depends(get_db), ): query = [Link](User) if
search: query = [Link]([Link](f”%
{search}%“)) if is_active is not None: query =
[Link](User.is_active == is_active)

total = [Link]() items =

[Link](skip).limit(limit).all() return {“total”:
total, “skip”: skip, “limit”: limit, “items”: items} ```

## 6.6 Tests (pytest)

bash pip install pytest httpx pytest-asyncio

### Configuration d’une DB de test ```python #

tests/[Link] import pytest from sqlalchemy
import create_engine from [Link] import
sessionmaker from [Link] import
TestClient from [Link] import app from
[Link] import Base from [Link] import
get_db

TEST_DATABASE_URL =
“postgresql://user:password@localhost:5432/test_db”
engine = create_engine(TEST_DATABASE_URL)
TestingSessionLocal = sessionmaker(bind=engine)

@[Link](scope=“function”) def db_session():

[Link].create_all(bind=engine) session =
TestingSessionLocal() yield session [Link]()
[Link].drop_all(bind=engine)

@[Link](scope=“function”) def
client(db_session): def override_get_db(): yield
db_session app.dependency_overrides[get_db] =
override_get_db yield TestClient(app)
app.dependency_overrides.clear() ```

### Exemple de test ```python #

tests/test_users.py def test_create_user(client):
response = [Link](“/users/”, json={ “username”:
“alice”, “email”: “alice@[Link]”, “password”:
“secret123” }) assert response.status_code == 201
data = [Link]() assert data[“username”] ==
“alice” assert “password” not in data

def test_get_user_not_found(client): response =

[Link](“/users/999”) assert response.status_code
== 404

def test_login_success(client):
[Link](“/auth/register”, json={ “username”:
“bob”, “email”: “bob@[Link]”, “password”:
“secret123” }) response = [Link](“/auth/login”,
data={“username”: “bob@[Link]”, “password”:
“secret123”}) assert response.status_code == 200
assert “access_token” in [Link]() ```

### Lancer les tests

## Exercices — Partie 6

1. Ajoute un endpoint WebSocket /ws/chat/{room}

qui broadcast les messages entre clients d’une
même room. 2. Utilise BackgroundTasks pour logger
l’inscription d’un utilisateur dans un fichier. 3. Mets
en cache Redis les résultats de GET /items pendant
60 secondes. 4. Implémente la pagination + filtres
sur GET /posts (recherche par titre, filtre par
auteur). 5. Écris au moins 10 tests pytest couvrant :
register, login, CRUD posts, cas d’erreur (404, 401,
403).
Partie 7 — Architecture avancée

7.1 Clean Architecture adaptée à FastAPI

L’idée : isoler la logique métier (business logic) des détails techniques (framework web, base de
données). Cela rend le code testable et remplaçable.

Pourquoi s’embêter avec ça ?

Sur un petit projet (Projet 1), mettre toute la logique directement dans les routes fonctionne très
bien. Le problème apparaît quand le projet grandit : - Une règle métier (“un post ne peut être
supprimé que par son auteur”) se retrouve dupliquée dans plusieurs routes qui font la même action
- Tester une règle métier oblige à démarrer tout FastAPI et une vraie base de données, ce qui rend
les tests lents et fragiles - Remplacer SQLAlchemy par un autre ORM, ou ajouter une deuxième
interface (ex: une commande CLI en plus de l’API HTTP) devient un chantier énorme parce que la
logique est mélangée avec le code HTTP et le code SQL

La Clean Architecture (popularisée par Robert C. Martin) répond à ça avec un principe simple : les
dépendances pointent vers l’intérieur. Le cœur du système (la logique métier) ne doit rien
savoir de FastAPI, ni de SQLAlchemy, ni de PostgreSQL. Ce sont ces couches externes qui
dépendent du métier, jamais l’inverse.

Sur un projet FastAPI de taille raisonnable, on n’applique pas la Clean Architecture dans toute sa
rigueur académique (avec interfaces abstraites partout) — on en garde l’esprit pratique via un
découpage Route → Service → Repository → Model, largement suffisant et déjà un vrai gain de
maintenabilité.

Couches

┌─────────────────────────────┐
│ API (routes FastAPI) │ ← reçoit HTTP, appelle les services
├─────────────────────────────┤
│ Services (logique métier) │ ← règles business, orchestration
├─────────────────────────────┤
│ Repositories (accès data) │ ← abstraction de la DB
├─────────────────────────────┤
│ Models (SQLAlchemy) │ ← persistance
└─────────────────────────────┘

Règle d’or : les couches supérieures dépendent des couches inférieures, jamais l’inverse. Les
routes ne connaissent jamais SQLAlchemy directement.

7.2 Séparation des responsabilités

app/
├── api/
│ └── routes/
│ └── [Link] # HTTP only : parsing requête, appel service, formatage réponse
├── services/
│ └── post_service.py # règles métier : "un post ne peut être publié que par son auteur"
├── repositories/
│ └── post_repository.py # accès DB : create, get, update, delete (aucune logique métier)
├── models/
│ └── [Link]
└── schemas/
└── [Link]
 7.3 Pattern Repository

# app/repositories/post_repository.py
from [Link] import Session
from [Link] import Post

class PostRepository:
def __init__(self, db: Session):
[Link] = db

def get(self, post_id: int) -> Post | None:

return [Link](Post).filter([Link] == post_id).first()

def list(self, skip: int = 0, limit: int = 20) -> list[Post]:

return [Link](Post).offset(skip).limit(limit).all()

def create(self, post: Post) -> Post:

[Link](post)
[Link]()
[Link](post)
return post

def delete(self, post: Post) -> None:

[Link](post)
[Link]()

7.4 Pattern Service

# app/services/post_service.py
from fastapi import HTTPException
from [Link].post_repository import PostRepository
from [Link] import PostCreate
from [Link] import Post
from [Link] import User

class PostService:
def __init__(self, repo: PostRepository):
[Link] = repo

def create_post(self, post_in: PostCreate, author: User) -> Post:

post = Post(title=post_in.title, content=post_in.content, owner_id=[Link])
return [Link](post)

def delete_post(self, post_id: int, current_user: User) -> None:

post = [Link](post_id)
if not post:
raise HTTPException(404, "Post introuvable")
if post.owner_id != current_user.id and current_user.role != "admin":
raise HTTPException(403, "Vous ne pouvez pas supprimer ce post")
[Link](post)

Route qui orchestre tout

 # app/api/routes/[Link]
from fastapi import APIRouter, Depends
from [Link] import Session
from [Link] import get_db, get_current_user
from [Link].post_repository import PostRepository
from [Link].post_service import PostService
from [Link] import PostCreate, PostOut

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

def get_post_service(db: Session = Depends(get_db)) -> PostService:

return PostService(PostRepository(db))

@[Link]("/", response_model=PostOut, status_code=201)

def create_post(
post_in: PostCreate,
service: PostService = Depends(get_post_service),
current_user=Depends(get_current_user),
):
return service.create_post(post_in, current_user)

@[Link]("/{post_id}", status_code=204)
def delete_post(
post_id: int,
service: PostService = Depends(get_post_service),
current_user=Depends(get_current_user),
):
service.delete_post(post_id, current_user)

Avantages
Testable : tu peux tester PostService sans FastAPI ni vraie DB (mock du repository)
Remplaçable : changer de DB ou d’ORM n’impacte que la couche repository
Lisible : chaque fichier a une seule responsabilité

Exercices — Partie 7
1. Refactore ton CRUD User (partie 4-5) selon le pattern Repository + Service.
2. Écris un test unitaire de PostService.delete_post avec un PostRepository mocké (sans DB
réelle).
3. Ajoute une règle métier : un utilisateur ne peut pas créer plus de 5 posts par jour (logique dans
le service, pas dans la route).
4. Documente dans un [Link] les responsabilités de chaque couche de ton projet.

# Partie 8 —
Déploiement

# Partie 9 — Projets
pratiques

## Projet 1 (Simple)
— API de gestion de
tâches (Todo App)

Objectif : consolider
les parties 1 à 4
(routes, Pydantic,
DB, CRUD).

### Architecture
 ### Modèle de
données

### Endpoints |
Méthode | URL |
Description | |———|
—–|————–| | POST
| /tasks | Créer une
tâche | | GET |
/tasks | Lister
(avec filtre is_done ,
priority ) | | GET |
/tasks/{id} |
Détail | | PUT |
/tasks/{id} |
Modifier | | PATCH |
/tasks/{id}/toggle
| Marquer fait/non
fait | | DELETE |
/tasks/{id} |
Supprimer |

### Étapes de
construction 1. Init
projet + venv +
FastAPI +
PostgreSQL (Docker)
2. Modèle Task +
migration Alembic 3.
Schémas Pydantic
( TaskCreate ,
TaskUpdate ,
TaskOut ) 4. CRUD
complet 5. Filtres par
is_done et
priority 6. Tests
pytest de base 7.
Dockerisation

### Fonctionnalités
à implémenter -
Pagination simple
( skip / limit ) - Tri
par date de création
ou priorité -
Validation : title
non vide, max 200
caractères

Projet 2 (Intermédiaire) — Plateforme de blog avec

authentification
Objectif : consolider les parties 3 à 7 (auth JWT, rôles, architecture propre, tests).

Architecture
 blog_api/
├── app/
│ ├── [Link]
│ ├── core/ ([Link], [Link])
│ ├── db/
│ ├── models/ ([Link], [Link], [Link], [Link])
│ ├── schemas/
│ ├── repositories/
│ ├── services/
│ ├── api/
│ │ ├── [Link]
│ │ └── routes/ ([Link], [Link], [Link], [Link])
├── tests/
├── alembic/
└── [Link]

Modèles de base de données

class User(Base):
id, username, email, hashed_password, role (user/admin), is_active, created_at

class Category(Base):
id, name (unique)

class Post(Base):
id, title, content, slug (unique), published (bool),
owner_id (FK User), category_id (FK Category), created_at, updated_at

class Comment(Base):
id, content, author_id (FK User), post_id (FK Post), created_at

Relations : User 1--N Post , User 1--N Comment , Post 1--N Comment , Category 1--N Post .

Endpoints
 Méthode URL Auth Description

POST /auth/register non Inscription

POST /auth/login non Connexion (JWT)

GET /auth/me oui Profil courant

Liste (filtres : catégorie, auteur,

GET /posts non
recherche, pagination)

GET /posts/{slug} non Détail

POST /posts oui Créer (auteur = user connecté)

oui
PUT /posts/{id} Modifier
(owner/admin)

oui
DELETE /posts/{id} Supprimer
(owner/admin)

POST /posts/{id}/comments oui Commenter

GET /posts/{id}/comments non Lister les commentaires

oui
DELETE /comments/{id} Supprimer un commentaire
(owner/admin)

GET /categories non Lister

POST /categories oui (admin) Créer une catégorie

Étapes de construction
1. Setup projet avec architecture Clean (repositories/services)
2. Modèles + migrations pour toutes les entités
3. Auth JWT complète (register/login/me) + hashing
4. Rôles user/admin + dépendance require_admin
5. CRUD Posts avec règles métier (seul l’auteur ou un admin peut modifier/supprimer)
6. CRUD Comments avec mêmes règles
7. Filtres avancés + pagination sur /posts (recherche full-text simple avec ilike )
8. Tests pytest (auth, permissions, CRUD, cas d’erreur)
9. Dockerisation + CI GitHub Actions

Fonctionnalités à implémenter
Génération automatique du slug à partir du titre
Un post en published=False n’est visible que par son auteur/admin
Rate limiting sur /auth/login
Cache Redis sur GET /posts (liste publique)

Projet 3 (Avancé, type production) — Système de chat en

temps réel (mini clone Slack/Discord)
Objectif : consolider l’ensemble du programme — architecture Clean, WebSockets, background
tasks, cache, tests, déploiement.
Fonctionnalités globales
Authentification JWT (access + refresh tokens)
Serveurs (workspaces) et channels (comme Discord/Slack)
Messages en temps réel via WebSocket
Rôles par serveur (owner, admin, member)
Notifications (background tasks + cache des compteurs non lus)
Historique des messages paginé
Upload de fichiers/images (stockage local ou S3-compatible)
Statut en ligne/hors ligne des utilisateurs

Architecture

chat_api/
├── app/
│ ├── [Link]
│ ├── core/ ([Link], [Link], redis_client.py)
│ ├── db/
│ ├── models/
│ │ ├── [Link]
│ │ ├── [Link] # workspace
│ │ ├── [Link] # user <-> server avec rôle
│ │ ├── [Link]
│ │ ├── [Link]
│ │ └── [Link]
│ ├── schemas/
│ ├── repositories/
│ ├── services/
│ ├── ws/
│ │ ├── connection_manager.py
│ │ └── chat_ws.py
│ ├── api/
│ │ ├── [Link]
│ │ └── routes/ ([Link], [Link], [Link], [Link], [Link])
│ └── tasks/
│ └── [Link]
├── tests/
├── alembic/
├── Dockerfile
└── [Link]

Modèles de base de données

 class User(Base):
id, username, email, hashed_password, avatar_url, status (online/offline), created_at

class Server(Base):
id, name, owner_id (FK User), created_at

class Membership(Base):
id, user_id (FK User), server_id (FK Server), role (owner/admin/member), joined_at
# contrainte unique (user_id, server_id)

class Channel(Base):
id, name, server_id (FK Server), is_private (bool), created_at

class Message(Base):
id, content, author_id (FK User), channel_id (FK Channel), created_at, edited_at (nullable)

class Attachment(Base):
id, message_id (FK Message), file_url, file_type, size

Endpoints REST

Méthode URL Description

POST /auth/register Inscription

POST /auth/login Connexion (access + refresh token)

POST /auth/refresh Renouveler l’access token

POST /servers Créer un serveur (créateur = owner)

GET /servers Lister mes serveurs

POST /servers/{id}/join Rejoindre un serveur

POST /servers/{id}/channels Créer un channel (admin/owner)

GET /servers/{id}/channels Lister les channels

GET /channels/{id}/messages Historique paginé des messages

Envoyer un message (fallback REST,

POST /channels/{id}/messages
ou via WS)

POST /channels/{id}/attachments Upload de fichier

PATCH /servers/{id}/members/{user_id} Changer le rôle d’un membre (owner)

DELETE /servers/{id}/members/{user_id} Exclure un membre (admin/owner)

WebSocket

WS /ws/channels/{channel_id}?token={jwt}

Authentification du WebSocket via le token JWT en query param (à valider avant accept() )
Vérifier que l’utilisateur est bien membre du serveur du channel avant d’accepter la connexion
Diffuser les nouveaux messages à tous les clients connectés à ce channel
Gérer la déconnexion propre (retirer de la liste des connexions actives, mettre à jour le statut)

Étapes de construction (ordre recommandé)

1. Semaine 1 : setup projet, architecture Clean, modèles + migrations complets
2. Semaine 2 : Auth JWT complète avec refresh tokens, tests d’auth
3. Semaine 3 : CRUD Servers/Memberships/Channels avec permissions par rôle
4. Semaine 4 : Messages en REST (historique paginé) + tests
5. Semaine 5 : WebSocket temps réel (connection manager, auth WS, broadcast par channel)
6. Semaine 6 : Upload de fichiers, notifications (background tasks), cache Redis (compteurs non
lus, présence en ligne)
7. Semaine 7 : Tests d’intégration complets (y compris WebSocket avec TestClient ),
durcissement sécurité (rate limiting, validation stricte)
8. Semaine 8 : Dockerisation complète, CI/CD, déploiement sur un VPS ou cloud avec Nginx +
HTTPS

Points de sécurité spécifiques à ce projet

Vérifier l’appartenance au serveur/channel à chaque action (REST et WS), pas seulement à la
connexion
Limiter la taille des uploads et valider les types de fichiers (whitelist d’extensions/MIME types)
Rate limiting sur l’envoi de messages (anti-spam)
Ne jamais faire confiance au channel_id envoyé par le client sans revalider les permissions côté
serveur

# Partie 10 — Annexes

## 10.1 Checklist “API prête

pour la production”

- [ ] Toutes les routes ont des

schémas Pydantic
( response_model inclus) - [ ]
Authentification JWT
fonctionnelle avec expiration -
[ ] Mots de passe hashés avec
bcrypt - [ ] Permissions/rôles
vérifiés sur les routes
sensibles - [ ] CORS configuré
de façon restrictive - [ ]
Erreurs gérées proprement
(pas de stack trace exposée
au client) - [ ] Variables
sensibles en .env , jamais
commit - [ ] Migrations
Alembic à jour et versionnées
- [ ] Tests pytest couvrant les
cas critiques (auth,
permissions, erreurs) - [ ] Rate
limiting sur les endpoints
sensibles (login, register) - [ ]
Logs structurés + monitoring
des erreurs (Sentry ou
équivalent) - [ ] Healthcheck
endpoint - [ ] Dockerfile +
docker-compose fonctionnels -
[ ] CI qui lance les tests
automatiquement - [ ] HTTPS
activé en production - [ ]
Documentation /docs
cohérente et à jour

## 10.2 Ordre de progression

recommandé

1. Parties 1-2 : bases, tu dois

savoir créer des routes et
valider des données sans
réfléchir 2. Partie 3-4 :
architecture propre +
première vraie base de
données → Projet 1 3. Partie
5 : sécurité et JWT, sans
compromis → refactore le
Projet 1 avec auth si tu veux
t’entraîner 4. Partie 6-7 :
professionnalise ton code
(tests, perf, architecture) →
Projet 2 5. Partie 8 :
apprends à déployer ce que tu
construis, à chaque projet 6.
Projet 3 : synthèse complète,
à traiter comme un vrai projet
professionnel (git, issues, PR,
CI/CD)

## 10.3 Ressources
complémentaires

- Documentation officielle
FastAPI :
[Link] (la
meilleure source, très bien
écrite) - Documentation
SQLAlchemy 2.0 :
[Link] -
Documentation Alembic :
[Link]
- Documentation Pydantic v2 :
[Link] -
OWASP API Security Top 10
(sécurité des APIs) :
[Link]
project-api-security/

## 10.4 Prochaines étapes

après ce programme

- GraphQL avec Strawberry

(alternative à REST) - Event-
driven architecture (Kafka,
RabbitMQ) pour des systèmes
distribués - Microservices et
communication inter-services
(gRPC) - Observabilité
avancée (OpenTelemetry,
tracing distribué) - Kubernetes
pour l’orchestration à grande
échelle
Fin du programme. La meilleure façon de progresser : construire les 3 projets dans l’ordre, sans
copier-coller aveuglément — comprends chaque ligne, casse volontairement ton code pour
comprendre les erreurs, et lis systématiquement la documentation officielle quand un point n’est
pas clair.