Aller au contenu principal

Gestion d’erreurs avec FastAPI

remarque

Ce chapitre vient après Relations SQL avec FastAPI. Son but est de rendre l’API plus propre, plus lisible et plus robuste quand quelque chose se passe mal.

Introduction

Jusqu’ici, on a appris à :

  • construire une API de base,
  • ajouter des utilisateurs,
  • protéger des routes,
  • relier les tâches à leur propriétaire.

Mais il manque encore une couche très importante :

savoir gérer proprement les erreurs.

Sans cela, on se retrouve vite avec :

  • des HTTPException(...) dispersées partout,
  • des messages incohérents,
  • des routes trop chargées,
  • une logique métier difficile à relire.

Ce que ce chapitre va accomplir

À la fin de ce chapitre, tu auras :

  • une hiérarchie d’exceptions métier,
  • des handlers centralisés,
  • des erreurs JSON cohérentes,
  • des services plus expressifs,
  • des routes plus propres,
  • une meilleure séparation entre :
    • la logique métier,
    • la traduction HTTP,
    • la réponse finale envoyée au client.

Pourquoi ce chapitre vient maintenant

Après l’authentification et les relations SQL, l’API commence à porter de vraies règles :

  • un email peut déjà exister,
  • un token peut être invalide,
  • une tâche peut ne pas exister,
  • une tâche peut appartenir à un autre utilisateur,
  • une requête de mise à jour peut être vide.

Toutes ces situations sont normales.

Le vrai sujet n’est donc pas seulement :

  • “comment éviter les erreurs ?”

mais plutôt :

  • “comment leur donner une forme propre, prévisible et maintenable ?”

Sources d’inspiration réelles

Ce chapitre s’appuie principalement sur :

  • 26-Error Handling/133-exceptions.py
  • 26-Error Handling/134-exceptions.py

Ce que ces sources montrent :

  • une base d’exceptions métier,
  • des sous-classes spécialisées,
  • une fonction qui enregistre automatiquement des handlers,
  • une volonté de centraliser la logique d’erreur au lieu de la répéter dans chaque route.

Choix pédagogique de ce chapitre

Les sources ont une bonne intuition :

  • créer une hiérarchie d’exceptions personnalisées,
  • brancher des handlers au niveau applicatif.

Mais pour notre wiki, on va faire une version un peu plus propre pédagogiquement :

  • on garde l’idée de la hiérarchie métier,
  • on garde l’idée des handlers centralisés,
  • mais au lieu de relancer des HTTPException depuis les handlers, on va directement retourner des JSONResponse.

Pourquoi ? Parce que c’est plus lisible quand on apprend.

Le vrai problème du code actuel

Dans une API qui grandit, si tu écris partout ceci :

raise HTTPException(status_code=404, detail="Task not found")

alors au début ça semble simple.

Mais plus le projet grossit, plus tu te retrouves avec :

  • des messages incohérents,
  • des codes répétés,
  • des routes qui deviennent verbeuses,
  • des services qui parlent déjà HTTP alors qu’ils devraient d’abord exprimer le métier.

La vraie idée à comprendre

Une bonne gestion d’erreurs dans une API repose sur trois niveaux :

  1. Le métier dit ce qui ne va pas. Exemple : “la tâche n’existe pas”, “l’utilisateur n’est pas autorisé”, “aucune donnée à mettre à jour”.

  2. Le handler traduit cela en réponse API. Exemple : 404, 401, 400, 409.

  3. Le client reçoit une réponse cohérente.

Autrement dit :

  • le service ne devrait pas toujours penser en premier en termes de HTTPException,
  • il devrait d’abord penser en termes de règle métier violée.

Architecture visée après ce chapitre

On ajoute un nouveau fichier clé :

fastapi-base-api/
├── main.py
├── config.py
├── requirements.txt
├── .gitignore
├── .env
├── .env.example
├── core/
│ ├── __init__.py
│ ├── security.py
│ └── exceptions.py
├── api/
│ ├── __init__.py
│ ├── router.py
│ ├── dependencies.py
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── task.py
│ │ └── user.py
│ └── schemas/
│ ├── __init__.py
│ ├── task.py
│ └── user.py
├── database/
│ ├── __init__.py
│ ├── models.py
│ └── session.py
└── services/
├── __init__.py
├── task.py
└── user.py

Repères utiles avant le code

Avant de coder, retiens ces points :

  • une exception métier n’est pas forcément une erreur “technique”,
  • une exception métier sert à nommer clairement une situation invalide,
  • un handler permet d’éviter de répéter le même format de réponse partout,
  • plus les services deviennent centraux, plus cette séparation devient utile.

Étape 1 — créer core/exceptions.py

On commence par définir une base d’exceptions métier.

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse


