Introducción a Celery

En el desarrollo de aplicaciones modernas, la eficiencia y la capacidad de respuesta son cruciales. Sin embargo, muchas operaciones, como el procesamiento de imágenes, el envío de correos electrónicos masivos, o la generación de informes complejos, pueden ser tareas que consumen mucho tiempo y que, si se ejecutan de forma síncrona, pueden bloquear la aplicación principal, deteriorando la experiencia del usuario. Aquí es donde entra en juego Celery.

Celery es un sistema de cola de tareas distribuido y flexible para Python. Permite a las aplicaciones delegar tareas de larga duración a un proceso separado, liberando la aplicación principal para responder a otras solicitudes de inmediato. En este artículo, exploraremos paso a paso cómo funciona Celery, cómo integrarlo en tus proyectos, las mejores prácticas para su uso, y un ejemplo práctico que combina Celery con Redis, Supabase y Reflex.dev para una solución robusta y escalable.

¿Cómo funciona Celery?

Celery se basa en una arquitectura de tres componentes principales:

Productor (o Cliente): Es la parte de tu aplicación que crea y envía tareas a la cola de tareas. Por ejemplo, en una aplicación web, un usuario puede iniciar una operación que desencadena una tarea asíncrona.
Broker (o Intermediario de Mensajes): Es el corazón de Celery. Actúa como un intermediario entre el productor y los trabajadores. Almacena las tareas en una cola y las entrega a los trabajadores a medida que están disponibles. Los brokers populares incluyen RabbitMQ, Redis y Amazon SQS.
Trabajador (Worker): Son procesos que escuchan el broker, recogen las tareas de la cola y las ejecutan. Pueden ejecutarse en la misma máquina que la aplicación o en máquinas separadas para escalar horizontalmente.

El flujo de trabajo es el siguiente:

El productor llama a una función decorada como tarea de Celery.
Celery serializa la tarea (incluyendo el nombre de la función, argumentos y opciones) y la envía al broker.
El broker almacena la tarea en una cola.
Un trabajador de Celery conectado al broker recoge la tarea de la cola.
El trabajador deserializa la tarea y ejecuta la función correspondiente.

Opcionalmente, el trabajador puede almacenar el resultado de la tarea en un backend de resultados.

Configurando Celery

Para comenzar a usar Celery, primero necesitas instalarlo junto con el cliente para el broker que elijas. En este ejemplo, usaremos Redis como broker y backend de resultados debido a su simplicidad y eficiencia.

1. Instalación

Abre tu terminal e instala Celery y el cliente de Redis:

$ pip install celery redis

2. Creación de una Instancia de Celery

Necesitas crear una instancia de aplicación Celery en tu proyecto. Este objeto es el punto de entrada para todas las funcionalidades de Celery. Crearemos un archivo celery_app.py:

# celery_app.py

from celery import Celery

# Configura Celery para usar Redis como broker y backend de resultados

# Es importante que el host de Redis sea accesible desde donde se ejecuten los workers.

CELERY_BROKER_URL = 'redis://localhost:6379/0'

CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

app = Celery('my_celery_app',

broker=CELERY_BROKER_URL,

backend=CELERY_RESULT_BACKEND,

include=['tasks']) # Incluye el módulo donde definirás tus tareas

# Configuración adicional, si es necesaria

app.conf.update(

task_track_started=True,

task_serializer='json',

accept_content=['json'],

result_serializer='json',

timezone='America/Bogota', # Ajusta a tu zona horaria

enable_utc=True,

)

if __name__ == '__main__':

app.start()

En este código:

Celery('my_celery_app', ...): Crea una instancia de Celery. El primer argumento es el nombre del módulo actual (usado para auto-descubrimiento de tareas).
broker: Especifica la URL de tu broker (Redis en este caso).
backend: Especifica dónde se almacenarán los resultados de las tareas.
include=['tasks']: Le dice a Celery dónde buscar las tareas. Asegúrate de que este módulo (tasks.py) exista y contenga tus tareas.

Definiendo Tareas Asíncronas

Las tareas en Celery son funciones Python normales que decoras con @app.task. Estas funciones pueden entonces ser llamadas de forma asíncrona.

Notas importantes:

@app.task: Este decorador convierte la función process_data y send_welcome_email en una tarea de Celery.
bind=True: Permite que la tarea acceda a su propia instancia (self), lo que es útil para reintentos.
default_retry_delay y max_retries: Configuran el comportamiento de reintento para la tarea.

Crea un archivo tasks.py:

# tasks.py

from celery_app import app

import time

import random

import os

# Simula la conexión con Supabase (en un entorno real, usarías la SDK de Supabase)

# Para este ejemplo, solo simularemos una operación de base de datos.

def supabase_insert_data(data):

"""Simula una inserción de datos en Supabase."""

print(f"Simulando inserción en Supabase: {data}")

time.sleep(random.uniform(0.5, 2.0)) # Simula latencia de red/BD

print("Inserción simulada completa.")

return True

@app.task

def process_data(user_id, data):

