Desbloqueando la concurrencia: Una guía completa de Celery en Python

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

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

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

1. 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.
2. 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.
3. Instala dependencias en cada entorno virtual (si usas):
4. Para Celery (en el directorio principal): pip install celery redis
5. Para Reflex (dentro de my_reflex_app): pip install reflex
6. Inicia el trabajador de Celery: Abre una terminal en el directorio principal (my_celery_project/) y ejecuta:
7. celery -A celery_app worker --loglevel=info
8. Inicia la aplicación Reflex.dev: Abre otra terminal, ve al directorio my_reflex_app/ y ejecuta:
9. reflex run
10. Reflex te dará una URL (generalmente http://localhost:3000).
11. 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.