Aller au contenu principal
FastAPIPythonSQLAlchemyPostgreSQLFormation

CRUD : créer une API complète

30 min de lecture Apprendre FastAPI — Chapitre 3

Implémente les 4 opérations CRUD (Create, Read, Update, Delete) avec FastAPI et maîtrise les modèles Pydantic avancés.

Tu sais créer des routes, valider des paramètres et utiliser Pydantic. Mais une API sans les quatre opérations de base — créer, lire, mettre à jour, supprimer — c’est une vitrine, pas un produit. Ce chapitre te fait passer de l’exercice pédagogique à l’API de production : un CRUD complet avec des modèles bien séparés, une gestion d’erreurs centralisée, et les patterns qui évitent les problèmes classiques en équipe.

Les 4 opérations fondamentales

CRUD = Create, Read, Update, Delete. Chaque opération correspond à une méthode HTTP et un code de retour standard :

  • POST /tasks → 201 Created
  • GET /tasks et /tasks/{id} → 200 OK
  • PUT /tasks/{id} → 200 OK (mise à jour complète)
  • DELETE /tasks/{id} → 204 No Content

💡 PUT vs PATCH : PUT remplace toute la ressource, PATCH ne modifie que les champs envoyés. En pratique, la plupart des APIs utilisent PUT avec exclude_unset=True côté Pydantic pour gérer les deux cas.

Implémentation complète

Voici un CRUD de tâches en mémoire — la mécanique pure, sans base de données pour l’instant :

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI(title="API Tâches")

tasks_db: dict[int, dict] = {}
next_id: int = 1

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=1000)
    completed: bool = False
    priority: int = Field(default=1, ge=1, le=5)

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    completed: Optional[bool] = None
    priority: Optional[int] = Field(None, ge=1, le=5)

class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    completed: bool
    priority: int

Trois modèles, trois rôles :

  • TaskCreate — ce que le client envoie pour créer (pas d’id, c’est le serveur qui le génère)
  • TaskUpdate — tous les champs optionnels (on ne touche qu’à ce qui change)
  • TaskResponse — ce que l’API renvoie, avec l’id inclus

Les endpoints

@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    global next_id
    new_task = {"id": next_id, **task.model_dump()}
    tasks_db[next_id] = new_task
    next_id += 1
    return new_task

@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(completed: Optional[bool] = None, priority: Optional[int] = None):
    results = list(tasks_db.values())
    if completed is not None:
        results = [t for t in results if t["completed"] == completed]
    if priority is not None:
        results = [t for t in results if t["priority"] == priority]
    return results

@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Tâche non trouvée")
    return tasks_db[task_id]

@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, task: TaskUpdate):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Tâche non trouvée")
    update_data = task.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        tasks_db[task_id][key] = value
    return tasks_db[task_id]

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Tâche non trouvée")
    del tasks_db[task_id]

🔥 model_dump(exclude_unset=True) est la clé du update partiel. Si le client n’envoie que {"completed": true}, seul ce champ est modifié. Sans exclude_unset, les champs non fournis seraient écrasés par None.

Modèles Pydantic avancés

En production, tes modèles grandissent vite. L’héritage évite la duplication :

from datetime import datetime
from enum import Enum

