Aller au contenu principal

Module 6 — Requêtes asynchrones et tâches de fond

FastAPI accepte def et async def pour définir une route. Le débutant en déduit qu'il faut toujours écrire async def — c'est neuf, ça sonne moderne — et se retrouve avec un service dix fois plus lent que la version synchrone. Ce module explique quand async aide réellement, pourquoi il peut nuire, et comment lancer un travail long sans bloquer le service.

Le modèle d'exécution : une boucle unique, plusieurs travailleurs​

Uvicorn tourne sur asyncio, une boucle d'événement qui alterne entre les requêtes en cours. Tant que le code d'une route ne fait qu'attendre — lecture réseau, écriture disque asynchrone, appel HTTP — la boucle rend la main aux autres requêtes et le service traite mille clients avec un seul processus.

Le mot-clé qui fait rendre la main est await. Un await est le seul point de la fonction où d'autres tâches peuvent s'insérer.

import httpx

@app.get("/exemple")
async def exemple():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://exemple.fr/api/config")
    return r.json()

Ici, la fonction exemple traite pendant la latence réseau les requêtes d'autres appelants. C'est l'archétype d'un usage juste de async.

Où async ne sert à rien : le calcul CPU​

Un modèle scikit-learn tourne sur le processeur ; il n'attend rien, il calcule. Aucun await ne s'y trouve, donc aucune insertion possible dans la boucle. Écrire :

@app.post("/predict")
async def predict(dossier: DossierAbonne):
    # aucun await : la boucle est bloquée pendant la durée de predict_proba
    return modele.predict_proba([dossier_vers_vecteur(dossier)])[0, 1]

est plus mauvais qu'écrire la même chose en def. Uvicorn, voyant async def, exécute la fonction directement dans la boucle d'événement, et la boucle reste bloquée pendant les 20 ms de predict_proba. Aucune autre requête n'avance. Sur 100 requêtes simultanées, chacune attend la fin des 99 autres : le débit s'effondre.

La bonne règle : def par défaut, async seulement s'il y a await​

Écrire def (sans async) sur une route de scoring déclenche un comportement radicalement différent. FastAPI voit la fonction synchrone et l'exécute dans un fil d'exécution distinct de la boucle d'événement, tiré d'un pool géré par Starlette. Uvicorn continue de traiter d'autres requêtes pendant que le fil calcule.

@app.post("/predict")  # def, pas async def
def predict(dossier: DossierAbonne):
    return modele.predict_proba([dossier_vers_vecteur(dossier)])[0, 1]

Sur le fil rouge, la version def traite 300 requêtes par seconde derrière trois travailleurs Uvicorn ; la version async def avec le même modèle et sans await chute à 30. C'est le facteur 10 annoncé plus haut.

La règle inverse s'applique aux appels I/O : si la route lit une base de données via un pilote asynchrone (asyncpg, aiohttp), on écrit async def et on await. Le meilleur signal pour choisir est la présence d'au moins un await dans le corps.

Forcer un calcul bloquant dans un fil : run_in_threadpool​

On rencontre parfois une route qui est presque entièrement asynchrone (elle attend une base) mais qui contient un petit bloc CPU. On préfère alors garder async def et sous-traiter le bloc CPU au pool de fils.

from fastapi.concurrency import run_in_threadpool

@app.post("/predict")
async def predict(dossier: DossierAbonne):
    x = dossier_vers_vecteur(dossier).reshape(1, -1)
    # sort du contexte async, ne bloque pas la boucle :
    proba = await run_in_threadpool(lambda: float(modele.predict_proba(x)[0, 1]))
    return {"probabilite_resiliation": proba}

run_in_threadpool exécute la fonction passée dans le même pool que celui utilisé pour les routes synchrones. Le service reste réactif pendant le calcul.

Tâches de fond : rendre la main sans bloquer l'appelant​

Un besoin fréquent : l'appelant envoie un fichier CSV de 20 000 lignes, on lui répond « bien reçu, on te ping quand c'est prêt » et le calcul continue sans lui. FastAPI expose BackgroundTasks pour les cas simples.

from fastapi import BackgroundTasks, UploadFile
import pandas as pd
from pathlib import Path

def scorer_fichier(chemin_entree: Path, chemin_sortie: Path, version_modele: str) -> None:
    df = pd.read_csv(chemin_entree)
    X = df[VARIABLES].to_numpy(dtype=np.float64)
    df["probabilite_resiliation"] = modele.predict_proba(X)[:, 1]
    df["version_modele"] = version_modele
    df.to_csv(chemin_sortie, index=False)

