Formation FastAPI Python 2026
De zéro à la production — API IA complète

1. Lifespan context manager — chargement des modèles

Le pattern lifespan (FastAPI 0.95+) remplace les dépréciés on_startup/on_shutdown. Il garantit que le modèle IA est chargé une seule fois au démarrage, partagé entre tous les workers via un dictionnaire d’état.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from transformers import pipeline

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Démarrage : charger les modèles une seule fois
    app.state.classifier = pipeline(
        "text-classification",
        model="cardiffnlp/twitter-roberta-base-sentiment",
        device=-1  # CPU
    )
    app.state.summarizer = pipeline(
        "summarization",
        model="facebook/bart-large-cnn"
    )
    yield  # Application tourne ici
    # Arrêt : libérer les ressources GPU/mémoire
    del app.state.classifier
    del app.state.summarizer

app = FastAPI(
    title="DEV-AI NLP API",
    version="2.0.0",
    lifespan=lifespan
)

@app.post("/classify")
async def classify(text: str, request: Request):
    result = request.app.state.classifier(text)
    return {"label": result[0]["label"], "score": round(result[0]["score"], 4)}

2. Injection de dépendances — Depends()

Le système de dépendances FastAPI permet de partager une connexion DB, vérifier un token JWT, ou limiter le débit — sans dupliquer le code dans chaque route.

from fastapi import Depends, HTTPException, Header
from typing import Annotated

async def get_db():
    # Connexion DB avec garantie de fermeture
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def verify_api_key(
    x_api_key: Annotated[str, Header()]
) -> str:
    if x_api_key not in VALID_KEYS:
        raise HTTPException(403, "Clé API invalide")
    return x_api_key

@app.post("/summarize")
async def summarize(
    text: str,
    db: Annotated[Session, Depends(get_db)],
    key: Annotated[str, Depends(verify_api_key)]
):
    # db et key sont injectés automatiquement
    ...

3. Middleware stack — CORS, GZip, timing

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time

# CORS — restreindre aux origines autorisées en prod
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://dev-ai.fr"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

# GZip — compresse réponses > 1 Ko automatiquement
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Middleware custom — header X-Process-Time
@app.middleware("http")
async def add_process_time(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Process-Time"] = str(round(elapsed * 1000, 2)) + "ms"
    return response

1. BaseModel, Field, model_validator — Pydantic v2

Pydantic v2 (10× plus rapide que v1 grâce à Rust) change la syntaxe des validators. @field_validator remplace @validator, model_validator remplace @root_validator.

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Literal
import re

class NLPRequest(BaseModel):
    text: str = Field(
        ...,
        min_length=10,
        max_length=5000,
        description="Texte à analyser (10–5000 caractères)",
        examples=["FastAPI est excellent pour les API IA."]
    )
    task: Literal["sentiment", "summary", "ner"] = "sentiment"
    language: str = Field(default="fr", pattern=r"^[a-z]{2}$")

    @field_validator("text")
    @classmethod
    def no_html(cls, v: str) -> str:
        # Interdire les balises HTML (injection)
        if re.search(r"<[^>]+>", v):
            raise ValueError("HTML non autorisé dans le texte")
        return v.strip()

class BatchRequest(BaseModel):
    texts: list[str] = Field(..., min_length=1, max_length=32)
    task: Literal["sentiment", "summary"] = "sentiment"

    @model_validator(mode="after")
    def check_summary_length(self) -> "BatchRequest":
        # Summarization nécessite textes > 100 chars
        if self.task == "summary":
            for t in self.texts:
                if len(t) < 100:
                    raise ValueError("summary : texte trop court (< 100 chars)")
        return self

2. Response model — filtrer et sérialiser les sorties

from pydantic import BaseModel, computed_field
from datetime import datetime

class NLPResponse(BaseModel):
    task: str
    label: str
    score: float
    processing_ms: float

    @computed_field  # Pydantic v2 — champ calculé automatiquement
    @property
    def timestamp(self) -> str:
        return datetime.utcnow().isoformat()

    model_config = {"json_schema_extra": {
        "example": {
            "task": "sentiment", "label": "POSITIVE",
            "score": 0.9876, "processing_ms": 42.3
        }
    }}

# response_model filtre les champs non déclarés
@app.post("/analyze", response_model=NLPResponse)
async def analyze(payload: NLPRequest, request: Request):
    t0 = time.perf_counter()
    res = request.app.state.classifier(payload.text)[0]
    return NLPResponse(
        task=payload.task,
        label=res["label"],
        score=round(res["score"], 4),
        processing_ms=round((time.perf_counter() - t0) * 1000, 2)
    )

1. async def vs def — quand utiliser lequel

• async def : obligatoire pour les opérations I/O (appels HTTP, base de données, fichiers). Utiliser await avec httpx.AsyncClient, asyncpg, aiofiles.
• def : pour le code CPU-bound (inférence ML lourde). FastAPI l’exécute dans un thread pool via run_in_threadpool automatiquement.
• Piège : appeler une librairie bloquante (requests, time.sleep) dans un async def bloque l’event loop. Toujours utiliser asyncio.sleep et httpx async.

2. BackgroundTasks — tâches asynchrones post-réponse

from fastapi import BackgroundTasks
import asyncio

async def log_request(user_id: str, text: str, result: dict):
    # S'exécute après que la réponse est envoyée au client
    await asyncio.sleep(0)  # yield event loop
    await db.execute(
        "INSERT INTO logs(user_id,text,label) VALUES($1,$2,$3)",
        user_id, text[:200], result["label"]
    )

@app.post("/analyze", response_model=NLPResponse)
async def analyze(
    payload: NLPRequest,
    bg: BackgroundTasks,
    request: Request,
    user: User = Depends(get_current_user)
):
    result = request.app.state.classifier(payload.text)[0]
    # Réponse immédiate — log en arrière-plan
    bg.add_task(log_request, user.id, payload.text, result)
    return NLPResponse(**result)

3. WebSocket — streaming réponse LLM token par token

from fastapi import WebSocket, WebSocketDisconnect
import anthropic

@app.websocket("/ws/chat")
async def chat_ws(ws: WebSocket):
    await ws.accept()
    client = anthropic.AsyncAnthropic()
    try:
        while True:
            prompt = await ws.receive_text()

            # Stream token par token vers le client
            async with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            ) as stream:
                async for text in stream.text_stream():
                    await ws.send_text(text)

            # Signal de fin de stream
            await ws.send_text("[DONE]")

    except WebSocketDisconnect:
        pass  # Client déconnecté proprement