class AppError(Exception):
"""Base de toutes les exceptions métier de l'application."""

status_code = status.HTTP_400_BAD_REQUEST
detail = "Application error"


class EntityNotFound(AppError):
status_code = status.HTTP_404_NOT_FOUND
detail = "Entity not found"


class TaskNotFound(EntityNotFound):
detail = "Task not found"


class UserNotFound(EntityNotFound):
detail = "User not found"


class EmailAlreadyRegistered(AppError):
status_code = status.HTTP_409_CONFLICT
detail = "Email already registered"


class InvalidCredentials(AppError):
status_code = status.HTTP_401_UNAUTHORIZED
detail = "Email or password is incorrect"


class InvalidToken(AppError):
status_code = status.HTTP_401_UNAUTHORIZED
detail = "Invalid or expired token"


class ForbiddenAction(AppError):
status_code = status.HTTP_403_FORBIDDEN
detail = "You are not allowed to perform this action"


class NothingToUpdate(AppError):
status_code = status.HTTP_400_BAD_REQUEST
detail = "No data provided to update"


class InvalidTokenPayload(AppError):
status_code = status.HTTP_401_UNAUTHORIZED
detail = "Invalid token payload"


def _build_handler(exception_class: type[AppError]):
async def handler(request: Request, exc: AppError):
return JSONResponse(
status_code=exception_class.status_code,
content={"detail": exception_class.detail},
)

return handler


def add_exception_handlers(app: FastAPI):
for exception_class in AppError.__subclasses__():
app.add_exception_handler(exception_class, _build_handler(exception_class))

Pourquoi commencer comme ça ?

Parce qu’on donne enfin un nom métier aux problèmes.

Par exemple :

  • TaskNotFound
  • InvalidCredentials
  • NothingToUpdate

C’est déjà beaucoup plus clair à lire dans un service qu’un simple entier HTTP.

Étape 2 — améliorer l’enregistrement des handlers

Le morceau précédent avec AppError.__subclasses__() ne récupère que le premier niveau d’héritage.

Or TaskNotFound(EntityNotFound) est une sous-classe indirecte.

Donc, pour une version plus robuste, on peut écrire :

def get_all_subclasses(base_class: type[Exception]) -> list[type[Exception]]:
subclasses = []
for subclass in base_class.__subclasses__():
subclasses.append(subclass)
subclasses.extend(get_all_subclasses(subclass))
return subclasses


def add_exception_handlers(app: FastAPI):
for exception_class in get_all_subclasses(AppError):
app.add_exception_handler(exception_class, _build_handler(exception_class))
astuce

Cette version est plus fiable si ta hiérarchie d’exceptions commence à grandir.

Étape 3 — brancher les handlers dans main.py

Maintenant, il faut dire à FastAPI d’utiliser ces handlers.

from contextlib import asynccontextmanager

from fastapi import FastAPI
from scalar_fastapi import get_scalar_api_reference

from api.router import router
from core.exceptions import add_exception_handlers
from database.session import create_db_tables


@asynccontextmanager
async def lifespan(app: FastAPI):
await create_db_tables()
yield


app = FastAPI(
title="Task Organizer API",
description="Une API FastAPI avec gestion d’erreurs centralisée",
lifespan=lifespan,
)

add_exception_handlers(app)
app.include_router(router)


@app.get("/scalar", include_in_schema=False)
def get_scalar_docs():
return get_scalar_api_reference(
openapi_url=app.openapi_url,
title="Scalar API",
)

Ce que ce changement t’apporte

Tu n’as plus besoin de décider dans chaque route comment formatter l’erreur.

L’application sait maintenant répondre proprement dès qu’une exception métier connue est levée.

Étape 4 — alléger services/user.py

Avant, on levait surtout des HTTPException. Maintenant, on peut exprimer les erreurs via nos exceptions métier.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.schemas.user import UserCreate
from core.exceptions import EmailAlreadyRegistered, InvalidCredentials
from core.security import create_access_token, hash_password, verify_password
from database.models import User


class UserService:
def __init__(self, session: AsyncSession):
self.session = session

async def get_by_email(self, email: str) -> User | None:
result = await self.session.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()

async def get_by_id(self, user_id: int) -> User | None:
return await self.session.get(User, user_id)

async def add_user(self, credentials: UserCreate) -> User:
existing_user = await self.get_by_email(credentials.email)
if existing_user is not None:
raise EmailAlreadyRegistered()

user = User(
**credentials.model_dump(exclude={"password"}),
password_hash=hash_password(credentials.password),
)
self.session.add(user)
await self.session.commit()
await self.session.refresh(user)
return user

async def authenticate(self, email: str, password: str) -> str:
user = await self.get_by_email(email)
if user is None or not verify_password(password, user.password_hash):
raise InvalidCredentials()

