Celery et FastAPI : rendre la main avant d’avoir fait le travail

Le besoin était banal : recevoir des photos de tickets de caisse, en extraire les lignes, rendre un résultat exploitable. Le premier prototype faisait tout dans la requête. Il tenait tant que les images étaient petites et l’affluence faible, c’est-à-dire jusqu’au jour de la mise en service. Une extraction demande plusieurs secondes ; une requête HTTP qui dure plusieurs secondes occupe un travailleur, et une centaine d’envois simultanés suffit à ce que le serveur cesse de répondre : y compris aux requêtes qui n’avaient rien demandé de lourd.

FastAPI propose `BackgroundTasks`, et c’est le piège le plus tentant. La tâche part bien après la réponse, mais dans le même processus : elle dispute le processeur aux requêtes en cours, et si le processus s’arrête (un déploiement, un redémarrage, une saturation mémoire) le travail est perdu sans trace. C’est acceptable pour envoyer un courriel, jamais pour un traitement qu’on a promis à un utilisateur.

Le serveur accepte, dépose un message, et rend la main. Un travailleur séparé consomme la file. L’essentiel tient dans une règle qu’on oublie systématiquement la première fois : **l’image ne passe pas par la file**. Un courtier de messages transporte des instructions, pas des mégaoctets : on écrit l’image dans un stockage objet et l’on ne fait circuler que sa clé.

from celery import Celery
from fastapi import FastAPI, UploadFile
from fastapi.responses import JSONResponse

celery = Celery("receipts", broker="redis://redis:6379/0", backend="redis://redis:6379/1")
api = FastAPI()

@api.post("/receipts", status_code=202)
async def submit(upload: UploadFile) -> JSONResponse:
    # The image goes to object storage; the queue only ever gets a key.
    key = await store(upload)
    task = extract.delay(key)
    # 202 and not 200: the request is accepted, it is not fulfilled.
    return JSONResponse({"id": task.id}, status_code=202)

@api.get("/receipts/{task_id}")
def status(task_id: str) -> dict:
    r = celery.AsyncResult(task_id)
    return {"state": r.state, "result": r.result if r.successful() else None}

Les valeurs par défaut de Celery sont calibrées pour des tâches courtes et nombreuses. Un traitement d’image est l’inverse : long et coûteux. Trois d’entre elles se retournent contre vous, et il vaut mieux les changer avant l’incident que pendant.

celery.conf.update(
    # Defaults to 4: a worker reserves four messages ahead and holds them while
    # processing one. On long tasks, three images wait behind the current one
    # while another worker sits idle.
    worker_prefetch_multiplier=1,
    # Defaults to False: the message is acknowledged on receipt, so it is lost
    # if the worker dies midway. At True it is only acknowledged at the end —
    # at the cost of a possibly repeated execution.
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    # With no limit, one pathological image holds a worker indefinitely. The
    # soft limit raises a catchable exception; the hard one kills.
    task_soft_time_limit=110,
    task_time_limit=120,
)

Le troisième réglage a une conséquence qu’il faut assumer plutôt que découvrir : à partir du moment où un message peut être rejoué, la tâche doit pouvoir l’être aussi. Écrire le résultat sous une clé dérivée de l’entrée plutôt qu’en ajout, et vérifier avant de traiter, coûte trois lignes et évite un doublon dans un rapport client.

Je ne dis rien du choix du courtier. Redis tient parfaitement ce rôle tant qu’on accepte sa promesse de livraison, plus faible que celle d’un courtier transactionnel ; au-delà d’un certain enjeu la question se repose, et elle se repose avec des arguments que je n’ai pas eu à trancher ici. Je laisse aussi de côté la remontée d’avancement : le pourcentage qui bouge pendant que l’utilisateur attend. C’est faisable et c’est presque toujours du travail perdu : ce que veut celui qui a déposé une image, c’est savoir quand revenir, pas regarder une barre.

• Celery · configuration
• FastAPI · background tasks