Système utilisateur — inscription, connexion, authentification JWT

Dans ce chapitre, vous allez découvrir le mécanisme d'authentification JWT de FastAPI et comprendre ses différences avec l'authentification par session de Django/Flask.

JWT vs authentification par session

Django et Flask utilisentl'authentification par sessionLe serveur stocke l'état de l'utilisateur, et l'ID de session est stocké dans le cookie.

FastAPI recommandel'authentification JWT: Une fois l'utilisateur connecté, il obtient un jeton. Les requêtes suivantes peuvent transmettre le jeton pour l'authentification.

FonctionnalitésSession（Django/Flask）JWT（FastAPI）
Emplacement de stockageCôté serveur (mémoire/base de données)Côté client (Cookie/LocalStorage)
ExtensibilitéLe serveur doit partager la sessionSans état, prend naturellement en charge les systèmes distribués
Scénarios applicablesSites web rendus côté serveurAPI + SPA / Mobile
Contrôle de l'expirationLe serveur peut l'invalider à tout momentValide pendant toute la période de validité du jeton

Installer les dépendances

(venv) $ pip install passlib python-jose python-multipart
  • passlib: Hachage de mot de passe (analogue à werkzeug.security)
  • python-jose: Génération et vérification de JWT
  • python-multipart: Gérer la soumission de formulaire (formulaire de connexion OAuth2)

Définir le modèle User

Exemple

# Chemin du fichier : models.py, nouveaux ajouts
from passlib.context import CryptContext

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

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    hashed_password = Column(String(256), nullable=False)

    def set_password(self, password: str):
        """Hacher le mot de passe"""
        self.hashed_password = pwd_context.hash(password)

    def verify_password(self, password: str) -> bool:
        """Vérifier le mot de passe"""
        return pwd_context.verify(password, self.hashed_password)

Générer et exécuter les migrations :

(venv) $ alembic revision --autogenerate -m "新增 User 模型"
(venv) $ alembic upgrade head

Fonctions utilitaires JWT

Exemple

# Chemin du fichier : auth.py (nouveau fichier)
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

# Clé secrète (lue depuis les variables d'environnement en production)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24  # Durée de validité du jeton : 24 heures

# OAuth2PasswordBearer indique à FastAPI : fromRequest headers Authorization: Bearer <token> Mediummention récupérer le JWT
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict) -> str:
    """Générer un jeton JWT"""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_access_token(token: str) -> dict | Aucune:
    """Vérifier et décoder le jeton JWT"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

Enregistrement des routes

Exemple

# Chemin du fichier : routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token

router = APIRouter(prefix="/users", tags=["User"])
templates = Jinja2Templates(directory="templates")

@router.post("/register", response_model=UserResponse)
def register(
    username: str = Form(...),
    email: str = Form(...),
    password: str = Form(...),
    db: Session = Depends(get_db)
):
    """Inscription de l'utilisateur"""
    # Vérifier si le nom d'utilisateur et l'e-mail existent déjà
    if db.query(User).filter(User.username == username).first():
        raise HTTPException(status_code=400, detail="Nom d'utilisateur déjà utilisé")
    if db.query(User).filter(User.email == email).first():
        raise HTTPException(status_code=400, detail=E-mail déjà enregistré)

    user = User(username=username, email=email)
    user.set_password(password)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

Route de connexion (émission de JWT)

Exemple

# Ajouter à la fin du fichier : routers/users.py
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token

@router.post("/token")
def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    """
Se connecter et obtenir un Token
Utilisez OAuth2PasswordRequestForm : les noms des champs sont username et password (format formulaire).
    """
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not user.verify_password(form_data.password):
        raise HTTPException(
            status_code=401,
            detail="Nom d'utilisateur ou mot de passe invalide",
            headers={"WWW-Authenticate": "Bearer"}
        )

    # Émettre le JWT : contient uniquement user_id (ne doit pas contenir d'informations sensibles)
    access_token = create_access_token(data={"sub": str(user.id)})
    return {"access_token": access_token, "token_type": "bearer"}

Dans le jeton JWTsubIl s'agit du champ standard « subject », qui stocke l'identifiant de l'utilisateur. Le jeton lui-même n'est pas chiffré (juste encodé en Base64), doncAbsolument pasne mettez pas d'informations sensibles telles que des mots de passe dedans. Si vous voulez du chiffrement, utilisez JWE au lieu de JWT.

Utiliser la protection JWT dans les routes

Exemple

# Fonction de dépendance pour obtenir l'utilisateur actuel
from auth import oauth2_scheme, decode_access_token

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    """Analyser l'utilisateur actuel à partir du jeton JWT"""
    payload = decode_access_token(token)
    if payload is None:
        raise HTTPException(status_code=401, detail="Jeton invalide ou expiré")

    user_id = payload.get("sub")
    user = db.query(User).filter(User.id == int(user_id)).first()
    if user is None:
        raise HTTPException(status_code=401, detail=“L’utilisateur n’existe pas”)

    return user

# Injecter dans les routes qui nécessitent une connexion
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
    """Obtenir les informations de l'utilisateur actuel (connexion requise)"""
    return current_user

Pour les pages SSR, JWT est généralement stocké dans un Cookie (plutôt que dans l'en-tête Authorization), et dans les routes de template, le Token est extrait du Cookie pour vérification.

Résumé du chapitre

Dans ce chapitre, vous avez implémenté l'authentification JWT de FastAPI : passlib pour le hachage des mots de passe, python-jose pour l'émission et la vérification des JWT, OAuth2PasswordBearer pour l'extraction des Tokens, et Depends pour l'injection de dépendances d'authentification.

Contrairement à l'authentification par Session de Django/Flask, JWT est une solution sans état, plus adaptée aux scénarios de séparation front-end/back-end.

autres extensions