return create_access_token(
data={
"sub": str(user.id),
"email": user.email,
"name": user.name,
}
)

Ce qu’il faut remarquer

Le service n’est plus pollué par des codes HTTP numériques.

Il dit simplement :

  • EmailAlreadyRegistered()
  • InvalidCredentials()

C’est beaucoup plus lisible.

Étape 5 — améliorer api/dependencies.py

La dépendance qui récupère l’utilisateur courant devient aussi plus propre.

from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from core.exceptions import InvalidToken, InvalidTokenPayload, UserNotFound
from core.security import decode_access_token, oauth2_scheme
from database.models import User
from database.session import get_session
from services.task import TaskService
from services.user import UserService


SessionDep = Annotated[AsyncSession, Depends(get_session)]
TokenDep = Annotated[str, Depends(oauth2_scheme)]


def get_task_service(session: SessionDep) -> TaskService:
return TaskService(session)


def get_user_service(session: SessionDep) -> UserService:
return UserService(session)


TaskServiceDep = Annotated[TaskService, Depends(get_task_service)]
UserServiceDep = Annotated[UserService, Depends(get_user_service)]


async def get_current_user(token: TokenDep, service: UserServiceDep) -> User:
token_data = decode_access_token(token)
if token_data is None:
raise InvalidToken()

user_id = token_data.get("sub")
if user_id is None:
raise InvalidTokenPayload()

user = await service.get_by_id(int(user_id))
if user is None:
raise UserNotFound()

return user


CurrentUserDep = Annotated[User, Depends(get_current_user)]

Pourquoi c’est mieux

Parce qu’on garde la dépendance lisible :

  • si le token est faux → InvalidToken
  • si le payload est mauvais → InvalidTokenPayload
  • si l’utilisateur n’existe plus → UserNotFound

Encore une fois, le code devient plus expressif.

Étape 6 — améliorer services/task.py

C’est ici qu’on va centraliser plusieurs erreurs métier liées aux tâches.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.schemas.task import TaskCreate
from core.exceptions import NothingToUpdate, TaskNotFound
from database.models import Task, User


class TaskService:
def __init__(self, session: AsyncSession):
self.session = session

async def get_task(self, id: int) -> Task | None:
return await self.session.get(Task, id)

async def get_task_for_user(self, task_id: int, user_id: int) -> Task:
result = await self.session.execute(
select(Task).where(Task.id == task_id, Task.owner_id == user_id)
)
task = result.scalar_one_or_none()
if task is None:
raise TaskNotFound()
return task

async def get_tasks_for_user(self, user_id: int) -> list[Task]:
result = await self.session.execute(
select(Task).where(Task.owner_id == user_id)
)
return list(result.scalars().all())

async def add_task(self, task_create: TaskCreate, owner: User) -> Task:
new_task = Task(
**task_create.model_dump(),
owner_id=owner.id,
)
self.session.add(new_task)
await self.session.commit()
await self.session.refresh(new_task)
return new_task

async def update_task(self, task: Task, task_update: dict) -> Task:
if not task_update:
raise NothingToUpdate()

task.sqlmodel_update(task_update)
self.session.add(task)
await self.session.commit()
await self.session.refresh(task)
return task

async def delete_task(self, task: Task) -> None:
await self.session.delete(task)
await self.session.commit()

Ce que cette version améliore

On sort maintenant du routeur deux responsabilités importantes :

  • vérifier si la tâche existe vraiment pour cet utilisateur,
  • vérifier si la mise à jour contient réellement quelque chose.

Le service devient plus intelligent, et la route devient plus simple.

Étape 7 — alléger api/routers/task.py

La route n’a plus besoin de porter toute la logique d’erreur.

from fastapi import APIRouter, status

from api.dependencies import CurrentUserDep, TaskServiceDep
from api.schemas.task import TaskCreate, TaskRead, TaskUpdate


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


@router.get("/mine", response_model=list[TaskRead])
async def list_my_tasks(
service: TaskServiceDep,
current_user: CurrentUserDep,
):
return await service.get_tasks_for_user(current_user.id)


