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}
Validation automatique : Si 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
Bonne pratique : Le lifespan est l'endroit idéal pour initialiser les pools de connexions DB, charger des modèles ML, configurer le cache Redis, etc.
← Module 02 Module 04 — SQLAlchemy →