Sencillo sistema de cola de tareas en Python

Una cola de tareas sirve para ejecutar tareas en segundo plano, de forma asíncrona, sin bloquear la ejecución del programa principal. Es muy útil para tareas que requieren un tiempo prolongado para completarse. El clásico ejemplo es el envío de correos electrónicos. En lugar de enviar el correo cuando el usuario lo requiere, se encola la tarea y se envía en segundo plano, liberando al usuario para que pueda seguir interactuando con la aplicación. Otro ejemplo sería el procesamiento de imágenes, el usuario sube una imagen con mucho peso y se encola la tarea de crear una miniatura para su imagen de perfil.

En este artículo vamos a aprender cómo implementar un sencillo sistema de cola de tareas en Python utilizando la librería Huey. Una alternativa sencilla al popular Celery. Entre sus particularidades destaca:

• Es ligero.
• Fácil de usar.
• Tareas periódicas.
• Multiproceso, multihilo o gevent.
• Priorizar tareas.
• Bloquear para que no haya más de una tarea similar en ejecución.
• Integración con Django.

Y mucho más.

Necesitarás, antes de continuar, tener instalado y ejecutando un servidor de Redis. Puedes trabajar con configuraciones alternativas en memoria, en disco, en SQLite o modo immediate, pero no cubriré estos casos. Puedes consultar la documentación oficial realizar los cambios adecuados o usar Docker, ya que al final del artículo te mostraré cómo hacerlo.

Empezaremos instalando las dependencias necesarias:

Crea un archivo llamado requirements.txt con el siguiente contenido:

huey===2.5.2
redis===5.1.1

Revisa las últimas versiones de las librerías en pypi.org. Puede que haya una versión más reciente.

E instalamos:

pip install -r requirements.txt

Ahora prepararemos el primer archivo Python con el nombre de config.py:

import os
from huey import RedisHuey

huey = RedisHuey('mi-app', url=os.environ.get('STORE_URI', 'redis://redis:6379/0'))

Estamos creando una instancia de RedisHuey con el nombre mi-app. La URL de la base de datos de Redis se obtiene de la variable de entorno STORE_URI que la configuraremos más adelante o por defecto buscará en redis://redis:6379/0 (una clásica configuración con Docker). Si estas trabajando con una instalación sin contenedores, puedes apuntar a localhost, quedando como redis://localhost:6379/0.

Es el momento de crear el archivo tasks.py con las tareas que queremos encolar. Por ejemplo, una tarea que suma dos números:

from config import huey
import time


@huey.task()
def add_numbers(a, b):
    time.sleep(5)
    return a + b

Como puedes ver en el código, la tarea add_numbers recibe dos argumentos a y b, se detiene durante 5 segundos y devuelve la suma de ambos. El decorador @huey.task() es el que indica que la función es una tarea.

A continuación, crearemos un archivo llamado app.py:

from config import huey
from tasks import add_numbers

El archivo lo usaremos para importar la instancia de Huey y todas las tareas. No tiene ningún otro propósito. Es una buena práctica ya que evitará importaciones circulares.

Ya podemos ejecutar Huey.

huey_consumer.py app.huey -w 4 -k process -f

¿Donde esta huey_consumer.py? Huey viene con un script que se instala en el sistema, no es necesario crearlo.

No deseo entrar en detalles ya que la documentación oficial explica cada argumento en detalle y sus opciones. En resumen te diré que:

• app.huey es el módulo que contiene la instancia de Huey. Entra en el archivo app.py y busca la variable huey.
• -w 4 indica que se ejecutarán 4 procesos en paralelo.
• -k process indica que se utilizará el método de arranque process. Es apropiado cuando son trabajos de CPU y no de E/S (entrada/salida).
• -f aborta tareas pendientes. Aunque detengas el proceso, las tareas pendientes se ejecutarán cuando vuelvas a levantarlo. Si no quieres que esto ocurra, añade la bandera.

¡Vamos a encolar una tarea!

Entramos en el intérprete de Python:

python

Importamos la tarea:

from tasks import add_numbers

Y la ejecutamos:

resultado = add_numbers(2, 3)

Pasado 5 segundos, ejecutamos:

resultado()

Y deberíamos obtener 5. En caso de llamarlo antes de tiempo, obtendremos un None.

¡Lo hemos logrado!

Opcional: Docker

Primero definimos el archivo Dockerfile que contiene la imagen de Python y las dependencias necesarias:

FROM python:3.12-slim

# Sets python path to be able to import modules
ENV PYTHONPATH=":/usr/src/app"

# install dependencies
RUN pip3 install --no-cache-dir --upgrade pip

COPY ./requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

# set work directory
WORKDIR /usr/src/app

Ahora es el momento de crear compose.yaml con un contenido similar al siguiente:

services:

  huey:
    build: .
    restart: "no"
    entrypoint: huey_consumer.py app.huey -w 4 -k process -f
    volumes:
      - .:/usr/src/app
    environment:
      - STORE_URI=redis://redis:6379/0
    depends_on:
      - redis

  redis:
    image: redis:alpine
    restart: "no"
    expose:
      - 6379

Y finalmente, ejecutamos:

docker-compose up

Ya podemos paralelizar tareas en Python. ¡A disfrutarlo!