4. StreamingResponse — SSE pour clients HTTP classiques

from fastapi.responses import StreamingResponse
import json

@app.post("/stream")
async def stream_chat(payload: NLPRequest):
    client = anthropic.AsyncAnthropic()

    async def generate():
        async with client.messages.stream(
            model="claude-sonnet-4-6", max_tokens=1024,
            messages=[{"role": "user", "content": payload.text}]
        ) as stream:
            async for chunk in stream.text_stream():
                # Format SSE : "data: {...}\n\n"
                yield f"data: {json.dumps({'text': chunk})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"}
    )

1. OAuth2 + JWT avec python-jose

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from datetime import datetime, timedelta

SECRET_KEY = "256-bits-random-secret"  # os.environ["JWT_SECRET"]
ALGORITHM = "HS256"

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(sub: str, expires_delta: timedelta) -> str:
    payload = {
        "sub": sub,
        "exp": datetime.utcnow() + expires_delta,
        "iat": datetime.utcnow()
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(401)
    except JWTError:
        raise HTTPException(401, "Token invalide ou expiré")
    return await get_user(user_id)

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form.username, form.password)
    if not user:
        raise HTTPException(401)
    token = create_access_token(user.id, timedelta(minutes=30))
    return {"access_token": token, "token_type": "bearer"}

2. Rate limiting avec Slowapi

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/analyze")
@limiter.limit("20/minute")  # 20 requêtes/min par IP
async def analyze(request: Request, payload: NLPRequest):
    ...

# Rate limit différent selon le plan utilisateur
def get_user_limit(request: Request) -> str:
    user = get_user_from_token(request)
    return "100/minute" if user.is_premium else "10/minute"

limiter_dynamic = Limiter(key_func=get_remote_address,
                          default_limits=["500/day"])

3. Security headers — OWASP API Security Top 10

from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["Strict-Transport-Security"] = "max-age=63072000"
        response.headers["Referrer-Policy"] = "no-referrer"
        response.headers["Content-Security-Policy"] = "default-src 'none'"
        # Masquer le framework en prod
        response.headers.pop("Server", None)
        return response

app.add_middleware(SecurityHeadersMiddleware)

1. Tests avec pytest + httpx AsyncClient

FastAPI fournit TestClient (synchrone, wrapping httpx) et supporte AsyncClient pour les routes async. Les fixtures conftest.py permettent d’injecter une base de données de test et de mock les dépendances.

# conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.dependencies import get_current_user
from app.models import User

@pytest.fixture
async def async_client():
    # Override dépendance auth en test
    app.dependency_overrides[get_current_user] = lambda: User(id="test-user")
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client
    app.dependency_overrides.clear()

# test_analyze.py
@pytest.mark.asyncio
async def test_analyze_sentiment(async_client):
    r = await async_client.post("/analyze", json={
        "text": "FastAPI est excellent pour les API IA modernes.",
        "task": "sentiment"
    })
    assert r.status_code == 200
    data = r.json()
    assert "label" in data
    assert 0 <= data["score"] <= 1

async def test_rate_limit(async_client):
    # Dépasser la limite — doit retourner 429
    responses = [await async_client.post("/analyze", json={
        "text": "test " * 10, "task": "sentiment"
    }) for _ in range(25)]
    codes = [r.status_code for r in responses]
    assert 429 in codes

