1. Architecture REST
REST (Representational State Transfer) est un style architectural pour les API HTTP. Les 6 contraintes de REST garantissent une API scalable et maintenable.
| Contrainte REST | Application pratique |
|---|---|
| Stateless | Chaque requête contient toutes les infos nécessaires (token JWT) |
| Uniform Interface | URLs stables, verbes HTTP sémantiques, JSON standard |
| Client-Server | Frontend et backend séparés |
| Cacheable | Réponses GET cachées avec ETag / Cache-Control |
| Layered System | Proxy, CDN, load balancer transparents |
| Code on Demand | Optionnel — servir du JS (rare) |
# ── Conventions REST pour TaskAPI
# Ressources au pluriel, noms pas verbes
# GET /api/tasks → lister
# POST /api/tasks → créer
# GET /api/tasks/{id} → lire un
# PUT /api/tasks/{id} → remplacer entièrement
# PATCH /api/tasks/{id} → modification partielle
# DELETE /api/tasks/{id} → supprimer
# ── Codes HTTP sémantiques
# 200 OK → GET, PUT, PATCH réussis
# 201 Created → POST réussi
# 204 No Content → DELETE réussi
# 400 Bad Request → erreur client (validation)
# 401 Unauthorized → non authentifié
# 403 Forbidden → authentifié mais pas autorisé
# 404 Not Found → ressource inexistante
# 409 Conflict → état conflictuel (email déjà utilisé)
# 422 Unprocessable → erreur de validation Pydantic
# 429 Too Many Req → rate limiting
# 500 Internal → erreur serveur
2. JWT — Théorie
Un JSON Web Token est un jeton signé composé de 3 parties encodées en Base64 URL.
header.payload.signature
Exemple :
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9 ← header (algo + type)
.eyJzdWIiOiIxMjM0IiwiZXhwIjoxNzE2MDAwfQ ← payload (claims)
.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQ ← signature (HMAC-SHA256)
from __future__ import annotations
# ── Claims standards du payload JWT
payload = {
"sub": "42", # subject = user_id
"exp": 1716000000, # expiration timestamp
"iat": 1715990000, # issued at
"type": "access", # access | refresh
"email": "alice@test.com", # claim custom
}
# ── Deux types de tokens
# Access token : durée courte (15 min → 1h) — utilisé dans les headers
# Refresh token : durée longue (7j → 30j) — stocké en cookie httpOnly
# ── Stockage côté client (sécurité)
# ✅ localStorage : simple, mais vulnérable aux XSS
# ✅ cookie httpOnly : protégé des XSS, attention au CSRF
# ❌ Ne jamais stocker le mot de passe dans le token
3. OAuth2PasswordBearer
FastAPI fournit OAuth2PasswordBearer qui extrait automatiquement le Bearer token du header Authorization.
from __future__ import annotations
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
app = FastAPI()
# ── tokenUrl = endpoint qui délivre le token (/auth/login)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
# ── Endpoint de login : reçoit form-data (standard OAuth2)
@app.post("/auth/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()) -> dict:
# form_data.username et form_data.password
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Email ou mot de passe incorrect",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token({"sub": str(user.id)})
return {
"access_token": access_token,
"token_type": "bearer",
"expires_in": 3600,
}
# ── Route protégée
@app.get("/api/me")
async def get_me(token: str = Depends(oauth2_scheme)) -> dict:
# token = valeur extraite de "Authorization: Bearer "
payload = verify_token(token)
return {"user_id": payload["sub"]}
4. Hachage bcrypt
bcrypt est l'algorithme recommandé pour hacher les mots de passe. Il intègre un sel aléatoire et un facteur de coût adaptable.
from __future__ import annotations
from passlib.context import CryptContext
# ── Contexte passlib — supporte plusieurs algorithmes
pwd_context = CryptContext(
schemes=["bcrypt"],
deprecated="auto", # migration auto vers bcrypt si ancien algo
bcrypt__rounds=12, # facteur de coût (2^12 itérations)
)
def hash_password(plain_password: str) -> str:
"""Retourne le hash bcrypt du mot de passe."""
# Le sel est généré et inclus dans le hash automatiquement
return pwd_context.hash(plain_password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Vérifie un mot de passe contre son hash."""
return pwd_context.verify(plain_password, hashed_password)
# ── Usage
hashed = hash_password("MonMotDePasse123!")
print(hashed)
# $2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW
print(verify_password("MonMotDePasse123!", hashed)) # True
print(verify_password("mauvais_mdp", hashed)) # False
# ── JAMAIS stocker le mot de passe en clair
# ✅ db.save(hash_password(user.password))
# ❌ db.save(user.password)
5. get_current_user
La dépendance get_current_user est le gardien de toutes les routes protégées. Elle valide le token JWT et retourne l'utilisateur courant.
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from database import get_db
from models import User
SECRET_KEY = "votre-secret-key-tres-longue-et-aleatoire"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode["exp"] = expire
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db),
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token invalide ou expiré",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str | None = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = db.get(User, int(user_id))
if user is None or not user.is_active:
raise credentials_exception
return user
# ── Injection dans une route protégée
# @app.get("/api/me")
# async def get_me(current_user: User = Depends(get_current_user)) -> dict:
# return {"id": current_user.id, "email": current_user.email}
6. Middleware logging
Les middlewares FastAPI s'exécutent pour chaque requête, avant et après le handler de route.
from __future__ import annotations
import time
import uuid
import logging
from fastapi import FastAPI, Request, Response
from fastapi.middleware.base import BaseHTTPMiddleware
logger = logging.getLogger("taskapi")
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
# ── Avant la route
request_id = str(uuid.uuid4())[:8]
start = time.perf_counter()
# ── Exécuter la route
response = await call_next(request)
# ── Après la route
duration_ms = (time.perf_counter() - start) * 1000
response.headers["X-Request-ID"] = request_id
response.headers["X-Process-Time"] = f"{duration_ms:.2f}ms"
logger.info(
"%s %s %d %.2fms [%s]",
request.method, request.url.path,
response.status_code, duration_ms, request_id,
)
return response
app = FastAPI()
app.add_middleware(LoggingMiddleware)
7. CORS
Le Cross-Origin Resource Sharing permet à votre frontend (ex: React sur localhost:3000) d'appeler votre API (localhost:8000).
from __future__ import annotations
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# ── Configuration CORS
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000", # React dev
"http://localhost:5173", # Vite dev
"https://monapp.com", # Production
],
allow_credentials=True, # Autorise les cookies httpOnly
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
expose_headers=["X-Process-Time", "X-Request-ID"],
max_age=600, # Cache preflight 10 minutes
)
# ── En développement uniquement (à ne JAMAIS faire en prod)
# allow_origins=["*"] ← désactive toute protection CORS
8. Rate limiting
Le rate limiting protège votre API contre les attaques par force brute et le spam.
from __future__ import annotations
import time
from collections import defaultdict
from fastapi import FastAPI, HTTPException, Request, status
app = FastAPI()
# ── Rate limiter simple en mémoire (pour dev/test)
# En production : utiliser slowapi + Redis
_rate_limits: dict[str, list[float]] = defaultdict(list)
def rate_limit(max_requests: int = 10, window_seconds: int = 60):
"""Dépendance de rate limiting par IP."""
async def _check(request: Request) -> None:
client_ip = request.client.host if request.client else "unknown"
now = time.time()
key = f"{client_ip}:{request.url.path}"
# Purger les entrées hors de la fenêtre
_rate_limits[key] = [t for t in _rate_limits[key] if now - t < window_seconds]
if len(_rate_limits[key]) >= max_requests:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=f"Trop de requêtes. Max {max_requests}/{window_seconds}s",
headers={"Retry-After": str(window_seconds)},
)
_rate_limits[key].append(now)
return _check
from fastapi import Depends
# ── Appliquer sur la route de login (limite les tentatives brute-force)
@app.post("/auth/login")
async def login(
_=Depends(rate_limit(max_requests=5, window_seconds=60)),
) -> dict:
# max 5 tentatives de login par minute par IP
return {"token": "..."}
# ── En production avec slowapi :
# from slowapi import Limiter
# from slowapi.util import get_remote_address
# limiter = Limiter(key_func=get_remote_address)
# @app.post("/auth/login")
# @limiter.limit("5/minute")
# async def login(request: Request): ...