"""

Tarea de Celery que simula un procesamiento de datos complejo.

Luego intenta "guardar" el resultado en Supabase.

"""

print(f"[{os.getpid()}] Iniciando procesamiento de datos para el usuario {user_id}...")

time.sleep(random.uniform(3, 8)) # Simula una tarea que consume tiempo

processed_result = f"Datos procesados para {user_id}: {data.upper()} - {time.time()}"

# Simular interacción con Supabase

try:

if supabase_insert_data({'user_id': user_id, 'processed_data': processed_result}):

print(f"[{os.getpid()}] Datos de la tarea '{process_data.name}' procesados y 'guardados' para {user_id}.")

return processed_result

else:

print(f"[{os.getpid()}] Fallo al 'guardar' datos en Supabase para {user_id}.")

return None

except Exception as e:

print(f"[{os.getpid()}] Error en la interacción con Supabase: {e}")

raise # Vuelve a lanzar la excepción para que Celery la maneje

@app.task(bind=True, default_retry_delay=300, max_retries=5)

def send_welcome_email(self, recipient_email, username):

"""

Tarea de Celery que simula el envío de un correo electrónico.

Incluye un mecanismo de reintento en caso de fallo.

"""

print(f"[{os.getpid()}] Intentando enviar correo a {recipient_email}...")

try:

# Simula un fallo ocasional

if random.random() < 0.2: # 20% de probabilidad de fallo

raise Exception("¡Error de conexión con el servidor de correo!")

time.sleep(random.uniform(1, 3))

print(f"[{os.getpid()}] Correo de bienvenida enviado a {recipient_email} para {username}.")

return True

except Exception as exc:

print(f"[{os.getpid()}] Error al enviar correo a {recipient_email}: {exc}")

# Reintentar la tarea

try:

print(f"[{os.getpid()}] Reintentando enviar correo... Intento {self.request.retries + 1}")

self.retry(exc=exc)

except self.max_retries_reached:

print(f"[{os.getpid()}] Máximo de reintentos alcanzado para el correo a {recipient_email}. Tarea fallida.")

return False

except Exception as retry_exc:

print(f"[{os.getpid()}] Error al intentar reintentar la tarea: {retry_exc}")

return False

Ejecutando los Trabajadores de Celery

Una vez que tienes tu instancia de Celery y tus tareas definidas, necesitas iniciar uno o más trabajadores para procesar las tareas.

Primero, asegúrate de que tu servidor Redis esté en ejecución. Si lo tienes instalado localmente, usualmente puedes iniciarlo con redis-server.

Luego, desde el directorio de tu proyecto (donde están celery_app.py y tasks.py), ejecuta el siguiente comando en una nueva terminal:

$ celery -A celery_app worker --loglevel=info

celery -A celery_app: Le dice a Celery que use la aplicación Celery definida en celery_app.py.
worker: Indica que quieres iniciar un trabajador.
--loglevel=info: Establece el nivel de registro para ver información detallada sobre lo que hace el trabajador.

Verás la salida del trabajador, indicando que está listo para recibir tareas.

Monitoreando Celery con Flower (Opcional)

Flower es una herramienta de monitoreo basada en la web para Celery. Te permite ver el estado de las tareas, los trabajadores, estadísticas, etc.

Instalación:

Ejecución (en una nueva terminal, mientras Redis y los trabajadores están corriendo):

$ celery -A app.celery_app flower worker -l info

Luego, abre tu navegador y ve a http://localhost:5555.

celery

Con flower puedes monitorear el trabajo que hace Celery, observar los workers y tareas y de igual manera configurar la escalabilidad y capacidad de operar cada uno de los requerimientos.

Para mas información ingresa a: https://flower.readthedocs.io/en/latest/ y https://redis.io/docs/latest/

Ejemplo Práctico: Integrando Celery, Redis, Supabase y Reflex.dev

Ahora, construyamos un ejemplo completo donde una aplicación web (Reflex.dev) delega tareas a Celery, que a su vez simula interactuar con Supabase.

Asumiremos que celery_app.py y tasks.py son los mismos que los definidos anteriormente.

Explicación del código Reflex.dev:

Importaciones: Se importan las tareas process_data y send_welcome_email desde tasks.py. Es crucial ajustar el sys.path para que Python pueda encontrar tus módulos de Celery.
run_celery_task_async: Las llamadas a task.delay() de Celery son síncronas. Dado que Reflex se ejecuta en un bucle de eventos asíncrono, si llamas task.delay() directamente, bloquearías el hilo principal. Para evitar esto, utilizamos ThreadPoolExecutor para ejecutar la llamada a Celery en un hilo separado.
Estado (State): Almacena mensajes, el ID de la tarea, su estado y el resultado para mostrarlos en la interfaz de usuario.
start_data_processing y send_email: Estos métodos en el estado de Reflex son los que el usuario interactúa. Cuando se llaman, utilizan process_data.delay() o send_welcome_email.delay() para enviar la tarea al broker de Celery.
check_data_task_status y check_email_task_status: Estos métodos consultan el estado actual de la tarea usando app.AsyncResult(self.task_id). result.ready(), result.successful(), result.get(), y result.traceback son métodos útiles para obtener información sobre la tarea.
Interfaz de usuario (index): Una interfaz simple con campos de entrada y botones para iniciar las tareas, y texto para mostrar el estado y los resultados.

