Creer une API Python moderne avec FastAPI

FastAPI, Pydantic, SQLAlchemy, bonnes pratiques

⚡

FastAPI - Bases

Hello World- FastAPI se lance avec un serveur ASGI (uvicorn, different du serveur classique de Flask) et genere automatiquement une documentation interactive a partir du code : aucun fichier a ecrire en plus.

from fastapi import FastAPI
app = FastAPI()          # instance principale de l'application
@app.get("/")           # associe cette fonction a une requete GET sur /
def home():
    return {"message": "Hello"}   # FastAPI convertit automatiquement en JSON
# Lancer le serveur :
# uvicorn main:app --reload      # --reload = redemarre automatiquement a chaque modif
# Documentation auto : http://localhost:8000/docs

Parametres de chemin et de requete- Un parametre de chemin (dans l'URL, ex: {id}) est obligatoire. Un parametre de requete (avec une valeur par defaut, ex: actif=True) est optionnel et se passe avec ?actif=false dans l'URL. FastAPI verifie automatiquement le type grace au type hint : une erreur 422 est renvoyee si 'id' n'est pas un entier.

from fastapi import FastAPI
app = FastAPI()
@app.get("/utilisateur/{id}")
def get_utilisateur(id: int, actif: bool = True):
    # id       : parametre de chemin, obligatoire -> /utilisateur/5
    # actif    : parametre de requete, optionnel  -> /utilisateur/5?actif=false
    return {"id": id, "actif": actif}

Validation avec Pydantic v2- Un modele Pydantic (BaseModel) decrit la forme attendue des donnees JSON envoyees par le client. Si un champ manque ou a le mauvais type, FastAPI renvoie automatiquement une erreur 422 detaillee, sans qu'on ait a ecrire la moindre verification manuelle.

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Utilisateur(BaseModel):
    nom: str      # champ obligatoire, doit etre une chaine
    age: int      # champ obligatoire, doit etre un entier
@app.post("/utilisateur")
def creer_utilisateur(utilisateur: Utilisateur):
    # 'utilisateur' est deja valide a ce stade : plus besoin de le verifier
    return {"message": f"{utilisateur.nom} cree", "age": utilisateur.age}

Reponses HTTP & statuts- HTTPException permet d'arreter le traitement et de renvoyer immediatement un code d'erreur HTTP standard avec un message explicatif, plutot que de laisser l'API planter ou renvoyer un succes incorrect.

from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/produit/{id}")
def get_produit(id: int):
    if id != 1:
        # arrete immediatement la fonction et renvoie une erreur 404
        raise HTTPException(status_code=404, detail="Produit introuvable")
    return {"id": 1, "nom": "Clavier"}

Codes de statut HTTP courants- Un aide-memoire des codes les plus utilises dans une API REST : le premier chiffre indique la categorie (2xx = succes, 4xx = erreur du client, 5xx = erreur du serveur).

200 OK                    # succes standard (GET reussi)
201 Created               # ressource creee avec succes (POST reussi)
204 No Content            # succes, mais rien a renvoyer (DELETE reussi)
400 Bad Request           # requete mal formee (donnees invalides)
401 Unauthorized          # authentification manquante ou invalide
403 Forbidden             # authentifie, mais pas les droits suffisants
404 Not Found             # ressource demandee introuvable
422 Unprocessable Entity  # erreur de validation Pydantic (type/champ incorrect)
500 Internal Server Error # erreur imprevue cote serveur

Activer CORS- Par securite, un navigateur bloque par defaut les requetes faites depuis un site web vers une API sur un domaine ou un port different (ex: site sur le port 3000 qui appelle l'API sur le port 8000). CORSMiddleware autorise explicitement ces echanges cross-origin.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],       # "*" = autorise tous les domaines (a restreindre en prod)
    allow_credentials=True,     # autorise l'envoi de cookies/identifiants
    allow_methods=["*"],       # autorise GET, POST, PUT, DELETE...
    allow_headers=["*"],       # autorise tous les en-tetes HTTP personnalises
)

📁

Structure d'un projet API

Arborescence recommandee- Separer le code par role (modeles de base de donnees, schemas de validation, routes) evite d'avoir un unique fichier main.py de plusieurs milliers de lignes, et facilite le travail a plusieurs.

app/
 ├── main.py             # point d'entree : cree l'app et inclut les routers
 ├── database.py         # connexion a la base de donnees (engine, session)
 ├── models/             # modeles SQLAlchemy (structure des tables en BDD)
 │    └── utilisateur.py
 ├── schemas/            # modeles Pydantic (structure des donnees de l'API)
 │    └── utilisateur.py
 ├── routers/            # routes regroupees par ressource (utilisateur, produit...)
 │    └── utilisateur.py
 └── utils/              # fonctions transverses (securite, envoi d'email...)

Separer routes et modeles (APIRouter)- APIRouter permet de definir un groupe de routes dans un fichier a part, puis de les rattacher a l'application principale. 'prefix' evite de repeter '/utilisateur' devant chaque route, et 'tags' regroupe visuellement ces routes dans la documentation /docs.

# routers/utilisateur.py
from fastapi import APIRouter
router = APIRouter(prefix="/utilisateur", tags=["Utilisateur"])
@router.get("/{id}")   # devient GET /utilisateur/{id} grace au prefix
def get_user(id: int):
    return {"id": id}

Importer un router- include_router() rattache les routes definies ailleurs a l'application principale : main.py reste court et lisible, meme avec des dizaines de routes au total.

from fastapi import FastAPI
from routers.utilisateur import router as utilisateur_router
app = FastAPI()
app.include_router(utilisateur_router)   # toutes les routes de ce router sont activees

🗄️

SQLAlchemy + FastAPI

Connexion a la base- L'engine gere la connexion physique a la base de donnees. sessionmaker cree des 'sessions' (des conversations avec la BDD) a la demande. check_same_thread=False est specifique a SQLite, necessaire car FastAPI peut traiter plusieurs requetes sur des threads differents.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
DATABASE_URL = "sqlite:///./test.db"   # base SQLite locale (fichier test.db)
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
class Base(DeclarativeBase):
    # classe de base dont heriteront tous les modeles (tables) de l'application
    pass

Modele SQLAlchemy- Un modele SQLAlchemy decrit une table de la base de donnees : __tablename__ nomme la table, chaque Column() devient une colonne. primary_key=True identifie la cle primaire, index=True accelere les recherches sur ce champ.

from sqlalchemy import Column, Integer, String
from database import Base
class Utilisateur(Base):
    __tablename__ = "utilisateurs"     # nom de la table en base
    id = Column(Integer, primary_key=True, index=True)  # cle primaire, indexee
    nom = Column(String, index=True)                    # colonne texte, indexee
    age = Column(Integer)                               # colonne numerique

Schema Pydantic- On separe volontairement le modele de base de donnees (SQLAlchemy) du modele expose par l'API (Pydantic) : cela evite d'exposer directement la structure interne de la BDD, et permet de choisir precisement quels champs sont visibles/modifiables via l'API. from_attributes=True (Pydantic v2, ex-orm_mode) autorise a construire ce schema directement depuis un objet SQLAlchemy.

from pydantic import BaseModel
class UtilisateurSchema(BaseModel):
    nom: str
    age: int
    class Config:
        from_attributes = True   # permet UtilisateurSchema.from_orm(objet_sqlalchemy)

CRUD FastAPI + SQLAlchemy- get_db() ouvre une session, la met a disposition de la route via Depends() (injection de dependance), puis la ferme automatiquement grace a yield/finally, meme en cas d'erreur. commit() valide les changements en base, refresh() recharge l'objet avec les valeurs generees par la BDD (comme l'id auto-incremente).

from fastapi import Depends, APIRouter
from sqlalchemy.orm import Session
from database import SessionLocal
from models.utilisateur import Utilisateur
from schemas.utilisateur import UtilisateurSchema
router = APIRouter(prefix="/utilisateur", tags=["Utilisateur"])
def get_db():
    db = SessionLocal()
    try:
        yield db          # fournit la session a la route le temps de la requete
    finally:
        db.close()        # fermee systematiquement, meme si une erreur survient
@router.post("/")
def creer_utilisateur(data: UtilisateurSchema, db: Session = Depends(get_db)):
    user = Utilisateur(nom=data.nom, age=data.age)  # cree l'objet SQLAlchemy
    db.add(user)      # prepare l'insertion
    db.commit()       # ecrit reellement en base de donnees
    db.refresh(user)  # recupere l'id genere automatiquement par la BDD
    return user