Celery worker - Tareas en segundo plano

Un Celery worker es un componente clave de Celery, una biblioteca de Python que sirve para gestionar y ejecutar tareas en segundo plano (background tasks). Los workers son los procesos encargados de recibir las tareas enviadas a la cola (queue) por la aplicación principal y ejecutarlas de forma asíncrona. Esto es especialmente útil para tareas que pueden ser lentas o intensivas en recursos, como enviar correos electrónicos, procesar imágenes o realizar cálculos complejos.

¿Para qué sirve un Celery worker?

1. Ejecución de tareas asíncronas: Permite que tu aplicación principal delegue tareas sin bloquear el flujo principal.
2. Procesamiento distribuido: Puedes tener múltiples workers ejecutando tareas en paralelo, lo cual mejora la eficiencia y la escalabilidad.
3. Gestión de tareas programadas: Puede ejecutar tareas repetitivas o programadas (usando herramientas adicionales como celery beat).

Estructura básica de un Celery worker

Para configurar y usar un Celery worker, necesitas al menos los siguientes componentes:

1. Archivo de configuración de Celery (celery.py) Define la aplicación Celery y su configuración: 

 from celery import Celery

celery = Celery(
    'nombre_del_proyecto',
    broker='redis://localhost:6379/0',  # URL del broker de mensajes
    backend='redis://localhost:6379/0' # Almacén de resultados
)

celery.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True
)
 

2. Tareas definidas (tasks.py) Son las funciones que quieres ejecutar de forma asíncrona:

 from .celery import celery

@celery.task
def enviar_correo(email, asunto):
    print(f"Enviando correo a {email} con asunto: {asunto}")
 

3. Iniciar el worker Una vez configuradas las tareas y el broker (como Redis o RabbitMQ), puedes iniciar el worker con el siguiente comando en la terminal:

 celery -A nombre_del_proyecto.celery worker --loglevel=info
 

• -A nombre_del_proyecto.celery: Especifica la aplicación Celery.
• worker: Inicia el worker para procesar tareas.
• --loglevel=info: Define el nivel de detalle de los logs.
4.  Enviar una tarea Desde cualquier parte de tu código, puedes enviar una tarea al worker: 

 from nombre_del_proyecto.tasks import enviar_correo

enviar_correo.delay('usuario@example.com', 'Bienvenido a Celery')
 

Flujo de trabajo del Celery worker

1. La aplicación principal envía una tarea al broker (como Redis).
2. El Celery worker recupera la tarea del broker.
3. El worker ejecuta la tarea de forma asíncrona.
4. Si se configuró un backend, los resultados se almacenan para ser consultados más adelante.

Resumen

Un Celery worker es esencial para manejar tareas en segundo plano de manera eficiente, separando la lógica de ejecución intensiva de la aplicación principal. Con una estructura modular, Celery permite escalar aplicaciones y procesar tareas distribuidas en múltiples workers para mejorar el rendimiento general del sistema.

 

---

 

Código  para configurar Celery en el proyecto metabuscador

 

metabuscador/
├── src/
│   └── metabuscador/
│       ├── __init__.py
│       ├── settings.py
│       ├── tasks/
│       │   ├── __init__.py
│       │   ├── celery_tasks/
│       │   │   ├── __init__.py
│       │   │   ├── celery.py
│       │   │   ├── tasks.py
│       │   │   └── utils.py
│       ├── apps/
│       │   ├── __init__.py
│       │   ├── buscador/
│       │   │   ├── __init__.py
│       │   │   ├── models.py
│       │   │   ├── views.py
│       │   │   ├── tests.py
│       │   │   ├── urls.py
│       │   │   └── templates/
│       │   │       ├── base.html
│       │   │       └── buscador.html
├── static/
│   ├── css/
│   ├── js/
│   └── img/
├── templates/
│   └── base.html
├── tests/
├── docs/
├── .venv/
├── logs/
│   ├── celery.log
│   └── app.log
├── manage.py
├── requirements.txt
├── README.md
├── setup.py
└── .gitignore
  

 

¿Cómo funciona el envío de tareas?

En Celery, definimos funciones como tareas en un archivo específico (por ejemplo, tasks.py). Luego, puedes "enviar" o "invocar" estas tareas desde cualquier lugar de tu proyecto, lo que significa que las colocas en la cola (queue) para que los workers las procesen.

¿Qué nombre debe tener el archivo?

El archivo donde defines las tareas generalmente se llama tasks.py, pero puedes usar cualquier nombre si lo organizas bien en tu estructura de proyecto. Por ejemplo, en este caso, tienes este archivo estructurado como:

 metabuscador/
