Quand ton API appelle un modèle d'IA, génère une image ou construit un PDF, l'utilisateur attend. Dix secondes, trente secondes, parfois une minute. Pendant tout ce temps, la requête HTTP reste ouverte, ton serveur est occupé, et ton visiteur est à deux doigts de fermer l'onglet. La solution ne s'appelle ni patience ni multithreading bricolé : elle s'appelle Celery. Une file d'attente de tâches qui découple ton API de tout ce qui prend du temps. Tu réponds immédiatement, un worker s'occupe du travail dans son coin, et tu récupères le résultat quand il est prêt.
Celery est l'outil standard de Python pour les tâches en arrière-plan. Il n'exécute rien lui-même : il orchestre. Ton code dépose une tâche dans une file, des workers la consomment, et tout le monde reste fluide. On va monter ça ensemble, avec du vrai code.
Étape 1 — Installe Celery et son broker
Celery a besoin d'un broker : un intermédiaire qui stocke les tâches en attendant qu'un worker les récupère. Le plus courant et le plus simple, c'est Redis.
pip install celery redisRedis joue deux rôles ici : broker (la file d'attente) et backend (l'endroit où sont stockés les résultats des tâches). Un seul service, deux usages.
Étape 2 — Écris ta première tâche
Crée un fichier tasks.py :
from celery import Celery
app = Celery(
"iancr",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/0",
)
@app.task
def additionner(a, b):
return a + bLe décorateur @app.task transforme une fonction ordinaire en tâche distribuable. Une fois enregistrée, elle peut partir dans la file et être exécutée par n'importe quel worker, sur n'importe quelle machine.
Étape 3 — Lance un worker
Dans un second terminal :
celery -A tasks worker --loglevel=infoLe worker démarre et liste les tâches qu'il connaît. C'est un processus séparé qui attend du travail. Important : le worker et ton script ne partagent rien d'autre que Redis. C'est ça, le découplage.
Étape 4 — Déclenche la tâche
De retour dans ton script Python :
from tasks import additionner
resultat = additionner.delay(3, 4)
print(resultat.id) # l'identifiant unique de la tâche
print(resultat.get(timeout=10)) # 7.delay() envoie la tâche dans la file et revient immédiatement. Le calcul n'est pas fait au moment de l'appel : il est mis en attente. .get() bloque jusqu'au résultat, avec un timeout pour ne jamais rester pendu.
Étape 5 — Le vrai cas d'usage : un appel IA capricieux
Là où Celery brille, c'est sur les tâches lentes et fragiles. Un appel à une API d'IA peut durer vingt secondes, ou échouer sur un rate limit. Voici une tâche qui interroge un modèle avec trois tentatives automatiques :
import httpx
from celery import Celery
app = Celery("iancr", broker="redis://localhost:6379/0", backend="redis://localhost:6379/0")
@app.task(bind=True, max_retries=3, default_retry_delay=5)
def resumer(self, texte: str) -> str:
try:
reponse = httpx.post(
"https://api.exemple.com/resumer",
json={"texte": texte},
timeout=60,
)
reponse.raise_for_status()
return reponse.json()["resume"]
except Exception as exc:
raise self.retry(exc=exc)bind=True donne accès à self, et donc à self.retry(). En cas d'échec, la tâche retourne dans la file et réessaie cinq secondes plus tard, jusqu'à trois fois. Ton appel fragile devient résilient, sans une seule boucle écrite à la main.
Étape 6 — Réponds tout de suite côté API
Le bénéfice se voit quand on branche ça sur une API. Avec FastAPI, tu réponds en quelques millisecondes au lieu de faire patienter trente secondes :
from fastapi import FastAPI
from tasks import resumer
app = FastAPI()
@app.post("/resumer")
def resumer_endpoint(texte: str):
tache = resumer.delay(texte)
return {"task_id": tache.id, "status": "en cours"}L'utilisateur reçoit un identifiant de tâche instantanément. La requête ne traîne plus : le gros du travail est parti ailleurs.
Étape 7 — Récupère le résultat
Ton frontend, ou l'utilisateur, revient plus tard chercher le résultat :
from celery.result import AsyncResult
from tasks import app
def statut(task_id: str):
res = AsyncResult(task_id, app=app)
if res.state == "SUCCESS":
return {"status": "terminé", "resultat": res.result}
return {"status": res.state}PENDING, STARTED, SUCCESS, FAILURE : tu connais l'état de chaque tâche à tout moment, et le résultat est disponible dès qu'il est prêt.
Étape 8 — Enchaîne les tâches
Une tâche peut alimenter la suivante. Résumer un texte, le traduire, l'envoyer par email : trois étapes, une seule ligne.
from celery import chain
from tasks import resumer, traduire, envoyer_email
flux = chain(resumer.s(texte) | traduire.s("anglais") | envoyer_email.s("moi@exemple.fr"))
flux()La signature .s() prépare la tâche sans l'exécuter. Le pipe enchaîne : la sortie de l'une devient l'entrée de la suivante. traduire et envoyer_email sont deux autres tâches @app.task, définies sur le même modèle. Tu viens de construire un mini-pipeline, sans orchestrateur maison.
Celery ne rend pas ton code plus rapide. Il rend ton API plus réactive en déplaçant la lenteur ailleurs : dans des workers qui encaissent le travail pendant que l'utilisateur vaque à ses occupations. C'est la différence entre une barre de chargement qui n'en finit pas et une réponse immédiate. Une fois qu'on a goûté au .delay(), on ne revient plus aux appels bloquants.
Auteur
Krugz
Développeur fullstack
Développeur fullstack qui tente, tant bien que mal, de suivre l'actualité de l'IA au milieu d'une marée d'informations imbuvables et des milliers d'outils et de modèles qui sortent chaque jour.