class UserRole(str, Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"

class UserBase(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: str
    role: UserRole = UserRole.viewer

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=2)
    email: Optional[str] = None
    role: Optional[UserRole] = None

class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True

UserCreate hérite de UserBase et ajoute password. UserResponse hérite aussi de UserBase mais ajoute id et created_at — et n’inclut jamais password. C’est la séparation entrée/sortie qui protège tes données sensibles.

Validation inter-champs

Quand la validation dépend de plusieurs champs, utilise model_validator :

from pydantic import model_validator

class DateRange(BaseModel):
    start_date: str
    end_date: str

    @model_validator(mode="after")
    def validate_range(self):
        if self.start_date > self.end_date:
            raise ValueError("start_date doit précéder end_date")
        return self

🎯 Règle d’or : field_validator pour le format d’un champ, model_validator pour les relations entre champs. La logique métier (unicité, existence d’une relation) appartient à la couche service, pas aux modèles.

Gestion d’erreurs professionnelle

HTTPException suffit pour les cas simples. Pour une API de production, centralise la gestion d’erreurs :

from fastapi import Request
from fastapi.responses import JSONResponse

class ResourceNotFound(Exception):
    def __init__(self, resource: str, resource_id: int):
        self.resource = resource
        self.resource_id = resource_id

@app.exception_handler(ResourceNotFound)
async def not_found_handler(request: Request, exc: ResourceNotFound):
    return JSONResponse(
        status_code=404,
        content={
            "error": "not_found",
            "message": f"{exc.resource} #{exc.resource_id} introuvable",
            "path": str(request.url)
        }
    )

# Utilisation dans les endpoints
@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    if task_id not in tasks_db:
        raise ResourceNotFound("Task", task_id)
    return tasks_db[task_id]

L’avantage : un format d’erreur uniforme sur toute ton API, avec assez de contexte pour débugger sans exposer les internals.

⚠️ Ne renvoie jamais de stacktrace en production. En dev, FastAPI affiche les erreurs en détail. En prod, configure debug=False et utilise un logger structuré pour capturer les erreurs côté serveur.

Les pièges du CRUD

Le 404 oublié. Chaque endpoint qui prend un {id} doit vérifier que la ressource existe. Ça paraît évident, mais c’est la première source de 500 non gérées. Crée un helper get_or_404() et utilise-le partout.

Le DELETE qui renvoie du contenu. Convention HTTP : un DELETE réussi renvoie 204 (No Content) sans body. Si tu renvoies un 200 avec {"message": "deleted"}, tu casses le contrat REST et certains clients HTTP planteront en essayant de parser une réponse sur un 204.

Le PUT qui crée. Certaines APIs permettent à PUT de créer la ressource si elle n’existe pas (upsert). C’est valide en REST, mais confusant. Préfère séparer clairement : POST crée, PUT met à jour, et renvoie 404 si la ressource n’existe pas.

response_model oublié. Sans response_model, FastAPI renvoie tout ce que ta fonction retourne — y compris des champs sensibles comme hashed_password. Toujours spécifier le modèle de sortie sur les endpoints qui touchent des données utilisateur.

Les IDs séquentiels qui leakent. Un ID auto-incrémenté (/users/1, /users/2, /users/3) permet à n’importe qui de deviner le nombre total d’utilisateurs et d’énumérer les ressources. Pour les APIs publiques, préfère des UUIDs ou des identifiants opaques.

💡 Teste systématiquement les cas d’erreur, pas que le happy path. Un curl -X GET /tasks/999 doit renvoyer un 404 propre, pas un crash serveur. Un curl -X POST /tasks avec un body vide doit renvoyer un 422 avec les champs manquants — pas un 500.

Ce qu’on retient

🎯 Un CRUD bien construit suit des conventions prévisibles — les consommateurs de ton API savent à quoi s’attendre sans lire la doc.

Les essentiels :

  • 3 modèles par ressource — Create (input), Update (partiel), Response (output)
  • exclude_unset=True — pour les mises à jour partielles sans écraser les champs existants
  • response_model — filtre automatiquement la sortie, protège les données sensibles
  • Exception handlers — format d’erreur uniforme, pas de stacktraces en prod
  • Codes HTTP corrects — 201 pour Create, 204 pour Delete, 404 pour Not Found
  • Héritage PydanticUserBaseUserCreate / UserResponse évite la duplication

Prochain chapitre : on connecte tout ça à une vraie base de données avec SQLAlchemy et PostgreSQL — fini le dictionnaire en mémoire. 🚀

Articles liés