Aprender a apio puede resultar complicado. Si bien su documentación es completa, tiende a omitir lo básico.

Esta publicación definirá cuatro de los conceptos principales de Celery, discutirá la relación entre Celery y Kombu y utilizará algunos ejemplos de código para ilustrar cómo Celery podría ser útil en aplicaciones reales. Los ejemplos utilizarán el marco web Django y su decorador @shared_task, pero los conceptos también son aplicables a Flask, FastAPI y otros.

Tareas, corredores, trabajadores y backends

Será difícil encontrar un lugar en la documentación actual de Celery que explique claramente lo que se considera un broker o backend, pero con suficiente investigación podrás encontrar y inferir definiciones.

A continuación se detallan conceptos que debes conocer antes de comenzar con el apio.

Tarea

Una tarea es un trabajo que Celery realizará asincrónicamente (en este contexto, esa es una palabra elegante para "no inmediatamente"). En una aplicación web, una tarea podría ser enviar un correo electrónico después de que un usuario envía un formulario. Enviar un correo electrónico puede ser una operación de varios segundos y obligar a un usuario a esperar a que se envíe un correo electrónico antes de redirigir puede hacer que una aplicación parezca lenta.

Las tareas se definen utilizando decoradores en Celery. A continuación utilizamos el decorador @shared_task para convertir send_thank_you_email() en una tarea de Apio que se puede utilizar en el controlador de envío de formularios 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})

Cuando una tarea se define usando un decorador en Celery, agrega un método delay() a la tarea. Puede ver la tarea send_thank_you_email llamando al método delay() en el ejemplo anterior después de que el formulario se haya guardado correctamente. Cuando se llama a delay(), enviará la tarea send_thank_you_email y sus datos al broker donde se almacena y luego será ejecutada por un trabajador, momento en el cual el usuario ser enviado por correo electrónico.

El beneficio de enviar el trabajo a Celery se vuelve más obvio si necesita enviar correos electrónicos adicionales después de guardar el formulario. Por ejemplo, es posible que desee enviar un correo electrónico al equipo de atención al cliente para informarle que ha recibido nuevos comentarios. Con el apio, esto casi no agrega tiempo adicional a la respuesta.

Las tareas de apio también permiten una configuración avanzada adicional. En caso de que no se pueda enviar un correo electrónico, puede codificar su tarea para reintentar y configurar automáticamente ajustes como max_retries, retry_backoff, retry_jitter, etc.

Corredor

El glosario de propuestas de mejora del apio tiene lo siguiente que decir sobre Message Brokers:

Enterprise Integration Patterns define un Message Broker como un bloque arquitectónico que puede recibir mensajes de múltiples destinos, determinar el destino correcto y enrutar el mensaje al canal correcto.

Para nuestros propósitos con Celery, consideraremos un broker un "transporte de mensajes" donde se almacenan las tareas creadas. Los corredores en realidad no ejecutan la tarea: ese es el trabajo de un trabajador. En cambio, los intermediarios son el lugar donde las tareas programadas se almacenan en cuando se programa una tarea, y se extraen de cuando un trabajador finalmente ejecuta una tarea. Un corredor es un componente requerido para que Celery funcione, y Celery se conectará exactamente con un corredor.

La página Backends y Brokers de Apio enumera algunos de los brokers compatibles, y hay otros brokers experimentales que admite y que no están en la lista (como SQLAlchemy). Estos intermediarios (o "transportes de mensajes") son administrados por una biblioteca Python mantenida por Celery para transportes de mensajes llamada Kombu. Al buscar información sobre la configuración de brokers, a veces resulta útil consultar la documentación de Kombu en lugar de la de Celery.

Algunos corredores tienen funciones avanzadas como distribución de tareas y prioridad, mientras que otros operan como colas simples.

Obrero

Un trabajador es una instancia de Celery que extrae tareas del intermediario y ejecuta las funciones de tareas definidas en su aplicación Python. Celery puede ejecutar código Python en sus trabajadores porque el propio Celery está escrito en Python.

Muchos trabajadores pueden correr simultáneamente para ejecutar tareas. Cuando ejecuta el comando de trabajador de apio, activará un trabajador para cada núcleo de su computadora de forma predeterminada. Si su computadora tiene 16 núcleos, ejecutar apio trabajador iniciará 16 trabajadores.

Si no hay trabajadores en ejecución, los mensajes (tareas) se acumularán en el intermediario hasta que los trabajadores estén disponibles para ejecutarlos.

backend

La página de tareas en la Guía del usuario de Apio tiene lo siguiente que decir sobre backends:

Si desea realizar un seguimiento de las tareas o necesita los valores devueltos, Celery debe almacenar o enviar los estados a algún lugar para poder recuperarlos más tarde. Hay varios backends de resultados integrados para elegir: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc) y Redis, o puedes definir el tuyo propio.

TLDR: un backend rastrea los resultados y los resultados devueltos de tareas asincrónicas. ¿Qué significa eso realmente y cuándo podría resultar útil?

Imagina que estás creando una aplicación de contabilidad en Django que puede generar un informe anual. El informe podría tardar unos minutos en generarse.

Para brindarles a sus usuarios una experiencia más receptiva, utiliza una solicitud AJAX para iniciar una tarea de generación de informes. Esa solicitud devuelve una identificación de la tarea, que puede usar para sondear el servidor cada pocos segundos para ver si se genera el informe. Una vez completada la tarea, devolverá el ID del informe, que el cliente puede usar para mostrar un enlace al informe a través de JavaScript.

Podemos implementar esto con Celery y Django usando el siguiente código:

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})

En este ejemplo, el ID del activo devuelto por generate_report_task() se almacena en un backend. El backend almacena el resultado y el resultado devuelto. El backend no almacena el estado de las tareas aún por procesar: estas solo se agregarán una vez que haya habido un resultado. Una tarea que devuelve "PENDIENTE" tiene un estado completamente desconocido: es posible que una tarea asociada ni siquiera exista. Las tareas normalmente devolverán "ÉXITO" o "FALLO", pero puedes ver todos los estados en los documentos de estado de Apio.

No es necesario tener un backend para que Celery ejecute tareas. Sin embargo, es necesario si alguna vez necesita verificar el resultado de una tarea o devolver el resultado de una tarea. Si intenta verificar el estado de una tarea cuando Celery no tiene un backend configurado, se generará una excepción.

Espero que esta publicación te ayude a comprender las piezas individuales de apio y por qué podrías considerar usarlo. Si bien la documentación oficial es difícil de asimilar, aprender profundamente sobre Celery puede desbloquear nuevas posibilidades dentro de sus aplicaciones Python.