2. Dockerfile multi-stage — image de production minimale

# ── Stage 1 : builder ─────────────────────────────────────
FROM python:3.12-slim AS builder
WORKDIR /build

# Installer uv (pip ultra-rapide)
RUN pip install uv
COPY requirements.txt .
RUN uv pip install --system --no-cache -r requirements.txt

# ── Stage 2 : production ──────────────────────────────────
FROM python:3.12-slim
WORKDIR /app

# Copier uniquement les dépendances installées
COPY --from=builder /usr/local/lib/python3.12 /usr/local/lib/python3.12
COPY --from=builder /usr/local/bin /usr/local/bin

# Utilisateur non-root (sécurité)
RUN adduser --disabled-password --gecos "" appuser
USER appuser

COPY --chown=appuser:appuser ./app ./app

EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s \
  CMD curl -f http://localhost:8000/healthz || exit 1

CMD ["uvicorn", "app.main:app", \
     "--host", "0.0.0.0", "--port", "8000", \
     "--workers", "4", "--log-level", "info"]

3. GitHub Actions CI/CD — test → lint → build → push

# .github/workflows/ci.yml
name: CI/CD DEV-AI API

on:
  push:
    branches: [main]

jobs:
  test-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with: { python-version: "3.12" }

      - name: Install + lint
        run: |
          pip install uv
          uv pip install --system -r requirements-dev.txt
          ruff check app/
          mypy app/ --ignore-missing-imports

      - name: Run tests
        run: pytest tests/ -v --cov=app --cov-report=xml

      - name: Build & push Docker image
        if: success()
        run: |
          docker build -t ghcr.io/${{ github.repository }}:${{ github.sha }} .
          echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
          docker push ghcr.io/${{ github.repository }}:${{ github.sha }}

      - name: Deploy via SSH
        run: |
          ssh -o StrictHostKeyChecking=no ${{ secrets.VPS_USER }}@${{ secrets.VPS_HOST }} \
            "docker pull ghcr.io/${{ github.repository }}:${{ github.sha }} && \
             docker stop devai-api || true && \
             docker run -d --name devai-api -p 8000:8000 --env-file .env \
               ghcr.io/${{ github.repository }}:${{ github.sha }}"

1. Logs structurés JSON avec structlog

Les logs structurés (JSON) sont indexés par Loki, Datadog ou CloudWatch sans configuration supplémentaire. structlog injecte automatiquement le contexte (request_id, user_id, latence).

import structlog
import uuid

structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)
logger = structlog.get_logger()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    req_id = str(uuid.uuid4()[:8])
    t0 = time.perf_counter()
    response = await call_next(request)
    latency = round((time.perf_counter() - t0) * 1000, 2)
    logger.info(
        "request",
        req_id=req_id,
        method=request.method,
        path=request.url.path,
        status=response.status_code,
        latency_ms=latency,
        ip=request.client.host
    )
    response.headers["X-Request-ID"] = req_id
    return response

2. Métriques Prometheus — compteurs, histogrammes

from prometheus_client import (
    Counter, Histogram, Gauge,
    generate_latest, CONTENT_TYPE_LATEST
)
from fastapi.responses import Response

# Compteur de requêtes par endpoint et status
REQUEST_COUNT = Counter(
    "api_requests_total",
    "Nombre total de requêtes",
    ["method", "endpoint", "status"]
)

# Histogramme latences (buckets en secondes)
REQUEST_LATENCY = Histogram(
    "api_request_duration_seconds",
    "Latence des requêtes",
    ["endpoint"],
    buckets=[.05, .1, .25, .5, 1, 2.5, 5]
)

# Gauge — modèles chargés en mémoire
MODELS_LOADED = Gauge(
    "models_loaded_total", "Modèles actifs"
)

@app.get("/metrics", include_in_schema=False)
async def metrics():
    # Endpoint scrapé par Prometheus toutes les 15s
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )

3. Healthcheck endpoint — liveness + readiness

from pydantic import BaseModel
import psutil, time

class HealthResponse(BaseModel):
    status: str
    version: str
    uptime_s: float
    memory_mb: float
    models_loaded: list[str]

START_TIME = time.time()

@app.get("/healthz", response_model=HealthResponse)
async def healthz(request: Request):
    mem = psutil.Process().memory_info().rss / 1024**2
    models = [k for k in vars(request.app.state)
              if not k.startswith("_")]
    return HealthResponse(
        status="ok",
        version=app.version,
        uptime_s=round(time.time() - START_TIME, 1),
        memory_mb=round(mem, 1),
        models_loaded=models
    )

# Readiness — vérifie que les modèles sont bien chargés
@app.get("/readyz", status_code=200)
async def readyz(request: Request):
    if not hasattr(request.app.state, "classifier"):
        raise HTTPException(503, "Modèles non chargés")
    return {"ready": True}