@router.get("/", response_model=TaskRead)
async def get(
id: int,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
return await service.get_task_for_user(id, current_user.id)


@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
async def create(
task_create: TaskCreate,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
return await service.add_task(task_create, current_user)


@router.patch("/", response_model=TaskRead)
async def update(
id: int,
task_update: TaskUpdate,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
task = await service.get_task_for_user(id, current_user.id)
update_data = task_update.model_dump(exclude_unset=True)
return await service.update_task(task, update_data)


@router.delete("/")
async def delete(
id: int,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
task = await service.get_task_for_user(id, current_user.id)
await service.delete_task(task)
return {"message": f"Task with ID {id} has been deleted successfully"}

Pourquoi cette route est meilleure

Elle est plus lisible parce qu’elle se concentre sur son vrai rôle :

  • recevoir les paramètres,
  • appeler le service,
  • renvoyer la réponse.

Elle ne passe plus son temps à gérer tous les cas d’erreur à la main.

Étape 8 — option utile : gérer aussi une erreur interne proprement

Les sources montrent aussi l’idée de gérer plus proprement le 500.

Tu peux ajouter dans core/exceptions.py :

def add_internal_error_handler(app: FastAPI):
@app.exception_handler(Exception)
async def internal_error_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"detail": "Something went wrong"},
)

Puis dans main.py :

from core.exceptions import add_exception_handlers, add_internal_error_handler

add_exception_handlers(app)
add_internal_error_handler(app)
attention

Fais attention : un handler global sur Exception peut masquer des erreurs utiles pendant le développement. Il est souvent préférable de le garder simple en prod mais de conserver des logs détaillés côté serveur.

Vérification guidée

Relance l’application :

uvicorn main:app --reload

Test 1 — essayer de créer deux fois le même compte

curl -X POST http://127.0.0.1:8000/user/signup \
-H "Content-Type: application/json" \
-d '{
"name": "Enes",
"email": "enes@example.com",
"password": "StrongPass123"
}'

Puis rejoue exactement la même requête.

Réponse attendue :

{
"detail": "Email already registered"
}

avec un code 409 Conflict.

Test 2 — login avec mauvais mot de passe

curl -X POST http://127.0.0.1:8000/user/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=enes@example.com&password=WrongPassword"

Réponse attendue :

{
"detail": "Email or password is incorrect"
}

avec un code 401 Unauthorized.

Test 3 — appeler /user/me avec un faux token

curl http://127.0.0.1:8000/user/me \
-H "Authorization: Bearer fake_token"

Réponse attendue :

{
"detail": "Invalid or expired token"
}

Test 4 — récupérer une tâche inexistante

curl "http://127.0.0.1:8000/task/?id=999" \
-H "Authorization: Bearer $TOKEN_ENES"

Réponse attendue :

{
"detail": "Task not found"
}

Test 5 — envoyer un patch vide

curl -X PATCH "http://127.0.0.1:8000/task/?id=1" \
-H "Authorization: Bearer $TOKEN_ENES" \
-H "Content-Type: application/json" \
-d '{}'

Réponse attendue :

{
"detail": "No data provided to update"
}

Pourquoi ce chapitre est important pour la suite

Parce qu’une fois que la gestion d’erreurs est propre :

  • le middleware devient plus utile,
  • les tests deviennent plus lisibles,
  • les réponses de l’API deviennent plus cohérentes,
  • l’application devient plus facile à faire évoluer.

Autrement dit :

  • l’erreur handling n’est pas un bonus,
  • c’est une pièce de maturité du backend.

Pièges et erreurs fréquentes

  • continuer à lancer des HTTPException partout même quand l’app commence à grossir
  • créer des exceptions métier sans les brancher dans main.py
  • mélanger le niveau métier et le niveau HTTP
  • oublier qu’une mise à jour vide est aussi une erreur fonctionnelle
  • créer un handler Exception trop agressif qui masque tout pendant le debug

Bonnes pratiques à retenir

  • donner des noms métier explicites aux erreurs
  • centraliser les handlers
  • garder les services responsables des règles métier
  • garder les routes fines
  • viser des messages d’erreur cohérents et stables

Variantes et évolutions possibles

Après ce chapitre, tu peux aller vers :

  • middleware de logging et timing,
  • ajout d’identifiants de corrélation,
  • réponses d’erreur enrichies (code, detail, context),
  • tests dédiés au comportement d’erreur,
  • observabilité plus avancée.

Checklist de validation

Avant de passer à la suite, vérifie que tu peux :

  • nommer au moins deux exceptions métier utiles à ton application
  • montrer où les handlers sont enregistrés
  • expliquer la différence entre erreur métier et réponse HTTP
  • repérer quels morceaux de logique peuvent sortir des routes pour aller dans les services
  • expliquer pourquoi une erreur centralisée améliore la lisibilité du backend

Résumé final

Dans ce chapitre, tu as appris à faire passer ton API de :

  • routes qui gèrent leurs erreurs au cas par cas, à
  • une architecture où les erreurs métier sont nommées, centralisées et transformées proprement en réponses HTTP.

Tu as vu comment :

  • créer des exceptions métier,
  • enregistrer des handlers globaux,
  • alléger les services et les routes,
  • rendre les réponses d’erreur cohérentes.

Si tu retiens une seule chose, retiens celle-ci :

une API mature ne gère pas seulement ses cas heureux ; elle sait aussi exprimer proprement ses cas d’échec.

Pour aller plus loin