Simulando Supabase

En este ejemplo, la interacción con Supabase se simula directamente dentro de la función supabase_insert_data en tasks.py. En un entorno real, instalarías el SDK de Supabase (pip install supabase) y configurarías tu cliente Supabase con tu URL y clave de API. Las tareas de Celery serían el lugar ideal para realizar inserciones o actualizaciones de datos en Supabase, ya que no bloquearían la interfaz de usuario.

Redis como Broker y Backend de Resultados

Hemos configurado Celery para usar Redis (redis://localhost:6379/0) tanto para el broker como para el backend de resultados. Esto significa que:

Cuando Reflex llama a task.delay(), el mensaje de la tarea se envía a una cola en Redis.
El trabajador de Celery escucha esa cola en Redis, recoge la tarea y la ejecuta.
Una vez que la tarea termina, su resultado (o el rastreo de error) se almacena también en Redis, lo que permite a la aplicación Reflex consultar el estado y el resultado de la tarea más tarde.

Pasos para Ejecutar el Ejemplo

Instala Redis: Asegúrate de tener un servidor Redis en ejecución en localhost:6379. Puedes descargar Redis desde https://redis.io/download/ o instalarlo a través de tu gestor de paquetes. Inícialo con redis-server.
Crea los archivos: Asegúrate de tener los archivos celery_app.py y tasks.py en el mismo directorio principal, y la carpeta my_reflex_app con my_reflex_app.py dentro.
Instala dependencias en cada entorno virtual (si usas):
Para Celery (en el directorio principal): pip install celery redis
Para Reflex (dentro de my_reflex_app): pip install reflex
Inicia el trabajador de Celery: Abre una terminal en el directorio principal (my_celery_project/) y ejecuta:
celery -A celery_app worker --loglevel=info
Inicia la aplicación Reflex.dev: Abre otra terminal, ve al directorio my_reflex_app/ y ejecuta:
reflex run
Reflex te dará una URL (generalmente http://localhost:3000).
Interactúa con la aplicación: Abre tu navegador, ve a la URL de Reflex. Introduce datos y haz clic en los botones. Observa cómo las tareas se inician inmediatamente en Reflex, mientras el trabajo pesado se realiza en la terminal del trabajador de Celery. Puedes hacer clic en "Actualizar Estado" para ver el progreso de la tarea.

Mejores Prácticas con Celery

Manejo de Errores y Reintentos: Implementa bloques try-except dentro de tus tareas. Celery permite reintentar tareas automáticamente (@app.task(bind=True, default_retry_delay=..., max_retries=...)) lo cual es crucial para operaciones propensas a fallos temporales (ej. problemas de red).
Idempotencia: Diseña tus tareas para que sean idempotentes, es decir, que ejecutar la misma tarea varias veces con los mismos argumentos produzca el mismo resultado sin efectos secundarios no deseados. Esto es importante para reintentos o fallos del trabajador.
Serialización: Celery utiliza la serialización para enviar tareas y resultados. El formato json es el predeterminado y el más recomendado por su interoperabilidad. Evita pasar objetos complejos o grandes directamente; en su lugar, pasa IDs o URLs y permite que la tarea recupere los datos por sí misma.
Límites de Tareas (Rate Limiting): Puedes limitar la cantidad de veces que una tarea puede ejecutarse en un período de tiempo (@app.task(rate_limit='10/m')). Esto es útil para interactuar con APIs externas que tienen límites de solicitudes.
Logging: Configura el logging adecuado en tus tareas para facilitar la depuración y el monitoreo.
Colas Personalizadas: Si tienes diferentes tipos de tareas (ej. tareas de alta prioridad vs. baja prioridad), puedes definirlas en colas separadas y configurar trabajadores para que escuchen colas específicas.
Graceful Shutdown: Asegúrate de que tus trabajadores de Celery puedan terminar de procesar las tareas actuales antes de cerrarse por completo (celery multi stop).
Monitoreo: Usa herramientas como Flower para tener una visión clara del estado de tus colas, tareas y trabajadores.

Conclusión

Celery es una herramienta invaluable en el ecosistema de Python para construir aplicaciones concurrentes y escalables. Al delegar tareas de larga duración a procesos en segundo plano, puedes mejorar drásticamente la capacidad de respuesta de tu aplicación, la experiencia del usuario y la robustez general del sistema. Con una comprensión clara de sus componentes, configuración y mejores prácticas, estarás bien equipado para desbloquear el verdadero potencial de la concurrencia en tus proyectos Python.

Blogs

Publicaciones recientes

¿Listo para llevar a su compañía al siguiente nivel?

Mira todos los servicios que ofrecemos