Gestion des Paramètres dans FastAPI
La manière dont FastAPI reçoit les paramètres diffère légèrement de frameworks comme Django ou Flask. Cette section explore les différentes méthodes disponibles.
1. Paramètres de Chemin (Path Parameters)
Les paramètres de chemin sont intégrés directement dans l'URL et sont définis dans le décorateur de route. FastAPI valide et convertit automatiquement le type de données spécifié.
import uvicorn
from fastapi import FastAPI
application_api = FastAPI()
@application_api.get("/ressources/{identifiant_item}")
async def lire_ressource(identifiant_item: int):
"""
Récupère un paramètre via le chemin de l'URL.
Args:
identifiant_item: L'identifiant numérique de la ressource.
Returns:
Un dictionnaire contenant l'identifiant de la ressource.
"""
print(f"Identifiant reçu: {identifiant_item}")
return {"identifiant_recu": identifiant_item}
if __name__ == '__main__':
uvicorn.run(application_api, host='127.0.0.1', port=8000)
L'exemple ci-dessus montre identifiant_item comme un paramètre de chemin de type int. FastAPI se charge de la validation. Une URL sans un paramètre correspondant générera une erreur 404.
2. Paramètres de Requête (Query Parameters)
Ces paramètres sont ajoutés à la fin de l'URL après un point d'interrogation (?). Ils sont définis comme des arguments de fonction dans le gestionnaire de route.
import uvicorn
from fastapi import FastAPI, Query
application_api = FastAPI()
@application_api.get("/catalogue")
async def obtenir_catalogue(terme_recherche: str = Query(None, max_length=50)):
"""
Récupère des articles de catalogue avec un terme de recherche optionnel.
Args:
terme_recherche: Un terme de recherche, limité à 50 caractères.
Returns:
Un dictionnaire contenant les articles et le terme de recherche si fourni.
"""
elements_trouves = {"liste_elements": [{"id": "ProdA"}, {"id": "ProdB"}]}
if terme_recherche:
elements_trouves.update({"recherche_effectuee": terme_recherche})
return elements_trouves
if __name__ == '__main__':
uvicorn.run(application_api, host='127.0.0.1', port=8000)
Le paramètre terme_recherche est un paramètre de requête. Query(None, max_length=50) définit sa valeur par défaut à None s'il n'est pas fourni, et une longueur maximale de 50 caractères.
3. Corps de Requête (Request Body)
Pour les données envoyées dans le corps de la requête, comme JSON, FastAPI utilise des modèles Pydantic pour définir la structure attendue.
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel
application_api = FastAPI()
class Produit(BaseModel):
""" Modèle Pydantic pour un produit. """
nom: str
description: str = None
prix: float
taxe: float = None
@application_api.post("/produits/")
async def creer_produit(produit_data: Produit):
"""
Crée un nouveau produit à partir des données du corps de la requête.
Args:
produit_data: Les données du produit, conformes au modèle Produit.
Returns:
Un dictionnaire contenant les données du produit créé.
"""
print(produit_data)
return {"produit_cree": produit_data}
if __name__ == '__main__':
uvicorn.run(application_api, host='127.0.0.1', port=8000)
Le modèle Produit spécifie les champs et leurs types. FastAPI désérialise automatiquement le JSON entrant en un objet Produit. Si les données ne correspondent pas au modèle, une erreur 422 est renvoyée.
4. Données de Formulaire (Form Data)
Pour les données de formulaire HTML (application/x-www-form-urlencoded), utilisez la classe Form.
import uvicorn
from fastapi import FastAPI, Form
application_api = FastAPI()
@application_api.post("/authentification")
async def authentifier_utilisateur(nom_utilisateur: str = Form(...), mot_de_passe: str = Form(...)):
"""
Authentifie un utilisateur via les données de formulaire.
Args:
nom_utilisateur: Le nom d'utilisateur fourni via le formulaire.
mot_de_passe: Le mot de passe fourni via le formulaire.
Returns:
Un dictionnaire avec le nom d'utilisateur et le mot de passe.
"""
return {"utilisateur": nom_utilisateur, "mdp_saisi": mot_de_passe}
if __name__ == '__main__':
uvicorn.run(application_api, host="127.0.0.1", port=8000)
Ici, Form(...) indique que nom_utilisateur et mot_de_passe sont des champs de formulaire obligatoires.
5. Téléchargement de Fichiers (File Upload)
Utilisez les classes File ou UploadFile pour gérer les téléchargements de fichiers.
import uvicorn
from fastapi import FastAPI, File, UploadFile
application_api = FastAPI()
@application_api.post("/fichiers/taille_brute")
async def telecharger_fichier_bytes(fichier_brut: bytes = File(...)):
"""
Télécharge un fichier et renvoie sa taille en octets.
Args:
fichier_brut: Le contenu du fichier en tant qu'octets.
Returns:
Un dictionnaire avec la taille du fichier.
"""
return {"taille_fichier": len(fichier_brut)}
@application_api.post("/fichiers/info_detaillee")
async def telecharger_fichier_details(fichier_info: UploadFile = File(...)):
"""
Télécharge un fichier et renvoie son nom.
Args:
fichier_info: L'objet UploadFile contenant les détails du fichier.
Returns:
Un dictionnaire avec le nom du fichier.
"""
return {"nom_fichier_telecharge": fichier_info.filename}
if __name__ == '__main__':
uvicorn.run(application_api, host="127.0.0.1", port=8000)
File convient pour lire le fichier directement en octets, tandis que UploadFile offre un accès plus structuré avec des attributs comme filename et content_type.
6. En-têtes de Requête (Request Headers)
La classe Header permet d'extraire des valeurs spécifiques des en-têtes HTTP de la requête.
import uvicorn
from fastapi import FastAPI, Header
application_api = FastAPI()
@application_api.get("/donnees/")
async def lire_donnees(agent_utilisateur: str = Header(None)):
"""
Récupère l'en-tête User-Agent de la requête.
Args:
agent_utilisateur: La valeur de l'en-tête User-Agent.
Returns:
Un dictionnaire avec la valeur de l'User-Agent.
"""
return {"agent_utilisateur": agent_utilisateur}
if __name__ == '__main__':
uvicorn.run(application_api, host="127.0.0.1", port=8000)
7. Injection de Dépendances (Dependency Injection)
Le système de dépendances de FastAPI est puissant pour gérer des ressources partagées, comme les sessions de base de données, ou pour valider des jetons.
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
application_api = FastAPI()
# Configuration simplifiée d'une base de données SQLite
URL_BDD = "sqlite:///ma_base.db"
moteur_bdd = create_engine(URL_BDD)
SessionLocale = sessionmaker(autocommit=False, autoflush=False, bind=moteur_bdd)
def obtenir_session_bdd():
""" Fournit une session de base de données et la ferme après utilisation. """
session_db = SessionLocale()
try:
yield session_db
except Exception as e:
print(f"Erreur de session BDD: {e}")
session_db.rollback()
finally:
session_db.close()
@application_api.get("/elements/")
async def lister_elements(identifiant_utilisateur: str, session_db: Session = Depends(obtenir_session_bdd)):
"""
Récupère des éléments liés à un utilisateur, en utilisant une session BDD injectée.
(Opération BDD simulée, à remplacer par une vraie requête)
Args:
identifiant_utilisateur: L'identifiant de l'utilisateur.
session_db: La session de base de données injectée.
Returns:
Un dictionnaire de données (simulé).
"""
# Ici, vous effectueriez des opérations de base de données avec session_db
# Exemple: data = session_db.query(MonModel).filter(MonModel.user_id == identifiant_utilisateur).all()
print(f"Opération BDD pour utilisateur: {identifiant_utilisateur}")
return {"message": "Données récupérées avec session BDD", "utilisateur": identifiant_utilisateur}
# Exemple de dépendance pour la validation de jeton
async def verifier_jeton_authentification(jeton_x: str = Header(..., alias="X-Auth-Token")):
""" Vérifie la validité d'un jeton d'authentification. """
if jeton_x != "jeton-super-secret-valide":
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Jeton X-Auth-Token invalide")
return True
@application_api.get("/donnees-protegees/", dependencies=[Depends(verifier_jeton_authentification)])
async def obtenir_donnees_protegees():
"""
Récupère des données nécessitant une authentification par jeton.
La dépendance verifier_jeton_authentification est exécutée avant.
"""
return [{"information": "Confidentielle A"}, {"information": "Confidentielle B"}]
if __name__ == '__main__':
uvicorn.run(application_api, host="127.0.0.1", port=8000)
FastAPI assure que les dépendances sont appelées une seule fois par requête, même si elles sont utilisées à plusieurs endroits.
8. Tâches d'Arrière-plan (Background Tasks)
Utilisez BackgroundTasks pour exécuter des opérations après l'envoi de la réponse HTTP au client.
from fastapi import FastAPI, BackgroundTasks
application_api = FastAPI()
def envoyer_notification_email(adresse_email: str, contenu_message: str = ""):
""" Fonction simulant l'envoi d'une notification par email. """
with open("journal_notifications.log", mode="a") as fichier_notif:
log_content = f"[{adresse_email}] Notification: {contenu_message}\n"
fichier_notif.write(log_content)
@application_api.post("/notifier/{adresse_email}")
async def envoyer_notification(adresse_email: str, taches_fond: BackgroundTasks):
"""
Envoie une notification en arrière-plan après avoir renvoyé la réponse.
Args:
adresse_email: L'adresse email du destinataire.
taches_fond: L'objet BackgroundTasks pour ajouter la tâche.
Returns:
Un dictionnaire confirmant l'envoi en arrière-plan.
"""
taches_fond.add_task(envoyer_notification_email, adresse_email, contenu_message="Votre commande a été traitée.")
return {"statut": "Notification envoyée en arrière-plan"}
Ceci est idéal pour des opérations non critiques qui ne doivent pas bloquer la réponse de l'API, comme l'envoi d'emails ou le traitement de données.
9. Objet Requête (Request Object)
Accédez à l'objet Request de Starlette pour obtenir des informations complètes sur la requête HTTP.
from fastapi import FastAPI, Request
application_api = FastAPI()
@application_api.get("/informations-requete/")
async def obtenir_informations_requete(requete_http: Request):
"""
Récupère l'adresse IP du client à partir de l'objet Request.
Args:
requete_http: L'objet Request de Starlette.
Returns:
Un dictionnaire avec l'adresse IP du client.
"""
return {"adresse_ip_client": requete_http.client.host}
L'objet Request est utile pour des validations spécifiques basées sur le client ou d'autres attributs de la requête.
Journalisation dans FastAPI
La journalisation est essentielle pour le débogage et la surveillance des applications. Python offre des outils flexibles avec les modules Logger et Handler.
- Logger : C'est le point d'entrée du système de journalisation. Il crée des enregistrements de log et gère le filtrage initial basé sur le niveau de sévérité.
- Handler : Un
Handlerest responsable de diriger les enregistrements de log vers une destination spécifique (console, fichier, réseau, etc.). UnLoggerpeut avoir plusieursHandlers, chacun avec son propre niveau de filtrage et son formateur.
1. Journalisation Basique vers Fichier et Console
Une configuration standard permet d'envoyer les logs à la fois à la console et à un fichier.
import logging
from fastapi import FastAPI
# Initialisation et configuration du logger
logger_app = logging.getLogger(__name__)
logger_app.setLevel(logging.INFO)
# Création d'un gestionnaire pour les fichiers
gestionnaire_fichier = logging.FileHandler('journal_application.log')
gestionnaire_fichier.setLevel(logging.INFO)
# Création d'un gestionnaire pour la console
gestionnaire_console = logging.StreamHandler()
gestionnaire_console.setLevel(logging.INFO)
# Définition du format des logs
formateur_log = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
gestionnaire_fichier.setFormatter(formateur_log)
gestionnaire_console.setFormatter(formateur_log)
# Ajout des gestionnaires au logger
logger_app.addHandler(gestionnaire_fichier)
logger_app.addHandler(gestionnaire_console)
application_api_log = FastAPI()
@application_api_log.get("/racine")
async def lire_racine():
"""
Exemple de point de terminaison qui enregistre une action.
"""
logger_app.info("Traitement du point de terminaison racine.")
return {"message": "Bonjour Monde!"}
2. Journalisation Avancée avec Uvicorn
Pour capturer également les logs de Uvicorn (accès HTTP, erreurs), on peut modifier la configuration de journalisation transmise à uvicorn.run.
import logging
from logging.handlers import TimedRotatingFileHandler
from fastapi import FastAPI
import uvicorn
app_uvicorn_log = FastAPI()
CONFIGURATION_LOGS_UVICORN = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"()": "uvicorn.logging.DefaultFormatter",
"fmt": "%(levelprefix)s %(message)s",
"use_colors": None,
},
"acces_http": {
"()": "uvicorn.logging.AccessFormatter",
"fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
},
},
"handlers": {
"standard_sortie": {
"formatter": "standard",
"class": "logging.handlers.TimedRotatingFileHandler",
"filename": "./journal_uvicorn.log",
"when": "midnight",
"interval": 1,
"backupCount": 7,
},
"acces_sortie": {
"formatter": "acces_http",
"class": "logging.handlers.TimedRotatingFileHandler",
"filename": "./journal_uvicorn.log",
"when": "midnight",
"interval": 1,
"backupCount": 7,
},
},
"loggers": {
"": {"handlers": ["standard_sortie"], "level": "INFO"},
"uvicorn.error": {"level": "INFO"},
"uvicorn.access": {"handlers": ["acces_sortie"], "level": "INFO", "propagate": False},
},
}
@app_uvicorn_log.get("/")
async def lire_racine_uvicorn():
return {"message": "Bonjour, Monde!"}
if __name__ == "__main__":
uvicorn.run(app_uvicorn_log, host="127.0.0.1", port=8000, log_config=CONFIGURATION_LOGS_UVICORN)
Cette configuration redirige les logs de Uvicorn (y compris les accès) vers un fichier journal_uvicorn.log avec une rotation quotidienne.
3. Journalisation via Middleware
L'utilisation de middlewares permet d'intercepter les requêtes et les réponses pour enregistrer des informations pertinentes sur chaque transaction.
from fastapi import FastAPI, Request
from loguru import logger
import uvicorn
# Création de l'application FastAPI
app_middleware_log = FastAPI()
# Configuration de Loguru
logger.add("./requetes_api.log", rotation="100 MB", compression="zip")
@app_middleware_log.middleware("http")
async def enregistrer_requetes_middleware(requete: Request, appel_suivant):
"""
Middleware pour enregistrer les requêtes et les réponses HTTP.
"""
logger.info(f"Requête entrante: {requete.method} {requete.url}")
reponse = await appel_suivant(requete)
logger.info(f"Réponse sortante: {reponse.status_code} pour {requete.method} {requete.url}")
return reponse
@app_middleware_log.get("/")
async def racine_middleware():
return {"message": "Salut depuis le middleware!"}
if __name__ == "__main__":
uvicorn.run(app_middleware_log, host="127.0.0.1", port=8000)
Loguru simplifie la configuration de la journalisation. Le middleware enregistrer_requetes_middleware capture et log les détails de chaque requête et de sa réponse.
4. Journalisation d'Audit avec BaseHTTPMiddleware
Pour un journal d'audit plus détaillé, incluant le temps de traitement, on peut créer une classe de middleware personnalisée.
import time
import uvicorn
from loguru import logger
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
# Configuration de Loguru pour l'audit
logger.add("./journal_audit.log", rotation="50 MB", compression="zip")
app_audit_log = FastAPI()
class MiddlewareJournalAudit(BaseHTTPMiddleware):
"""
Middleware pour enregistrer les requêtes d'audit, y compris le temps de traitement.
"""
async def dispatch(self, requete: Request, appel_suivant):
debut_traitement = time.time()
logger.info(f"AUDIT - Requête reçue: {requete.method} {requete.url} depuis {requete.client.host}")
reponse = await appel_suivant(requete)
temps_traitement = time.time() - debut_traitement
logger.info(
f"AUDIT - Requête traitée: {requete.method} {requete.url} - Statut: {reponse.status_code} - Temps: {temps_traitement:.4f}s"
)
return reponse
app_audit_log.add_middleware(MiddlewareJournalAudit)
@app_audit_log.get("/audite")
async def racine_auditee():
return {"message": "Point de terminaison audité."}
if __name__ == "__main__":
uvicorn.run(app_audit_log, host="127.0.0.1", port=8000)
Ce middleware ajoute une entrée de log avant et après le traitement de la requête, fournissant une vue d'ensemble du temps de réponse et du statut.
5. Journalisation via Décorateur
Les décorateurs Python peuvent être utilisés pour ajouter des capacités de journalisation à des fonctions de route spécifiques, permettant une personnalisation fine.
import logging
from typing import Callable, Any
from functools import wraps
import uvicorn
from fastapi import FastAPI
# Configuration du logger de base
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger_decorateur = logging.getLogger(__name__)
def journaliser_action(nom_module: str, description_action: str):
"""
Décorateur de journalisation pour enregistrer le module et la description de l'action.
"""
def decorateur_interne(func: Callable) -> Callable:
@wraps(func)
async def enveloppe(*args: Any, **kwargs: Any) -> Any:
logger_decorateur.info(f"[{nom_module}] '{description_action}' - Démarrage de la requête: {func.__name__}")
resultat = await func(*args, **kwargs)
logger_decorateur.info(
f"[{nom_module}] '{description_action}' - Requête terminée: {func.__name__} - Résultat: {resultat}"
)
return resultat
return enveloppe
return decorateur_interne
app_decorateur_log = FastAPI()
@app_decorateur_log.get("/")
@journaliser_action(nom_module="Module Principal", description_action="Accès à la racine de l'API")
async def obtenir_racine():
return {"info": "Bienvenue sur l'API!"}
@app_decorateur_log.get("/articles/{article_id}")
@journaliser_action(nom_module="Module Articles", description_action="Lecture d'un article spécifique")
async def lire_un_article(article_id: int):
return {"article_id": article_id, "message": "Détails de l'article récupérés"}
if __name__ == '__main__':
uvicorn.run(app_decorateur_log, host="127.0.0.1", port=8000)
Ce décorateur fournit une manière propre d'ajouter des logs spécifiques à chaque point de terminaison avec des informations contextuelles.
Authentification et Autorisation dans FastAPI
La mise en place d'un système d'authentification est une tâche courante. Voici une approche structurée pour gérer l'inscription, la connexion par jeton JWT et l'accès aux routes protégées.
1. Configuration de l'Application
Un fichier de configuration centralise les paramètres de l'application, comme la connexion à la base de données et les clés secrètes pour les jetons.
import os
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
# Configuration de base pour la base de données et JWT
class ConfigurationPrincipale:
""" Paramètres essentiels de l'application. """
URL_BASE_DONNEES = "sqlite:///donnees_app.db"
moteur_sql = create_engine(URL_BASE_DONNEES, echo=False)
SessionLocale = sessionmaker(autocommit=False, autoflush=False, bind=moteur_sql)
CLE_SECRETE_JWT = "une-cle-secrete-pour-le-jwt-tres-longue-et-aleatoire" # À générer de manière sécurisée
ALGORITHME_JWT = "HS256"
DUREE_EXPIRATION_TOKEN_MINUTES = 60
# Exemple de configuration utilisant un fichier .env (avec pydantic-settings)
class ConfigurationEnv(BaseSettings):
""" Charge la configuration depuis un fichier .env. """
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="allow")
NOM_BDD: str = Field("donnees_app.db", description="Nom du fichier de la base de données")
REDIS_HOTE: str = Field("127.0.0.1", description="Adresse de l'hôte Redis")
REDIS_PORT: int = Field(6379, description="Port de Redis")
CLE_SECRETE_APP: str = Field("cle-par-defaut-si-non-dans-env", description="Clé secrète de l'application pour JWT")
class GestionnaireSessionBDD:
""" Contexte pour gérer la session de base de données. """
def __init__(self):
self.session = ConfigurationPrincipale.SessionLocale()
def __enter__(self) -> Session:
return self.session
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type:
self.session.rollback()
self.session.close()
La classe ConfigurationPrincipale regroupe les constantes de l'application. GestionnaireSessionBDD fournit une manière sécurisée d'interagir avec la base de données.
2. Modèles de Données et Schémas Pydantic
Définir les modèles de base de données (SQLAlchemy) et les schémas de validation (Pydantic) est crucial pour une gestion structurée des données.
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from config_app import ConfigurationPrincipale # Renommé pour correspondre au nouveau fichier
BaseSQL = declarative_base()
class UtilisateurBD(BaseSQL):
""" Modèle de table pour les utilisateurs. """
__tablename__ = "utilisateurs"
id = Column(Integer, primary_key=True, index=True, comment="ID unique de l'utilisateur")
nom_utilisateur = Column(String, unique=True, index=True, comment="Nom d'utilisateur")
mot_de_passe_hache = Column(String, comment="Mot de passe haché")
if __name__ == '__main__':
# Création des tables de la base de données
BaseSQL.metadata.create_all(bind=ConfigurationPrincipale.moteur_sql)
from pydantic import BaseModel, Field
class UtilisateurCreation(BaseModel):
""" Schéma Pydantic pour la création et la connexion d'un utilisateur. """
nom_utilisateur: str = Field(..., min_length=3, max_length=20)
mot_de_passe: str = Field(..., min_length=6)
class UtilisateurSchema(BaseModel):
""" Schéma Pydantic pour représenter un utilisateur (sans le mot de passe). """
id: int
nom_utilisateur: str
3. Opérations de Base de Données (ORM)
Une couche d'abstraction pour les opérations CRUD simplifie l'interaction avec la base de données.
import math
from sqlalchemy import desc, asc
from sqlalchemy.orm.query import Query
from sqlalchemy.orm.session import Session
class OperationsBaseBDD:
""" Classe utilitaire pour les opérations CRUD génériques sur la base de données. """
@classmethod
def recuperer_liste_paginee(cls, requete_bd: Query, filtres: set, tri: str = "-id", page_offset: int = 1, limite: int = 15) -> dict:
""" Récupère une liste d'enregistrements avec pagination. """
nombre_total = requete_bd.filter(*filtres).count()
resultat = {
"pagination": {
"total_enregistrements": nombre_total,
"pages_totales": cls._calculer_pages_totales(nombre_total, limite),
"page_actuelle": page_offset
},
"elements": []
}
if nombre_total > 0:
offset_reel = (page_offset - 1) * limite if page_offset > 0 else 0
requete_filtree = requete_bd.filter(*filtres)
regles_tri = cls._parser_regles_tri(tri)
elements_trouves = requete_filtree.order_by(*regles_tri).offset(offset_reel).limit(limite).all()
resultat["elements"] = [cls._convertir_en_dict(e) for e in elements_trouves]
return resultat
@classmethod
def recuperer_un_enregistrement(cls, requete_bd: Query, filtres: set, tri: str = "-id") -> dict:
""" Récupère un seul enregistrement correspondant aux filtres. """
enregistrement = requete_bd.filter(*filtres).order_by(*cls._parser_regles_tri(tri)).first()
return cls._convertir_en_dict(enregistrement) if enregistrement else {}
@staticmethod
def inserer_enregistrement(session_db: Session, modele_objet, donnees: dict) -> int:
""" Insère un nouvel enregistrement dans la base de données. """
nouvel_objet = modele_objet(**donnees)
session_db.add(nouvel_objet)
session_db.flush()
return nouvel_objet.id
@staticmethod
def mettre_a_jour_enregistrements(requete_bd: Query, donnees: dict, filtres: set) -> int:
""" Met à jour des enregistrements correspondant aux filtres. """
return requete_bd.filter(*filtres).update(donnees, synchronize_session=False)
@staticmethod
def supprimer_enregistrements(requete_bd: Query, filtres: set) -> int:
""" Supprime des enregistrements correspondant aux filtres. """
return requete_bd.filter(*filtres).delete(synchronize_session=False)
@staticmethod
def _calculer_pages_totales(nombre_elements: int, taille_page: int) -> int:
""" Calcule le nombre total de pages. """
if taille_page <= 0:
return 0
return math.ceil(nombre_elements / taille_page)
@staticmethod
def _parser_regles_tri(chaine_tri: str):
""" Convertit une chaîne de tri ('+champ,-autre') en objets de tri SQLAlchemy. """
regles = []
for item in chaine_tri.split(","):
if not item: continue
sens = item[0]
champ = item[1:] if sens in ['+', '-'] else item
if sens == "-":
regles.append(desc(champ))
else:
regles.append(asc(champ))
return regles
@staticmethod
def _convertir_en_dict(obj_modele):
""" Convertit un objet modèle SQLAlchemy en dictionnaire. """
if not obj_modele:
return {}
return {c.name: getattr(obj_modele, c.name) for c in obj_modele.__table__.columns}
4. Logique d'Authentification (JWT)
Cette partie gère le hachage des mots de passe, la création de jetons JWT et la vérification des utilisateurs via ces jetons.
from typing import Optional, Dict
from datetime import datetime, timedelta
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from models_auth import UtilisateurBD # Renommé
from operations_bdd import OperationsBaseBDD # Renommé
from config_app import ConfigurationPrincipale, GestionnaireSessionBDD # Renommé
# Contexte pour le hachage des mots de passe
contexte_hachage_mdp = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Schéma d'authentification OAuth2 pour la récupération des jetons
schema_oauth2 = OAuth2PasswordBearer(tokenUrl="api/utilisateurs/jeton") # Mis à jour le tokenUrl
def verifier_mot_de_passe(mdp_clair: str, mdp_hache: str) -> bool:
""" Vérifie si un mot de passe clair correspond à son hachage. """
return contexte_hachage_mdp.verify(mdp_clair, mdp_hache)
def obtenir_hachage_mot_de_passe(mdp: str) -> str:
""" Génère le hachage d'un mot de passe. """
return contexte_hachage_mdp.hash(mdp)
def generer_jeton_acces(donnees: Dict, duree_expiration: Optional[timedelta] = None) -> str:
""" Crée un jeton d'accès JWT. """
payload = donnees.copy()
expiration_delta = duree_expiration if duree_expiration else timedelta(minutes=ConfigurationPrincipale.DUREE_EXPIRATION_TOKEN_MINUTES)
expire_a = datetime.utcnow() + expiration_delta
payload.update({"exp": expire_a})
jeton_code = jwt.encode(payload, ConfigurationPrincipale.CLE_SECRETE_JWT, algorithm=ConfigurationPrincipale.ALGORITHME_JWT)
return jeton_code
async def recuperer_utilisateur_courant(jeton: str = Depends(schema_oauth2)) -> Dict:
"""
Dépendance pour récupérer l'utilisateur authentifié à partir d'un jeton JWT.
Lève une HTTPException si le jeton est invalide ou l'utilisateur introuvable.
"""
exceptions_credentials = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Impossible de valider les informations d'identification",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(jeton, ConfigurationPrincipale.CLE_SECRETE_JWT, algorithms=[ConfigurationPrincipale.ALGORITHME_JWT])
nom_utilisateur: str = payload.get("sub") # Utiliser "sub" pour le sujet du jeton
if nom_utilisateur is None:
raise exceptions_credentials
except JWTError:
raise exceptions_credentials
with GestionnaireSessionBDD() as session_db:
utilisateur_trouve = OperationsBaseBDD.recuperer_un_enregistrement(
session_db.query(UtilisateurBD), filtres={UtilisateurBD.nom_utilisateur == nom_utilisateur}
)
if not utilisateur_trouve:
raise exceptions_credentials
return utilisateur_trouve
5. Points de Terminaison API pour l'Authentification
Les routes pour l'inscription, la connexion et les pages protégées utilisent les dépendances et les fonctions d'aide définies précédemment.
from typing import Dict
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status
from auth_service import generer_jeton_acces, recuperer_utilisateur_courant, obtenir_hachage_mot_de_passe, verifier_mot_de_passe # Renommé
from models_auth import UtilisateurBD # Renommé
from schemas_auth import UtilisateurCreation # Renommé
from config_app import GestionnaireSessionBDD, ConfigurationPrincipale # Renommé
from operations_bdd import OperationsBaseBDD # Renommé
routeur_utilisateurs = APIRouter(prefix="/utilisateurs", tags=["Utilisateurs"])
@routeur_utilisateurs.post("/inscription")
async def enregistrer_nouvel_utilisateur(donnees_utilisateur: UtilisateurCreation):
"""
Enregistre un nouvel utilisateur.
Args:
donnees_utilisateur: Schéma Pydantic pour les informations d'inscription.
Returns:
Un dictionnaire de confirmation.
"""
with GestionnaireSessionBDD() as session_db:
utilisateur_existant = OperationsBaseBDD.recuperer_un_enregistrement(
session_db.query(UtilisateurBD), filtres={UtilisateurBD.nom_utilisateur == donnees_utilisateur.nom_utilisateur}
)
if utilisateur_existant:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Le nom d'utilisateur existe déjà.")
mdp_hache = obtenir_hachage_mot_de_passe(donnees_utilisateur.mot_de_passe)
nouvel_id_utilisateur = OperationsBaseBDD.inserer_enregistrement(
session_db, UtilisateurBD, {"nom_utilisateur": donnees_utilisateur.nom_utilisateur, "mot_de_passe_hache": mdp_hache}
)
session_db.commit()
return {"code": status.HTTP_201_CREATED, "message": "Inscription réussie", "donnees": {"id": nouvel_id_utilisateur, "nom_utilisateur": donnees_utilisateur.nom_utilisateur}}
@routeur_utilisateurs.post("/jeton")
async def obtenir_jeton_connexion(donnees_connexion: UtilisateurCreation):
"""
Authentifie un utilisateur et renvoie un jeton d'accès JWT.
Args:
donnees_connexion: Schéma Pydantic pour les informations de connexion.
Returns:
Un dictionnaire contenant le jeton d'accès et son type.
"""
with GestionnaireSessionBDD() as session_db:
utilisateur = OperationsBaseBDD.recuperer_un_enregistrement(
session_db.query(UtilisateurBD), filtres={UtilisateurBD.nom_utilisateur == donnees_connexion.nom_utilisateur}
)
if not utilisateur or not verifier_mot_de_passe(donnees_connexion.mot_de_passe, utilisateur.get("mot_de_passe_hache")):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Nom d'utilisateur ou mot de passe invalide.")
jeton_acces = generer_jeton_acces(
data={"sub": utilisateur.get("nom_utilisateur"), "user_id": utilisateur.get("id")},
duree_expiration=timedelta(minutes=ConfigurationPrincipale.DUREE_EXPIRATION_TOKEN_MINUTES)
)
return {"statut": status.HTTP_200_OK, "message": "Connexion réussie", "donnees": {"access_token": jeton_acces, "token_type": "bearer"}}
@routeur_utilisateurs.get("/tableau-de-bord")
async def acceder_tableau_de_bord(utilisateur_actuel: Dict = Depends(recuperer_utilisateur_courant)):
"""
Point de terminaison protégé, accessible uniquement après authentification réussie.
"""
print(f"L'utilisateur {utilisateur_actuel.get('nom_utilisateur')} est connecté.")
return {"code": status.HTTP_200_OK, "message": "Bienvenue sur le tableau de bord!", "donnees": {"id": utilisateur_actuel.get("id"), "nom_utilisateur": utilisateur_actuel.get("nom_utilisateur")}}
6. Démarrage de l'Application
La logique de démarrage de l'application FastAPI est encapsulée pour monter les routeurs.
from fastapi import FastAPI
from .routes_utilisateurs import routeur_utilisateurs # Renommé le fichier API
def creer_instance_app():
""" Initialise et configure l'application FastAPI. """
application = FastAPI(
title="API d'Authentification",
description="Exemple d'API FastAPI avec authentification JWT."
)
application.include_router(routeur_utilisateurs, prefix="/api")
return application
Le fichier principal pour lancer l'application avec Uvicorn :
import uvicorn
from application import creer_instance_app # Renommé le module
app_principale = creer_instance_app()
if __name__ == '__main__':
uvicorn.run(app_principale, host="127.0.0.1", port=8000)