├── src/
│   └── metabuscador/
│       ├── tasks/
│       │   ├── celery_tasks/
│       │   │   ├── tasks.py  # Archivo donde defines las tareas
 

Así que puedes usar tasks.py para definir las tareas y hacer referencia a ellas en tu código, como sigue.

Ejemplo en el archivo tasks.py

Define tus tareas en src/metabuscador/tasks/celery_tasks/tasks.py:

 from .celery import celery

@celery.task
def enviar_correo(email, asunto):
    return f"Correo enviado a {email} con el asunto: {asunto}"

@celery.task
def suma(a, b):
    return a + b
 

Enviar tareas desde otro archivo del proyecto

Puedes llamar o "enviar" estas tareas desde cualquier otro archivo de tu proyecto. Aquí tienes un ejemplo:

from src.metabuscador.tasks.celery_tasks.tasks import enviar_correo, suma

# Enviar la tarea al worker
resultado_suma = suma.delay(3, 5)
resultado_correo = enviar_correo.delay("usuario@example.com", "Bienvenido")

# Opcional: Obtener el resultado si lo necesitas
print(f"Resultado de la suma: {resultado_suma.get()}")
print(f"Resultado del correo: {resultado_correo.get()}")
  

¿Qué hace delay()?

El método .delay() envía la tarea a la cola de Celery para que el worker la procese en segundo plano. Esto asegura que la tarea no se ejecute en la misma línea de código y no bloquee otras operaciones en tu aplicación.

 

---

Invocar tareas Celery

El archivo desde donde envías tareas (o invocas tareas Celery) puede estar ubicado en cualquier parte de tu proyecto, siempre y cuando la ruta al archivo que contiene las tareas definidas (tasks.py) esté correctamente referenciada. Sin embargo, la ubicación depende de la lógica y estructura de tu proyecto. 

 

Considerando la estructura del proyecto metabuscador, aquí hay sugerencias sobre dónde podría estar ubicado el archivo desde el cual enviarías las tareas:

1. Dentro de tu aplicación principal (Ejemplo: buscador/views.py)

Este es el enfoque más común si estás enviando tareas desde una vista de tu aplicación (por ejemplo, en Django):

src/metabuscador/apps/buscador/views.py 

Contenido del archivo:

 from src.metabuscador.tasks.celery_tasks.tasks import enviar_correo

def procesar_solicitud():
    # Enviar la tarea al worker
    resultado = enviar_correo.delay("usuario@example.com", "Bienvenido")
    print(f"Tarea enviada, ID: {resultado.id}")
 

2. Desde un script separado (Ejemplo: scripts/enviar_tareas.py)

Si prefieres enviar tareas de manera puntual o probarlas, puedes crear un archivo en una carpeta llamada scripts:

metabuscador/scripts/enviar_tareas.py 

Contenido del archivo:

 from src.metabuscador.tasks.celery_tasks.tasks import enviar_correo, suma

# Ejemplo de envío de tareas
if __name__ == "__main__":
    resultado_suma = suma.delay(2, 3)
    print(f"Resultado de suma enviado, ID: {resultado_suma.id}")

    resultado_correo = enviar_correo.delay("usuario@example.com", "Bienvenido")
    print(f"Correo enviado, ID: {resultado_correo.id}")
 

Para ejecutarlo desde la terminal:

python metabuscador/scripts/enviar_tareas.py 

3. Dentro de tu lógica de negocio (Ejemplo: buscador/models.py o cualquier otro archivo)

Si quieres invocar tareas directamente desde los modelos o lógica interna, puedes hacerlo:

src/metabuscador/apps/buscador/models.py 

Contenido del archivo:

 from src.metabuscador.tasks.celery_tasks.tasks import enviar_correo

class Usuario:
    def registrar(self, email):
        # Lógica de registro
        enviar_correo.delay(email, "¡Gracias por registrarte!")
 

4. Para pruebas o automatizaciones (Ejemplo: en tests.py)

También puedes usar las tareas Celery en pruebas o automatizaciones:

src/metabuscador/apps/buscador/tests.py 

Contenido del archivo:

 from src.metabuscador.tasks.celery_tasks.tasks import suma

def test_suma():
    resultado = suma.delay(10, 20)
    assert resultado.get() == 30
 

Resumen:

El archivo que envía las tareas puede estar en cualquier parte del proyecto (como en views.py, models.py, o un script separado como scripts/enviar_tareas.py). Lo único importante es:

1. Importar correctamente la tarea desde su ubicación (tasks.py).
2. Tener tu entorno configurado adecuadamente, para que Celery y los workers puedan procesar las tareas.

Si quieres una recomendación, te sugiero empezar por views.py si estás trabajando con lógica web (como Django), o un script separado para pruebas.