1. Installation & Hello World
FastAPI est un framework web moderne, très rapide (basé sur Starlette + Pydantic), avec génération automatique de la documentation OpenAPI.
pip install fastapi==0.111.0 uvicorn[standard]==0.29.0 pydantic==2.7.0
# Lancer le serveur (reload automatique en dev)
uvicorn main:app --reload --port 8000
# Accéder à la doc automatique
# http://localhost:8000/docs ← Swagger UI
# http://localhost:8000/redoc ← ReDoc
from __future__ import annotations
from fastapi import FastAPI
app = FastAPI(
title="TaskAPI",
description="API de gestion de tâches",
version="1.0.0",
)
@app.get("/")
async def root() -> dict:
return {"message": "Bienvenue sur TaskAPI", "docs": "/docs"}
# ── Avantages vs Flask ────────────────────────────────────────
# ✅ Async natif (async/await)
# ✅ Validation automatique (Pydantic)
# ✅ Doc OpenAPI générée automatiquement
# ✅ Type hints = contrat explicite
2. Routes async
FastAPI supporte les fonctions async et synchrones. Utilisez async def dès que vous faites des I/O (base de données, HTTP externe, fichiers).
from __future__ import annotations
import asyncio
from fastapi import FastAPI
app = FastAPI()
# ── async : libère le thread pendant l'attente I/O
@app.get("/tasks")
async def list_tasks() -> list[dict]:
# Simule une requête DB asynchrone
await asyncio.sleep(0) # cède le contrôle à l'event loop
return [{"id": 1, "title": "Apprendre FastAPI"}]
# ── sync : OK pour logique pure (pas d'I/O bloquant)
@app.get("/ping")
def ping() -> dict:
return {"status": "ok"}
# ── POST avec corps JSON
@app.post("/tasks", status_code=201)
async def create_task(data: dict) -> dict:
# data est automatiquement parsé depuis le corps JSON
return {"created": True, "data": data}
# ── PUT, PATCH, DELETE
@app.put("/tasks/{task_id}")
async def update_task(task_id: int, data: dict) -> dict:
return {"id": task_id, "updated": True}
@app.delete("/tasks/{task_id}", status_code=204)
async def delete_task(task_id: int) -> None:
pass # 204 No Content
3. Path & Query params
FastAPI extrait et valide automatiquement les paramètres de chemin et de requête à partir des annotations de type.
from __future__ import annotations
from fastapi import FastAPI, Path, Query
app = FastAPI()
# ── Path param : dans l'URL {task_id}
@app.get("/tasks/{task_id}")
async def get_task(
task_id: int = Path(..., ge=1, description="ID de la tâche"),
) -> dict:
# ge=1 → doit être >= 1 (validation automatique → 422 si invalide)
return {"id": task_id}
# ── Query params : après le ?
# GET /tasks?page=2&limit=10&status=done
@app.get("/tasks")
async def list_tasks(
page: int = Query(1, ge=1, description="Numéro de page"),
limit: int = Query(10, ge=1, le=100, description="Éléments par page"),
status: str | None = Query(None, description="Filtre par statut"),
q: str | None = Query(None, min_length=2, description="Recherche texte"),
) -> dict:
return {
"page": page,
"limit": limit,
"status": status,
"search": q,
}
# ── Combinaison path + query
# GET /projects/42/tasks?status=todo
@app.get("/projects/{project_id}/tasks")
async def get_project_tasks(
project_id: int = Path(..., ge=1),
status: str | None = Query(None),
) -> dict:
return {"project_id": project_id, "status": status}
task_id n'est pas un entier valide, FastAPI retourne automatiquement une réponse 422 Unprocessable Entity avec le détail de l'erreur.
4. Pydantic BaseModel
Pydantic v2 permet de définir des schémas de données avec validation, sérialisation et documentation automatiques.
from __future__ import annotations
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field, field_validator, model_validator
# ── Enum pour les statuts
class TaskStatus(str, Enum):
TODO = "todo"
IN_PROGRESS = "in_progress"
DONE = "done"
CANCELLED = "cancelled"
# ── Schéma de création
class TaskCreate(BaseModel):
title: str = Field(..., min_length=3, max_length=200, description="Titre de la tâche")
description: str | None = Field(None, max_length=2000)
status: TaskStatus = Field(TaskStatus.TODO, description="Statut initial")
project_id: int = Field(..., ge=1)
due_date: datetime | None = None
# ── Validator sur un champ
@field_validator("title")
@classmethod
def title_must_not_be_empty(cls, v: str) -> str:
if v.strip() == "":
raise ValueError("Le titre ne peut pas être vide")
return v.strip()
# ── Validator sur le modèle entier (cross-field)
@model_validator(mode="after")
def check_due_date_future(self) -> TaskCreate:
if self.due_date and self.due_date < datetime.now():
raise ValueError("La date d'échéance doit être dans le futur")
return self
# ── Schéma de réponse (inclut les champs auto-générés)
class TaskResponse(BaseModel):
id: int
title: str
description: str | None
status: TaskStatus
project_id: int
due_date: datetime | None
created_at: datetime
# Pydantic v2 : lire depuis les attributs ORM
model_config = {"from_attributes": True}
# ── Usage
task_data = {
"title": " Implémenter l'API ",
"project_id": 1,
"status": "todo",
}
task = TaskCreate(**task_data)
print(task.title) # "Implémenter l'API" (stripped)
print(task.model_dump()) # dict Python
print(task.model_dump_json()) # JSON string
5. response_model
response_model filtre et valide la réponse de sortie, garantissant que seuls les champs autorisés sont retournés.
from __future__ import annotations
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class UserCreate(BaseModel):
email: str
password: str # ← champ sensible, ne doit PAS sortir
class UserResponse(BaseModel):
id: int
email: str
# password absent → ne sera jamais exposé
# ── response_model filtre automatiquement la réponse
@app.post(
"/users",
response_model=UserResponse,
status_code=201,
summary="Créer un utilisateur",
tags=["Users"],
)
async def create_user(user: UserCreate) -> dict:
# Même si on retourne password, il sera filtré
return {"id": 42, "email": user.email, "password": user.password}
# → réponse : {"id": 42, "email": "..."} (password absent ✅)
# ── response_model_exclude_unset : n'inclure que les champs fournis
@app.get(
"/users/{user_id}",
response_model=UserResponse,
response_model_exclude_unset=True,
)
async def get_user(user_id: int) -> dict:
return {"id": user_id, "email": "alice@example.com"}
# ── Liste de réponses
@app.get("/users", response_model=list[UserResponse])
async def list_users() -> list[dict]:
return [{"id": 1, "email": "alice@example.com"}]
6. Depends()
L'injection de dépendances avec Depends() permet de partager de la logique entre plusieurs routes (auth, pagination, DB session…).
from __future__ import annotations
from fastapi import Depends, FastAPI, Query
app = FastAPI()
# ── Dépendance simple : paramètres de pagination réutilisables
class PaginationParams:
def __init__(
self,
page: int = Query(1, ge=1),
limit: int = Query(20, ge=1, le=100),
) -> None:
self.page = page
self.limit = limit
self.offset = (page - 1) * limit
# ── Injection dans la route
@app.get("/tasks")
async def list_tasks(pagination: PaginationParams = Depends()) -> dict:
return {
"page": pagination.page,
"limit": pagination.limit,
"offset": pagination.offset,
}
# ── Dépendance sous forme de fonction
async def get_db_session():
"""Fournit une session DB (pattern generator)."""
# En vrai : yield session; finally: session.close()
db = {"connected": True} # simulé
try:
yield db
finally:
pass # fermeture connexion
@app.get("/projects")
async def list_projects(db=Depends(get_db_session)) -> list:
# db est disponible ici
return []
# ── Dépendances chaînées
async def get_current_user(db=Depends(get_db_session)) -> dict:
return {"id": 1, "email": "alice@example.com"}
@app.get("/me")
async def get_profile(user: dict = Depends(get_current_user)) -> dict:
return user
7. HTTPException
HTTPException permet de retourner des erreurs HTTP standardisées avec un message de détail.
from __future__ import annotations
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
app = FastAPI()
# Données simulées
tasks_db: dict[int, dict] = {
1: {"id": 1, "title": "Apprendre FastAPI", "status": "todo"},
}
# ── HTTPException basique
@app.get("/tasks/{task_id}")
async def get_task(task_id: int) -> dict:
task = tasks_db.get(task_id)
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tâche {task_id} introuvable",
)
return task
# ── Avec headers (ex : WWW-Authenticate pour 401)
@app.get("/protected")
async def protected_route(token: str | None = None) -> dict:
if token != "secret":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token invalide ou manquant",
headers={"WWW-Authenticate": "Bearer"},
)
return {"message": "Accès autorisé"}
# ── Exception handler personnalisé
class BusinessError(Exception):
def __init__(self, code: str, message: str) -> None:
self.code = code
self.message = message
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError) -> JSONResponse:
return JSONResponse(
status_code=400,
content={"error": exc.code, "message": exc.message},
)
@app.post("/tasks")
async def create_task(data: dict) -> dict:
if not data.get("title"):
raise BusinessError("MISSING_TITLE", "Le titre est obligatoire")
return {"created": True}
8. Startup & Lifespan event
Depuis FastAPI 0.93+, le pattern recommandé est le lifespan context manager pour gérer l'initialisation et le nettoyage de l'application.
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI
# ── Ressources partagées (connexion DB, cache…)
db_connection: dict | None = None
app_config: dict = {}
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# ── STARTUP : code avant le yield
global db_connection, app_config
print("🚀 Démarrage de l'application...")
# Simuler connexion DB
db_connection = {"connected": True, "host": "localhost"}
app_config = {"env": "development", "debug": True}
print("✅ Base de données connectée")
yield # ← l'application tourne ici
# ── SHUTDOWN : code après le yield
print("🛑 Arrêt de l'application...")
db_connection = None
print("✅ Connexions fermées")
# Passer lifespan à l'application
app = FastAPI(lifespan=lifespan)
@app.get("/health")
async def health_check() -> dict:
return {
"status": "ok",
"db_connected": db_connection is not None,
"config": app_config.get("env"),
}
# ── Ancien style (deprecated mais encore supporté)
# @app.on_event("startup")
# async def startup_event():
# ... # à éviter dans les nouveaux projets