@app.post("/predict/fichier", status_code=202)
async def predict_fichier(
    fichier: UploadFile,
    background: BackgroundTasks,
):
    lot_id = str(uuid.uuid4())
    chemin_in = Path(f"/tmp/{lot_id}.in.csv")
    chemin_out = Path(f"/tmp/{lot_id}.out.csv")
    chemin_in.write_bytes(await fichier.read())
    background.add_task(scorer_fichier, chemin_in, chemin_out, "churn-2026-08-30")
    return {"lot_id": lot_id, "statut": "en_cours", "sortie": str(chemin_out)}

Le code de statut 202 Accepted signale à l'appelant que la requête est prise en compte mais que le résultat n'est pas encore là. L'appelant consulte ensuite /predict/fichier/{lot_id} (à écrire séparément) pour récupérer le statut ou télécharger la sortie.

Les limites de BackgroundTasks​

BackgroundTasks s'exécute dans le même processus que le service. Trois conséquences.

Un redémarrage tue la tâche. Le service qui reçoit un SIGTERM de l'orchestrateur laisse tomber les tâches en cours ; le fichier de 20 000 lignes est perdu. Il faut soit un temps de grâce (Kubernetes terminationGracePeriodSeconds), soit une reprise idempotente (on retente le lot au démarrage).

Le calcul concurrence les requêtes en direct. Une tâche de fond qui occupe un cœur ralentit les requêtes qui arrivent en même temps. Sur un service avec trois travailleurs, trois fichiers en fond consomment tout ; plus rien ne répond en direct.

Il n'y a pas de file entre plusieurs conteneurs. Deux réplicas n'échangent rien : une tâche envoyée à l'un ne peut pas être reprise par l'autre.

Pour ces trois raisons, BackgroundTasks convient pour un envoi de mail, la génération d'un PDF léger, une notification. Pour un scoring de plusieurs minutes, on passe à une file de travail externe.

Vers une file de travail : Celery, RQ, ARQ, Dramatiq​

Le patron classique découple deux services : l'API FastAPI qui reçoit la requête et pousse un message dans une file (Redis, RabbitMQ), et un travailleur séparé qui consomme la file et exécute le calcul.

# côté API
from redis import Redis
from rq import Queue

file = Queue(connection=Redis(host="redis"))

@app.post("/predict/fichier", status_code=202)
async def predict_fichier(fichier: UploadFile):
    lot_id = str(uuid.uuid4())
    chemin_in = Path(f"/lots/{lot_id}.in.csv")
    chemin_in.write_bytes(await fichier.read())
    file.enqueue(scorer_fichier, str(chemin_in), lot_id)
    return {"lot_id": lot_id, "statut": "en_file_attente"}

L'API redevient très rapide, les travailleurs se dimensionnent indépendamment, et la file persiste les tâches : un redémarrage n'en perd aucune. C'est la voie recommandée dès que la tâche dépasse une minute ou que le volume dépasse quelques par heure.

Ne pas ouvrir de connexion async dans une route def​

Une erreur symétrique du async def bloquant : appeler une bibliothèque asynchrone (comme httpx.AsyncClient) depuis une route def. Ça ne compile pas, mais on peut être tenté d'utiliser asyncio.run(...) à l'intérieur — ce qui ouvre une deuxième boucle d'événement, en concurrence avec celle de Uvicorn, et provoque des blocages difficiles à diagnostiquer. Choisir un régime pour la route et s'y tenir.

Mesurer avant d'optimiser

Le choix def vs async def a un effet mesurable sur le débit ; le mesurer avec Locust (module 10) sur les deux versions du service évite les convictions théoriques. Sur une route qui appelle un modèle scikit-learn, la version def est presque toujours la bonne réponse.

En résumé​

  • async def ne sert qu'aux routes qui attendent vraiment (I/O réseau, base de données via pilote async, appels HTTP).
  • Pour une route de scoring bloquée par le CPU, écrire def — FastAPI la met dans un fil d'exécution et la boucle reste libre.
  • BackgroundTasks convient aux tâches courtes (mail, notification) ; pour un scoring de plusieurs minutes, passer à une file externe (Celery, RQ, Dramatiq).
  • Renvoyer 202 Accepted avec un lot_id sur les traitements asynchrones ; laisser l'appelant venir chercher le résultat.

Le module 7 protège l'accès au service par un jeton et introduit la limitation de débit.