Le céleri peut être intimidant à apprendre. Bien que sa documentation soit complète, elle a tendance à ignorer les bases.

Cet article définira quatre des principaux concepts de Celery, discutera de la relation entre Celery et Kombu et utilisera quelques exemples de code pour illustrer comment Celery pourrait être utile dans des applications réelles. Les exemples utiliseront le framework Web Django et son décorateur @shared_task, mais les concepts sont également applicables à Flask, FastAPI et autres.

Tâches, courtiers, travailleurs et backends

Vous aurez du mal à trouver une place dans la documentation actuelle de Celery qui énonce clairement ce qu'il considère comme un courtier ou backend, mais en creusant suffisamment, vous pouvez trouver et déduire des définitions.

Vous trouverez ci-dessous les concepts que vous devez connaître avant de commencer à utiliser le céleri.

Tâche

Une tâche est un travail que Celery effectuera de manière asynchrone (dans ce contexte, c'est un mot sophistiqué pour "pas immédiatement"). Dans une application Web, une tâche peut consister à envoyer un e-mail après qu'un utilisateur a soumis un formulaire. L'envoi d'un e-mail peut prendre plusieurs secondes, et forcer un utilisateur à attendre l'envoi d'un e-mail avant de le rediriger peut ralentir une application.

Les tâches sont définies à l'aide de décorateurs dans Celery. Ci-dessous, nous utilisons le décorateur @shared_task pour transformer send_thank_you_email() en une tâche Celery qui peut être utilisée dans le gestionnaire de soumission de formulaire submit_feedback().

from config.celery import shared_task
from django.core.mail import send_mail
from django.shortcuts import render, redirect
from feedback.forms import FeedbackForm

@shared_task
def send_thank_you_email(email_address):
    send_mail(
        "Thank you for your feedback!",
        "We appreciate your input.",
        "[email protected]",
        [email_address],
    )

def submit_feedback(request):
    if request.method == "POST":
        form = FeedbackForm(request.POST)
        if form.is_valid():
            form.save()

            # Push the task to the broker using the delay() method.
            send_thank_you_email.delay(form.cleaned_data["email"])

            return redirect("/thank-you/")
    else:
        form = FeedbackForm()

    return render(request, "feedback.html", {"form": form})

Lorsqu'une tâche est définie à l'aide d'un décorateur dans Celery, elle ajoute une méthode delay() à la tâche. Vous pouvez voir la tâche send_thank_you_email appeler la méthode delay() dans l'exemple ci-dessus une fois le formulaire enregistré avec succès. Lorsque delay() est appelé, il enverra la tâche send_thank_you_email et ses données au broker où elle est stockée et sera ensuite exécutée par un worker, auquel cas l'utilisateur être envoyé par e-mail.

L'avantage de transférer le travail vers Celery devient plus évident si vous devez envoyer des e-mails supplémentaires après avoir enregistré le formulaire. Par exemple, vous souhaiterez peut-être envoyer un e-mail à l'équipe de support client pour l'informer qu'elle a reçu de nouveaux commentaires. Avec le céleri, cela n’ajoute presque aucun temps de réponse supplémentaire.

Les tâches de céleri permettent également une configuration avancée supplémentaire. En cas d'échec de l'envoi d'un e-mail, vous pouvez coder votre tâche pour réessayer automatiquement et configurer des paramètres tels que max_retries, retry_backoff, retry_jitter, etc.

Courtier

Le glossaire des propositions d'amélioration du céleri contient les informations suivantes à propos des courtiers de messages :

Enterprise Integration Patterns définit un message Broker comme un élément architectural capable de recevoir des messages de plusieurs destinations, de déterminer la destination correcte et d'acheminer le message vers le bon canal.

Pour nos besoins avec Celery, nous considérerons un courtier un "transport de messages" où les tâches créées sont stockées. Les courtiers n'exécutent la tâche : c'est le travail d'un travailleur. Les courtiers sont plutôt l'endroit où les tâches planifiées sont stockées dans lorsqu'une tâche est planifiée, et extraites de lorsqu'un travailleur finit par exécuter une tâche. Un courtier est un composant requis pour que Celery fonctionne, et Celery se connectera à exactement un courtier.

La page Backends et courtiers de Celery en répertorie certains si ses courtiers sont pris en charge, et il existe d'autres courtiers expérimentaux qu'il prend en charge qui ne sont pas répertoriés (tels que SQLAlchemy). Ces courtiers (ou « transports de messages ») sont gérés par une bibliothèque Python maintenue par Celery pour les transports de messages appelée Kombu. Lorsque vous recherchez des informations sur la configuration des courtiers, il est parfois utile de consulter la documentation de Kombu plutôt que celle de Celery.

Certains courtiers disposent de fonctionnalités avancées telles que la distribution et la priorité des tâches, tandis que d'autres fonctionnent comme de simples files d'attente.

Ouvrier

Un worker est une instance de Celery qui extrait les tâches du courtier et exécute les fonctions de tâche définies dans votre application Python. Celery est capable d'exécuter du code Python dans ses ouvriers car Celery lui-même est écrit en Python.

De nombreux travailleurs peuvent s'exécuter simultanément pour exécuter des tâches. Lorsque vous exécutez la commande Celery Worker, elle lancera par défaut un travailleur pour chaque cœur de votre ordinateur. Si votre ordinateur possède 16 cœurs, l'exécution de Celery Worker démarrera 16 Workers.

Si aucun travailleur n'est en cours d'exécution, les messages (tâches) s'accumuleront dans le courtier jusqu'à ce que les travailleurs soient disponibles pour les exécuter.

Back-end

La page des tâches du guide de l'utilisateur de Celery contient les informations suivantes à propos des backends :

Si vous souhaitez garder une trace des tâches ou avoir besoin des valeurs de retour, Celery doit stocker ou envoyer les états quelque part afin qu'ils puissent être récupérés plus tard. Vous avez le choix entre plusieurs backends de résultats intégrés : SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc) et Redis – ou vous pouvez définir le vôtre.

TLDR : un backend suit les résultats et les résultats renvoyés des tâches asynchrones. Qu'est-ce que cela signifie réellement et quand cela pourrait-il être utile ?

Imaginez que vous créez une application de comptabilité dans Django capable de générer un rapport annuel. La génération du rapport peut prendre quelques minutes.

Pour offrir à vos utilisateurs une expérience plus réactive, vous utilisez une requête AJAX pour lancer une tâche de génération de rapport. Cette requête renvoie un ID de la tâche, qu'elle peut utiliser pour interroger le serveur toutes les quelques secondes pour voir si le rapport est généré. Une fois la tâche terminée, elle renverra l'ID du rapport, que le client pourra utiliser pour afficher un lien vers le rapport via JavaScript.

Nous pouvons implémenter cela avec Celery et Django en utilisant le code suivant :

from celery import shared_task
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from accounting.models import Asset
from accounting.reports import AnnualReportGenerator

@shared_task
def generate_report_task(year):
    # This could take minutes...
    report = AnnualReportGenerator().generate(year)
    asset = Asset.objects.create(
        name=f"{year} annual report",
        url=report.url,
    )
    return asset.id

@require_http_methods(["POST"])
def generate_annual_report_view(request):
    year = request.POST.get("year")
    task = generate_report_task.delay(year)
    return JsonResponse({"taskId": task.id})

def get_annual_report_generation_status_view(request, task_id):
    task = generate_report_task.AsyncResult(task_id)

    # The status is typically "PENDING", "SUCCESS", or "FAILURE"
    status = task.status
    return JsonResponse({"status": status, "assetId": task.result})

Dans cet exemple, l'ID d'actif renvoyé par generate_report_task() est stocké dans un backend. Le backend stocke le résultat et le résultat renvoyé. Le backend ne stocke pas le statut des tâches non encore traitées : celles-ci ne seront ajoutées qu'une fois le résultat obtenu. Une tâche qui renvoie « PENDING » a un statut totalement inconnu : une tâche associée peut même ne pas exister. Les tâches renvoient généralement « SUCCÈS » ou « ÉCHEC », mais vous pouvez voir tous les statuts dans la documentation sur l'état de Celery.

Avoir un backend n'est pas requis pour que Celery exécute des tâches. Cependant, il est requis si jamais vous avez besoin de vérifier le résultat d'une tâche ou de renvoyer le résultat d'une tâche. Si vous essayez de vérifier l'état d'une tâche alors que Celery n'a pas de backend configuré, une exception sera levée.

J'espère que cet article vous aidera à comprendre les différents morceaux de céleri et pourquoi vous pourriez envisager de l'utiliser. Bien que la documentation officielle soit difficile à comprendre, l'apprentissage approfondi de Celery peut débloquer de nouvelles possibilités au sein de vos applications Python.