Objetivos

Al finalizar el tutorial el estudiante estará en capacidad de:

• Instalar Celery, Redis
• Trabajar con ambientes virtuales en Python 3.8 o superior.
• Crear y gestionar tareas asíncronas con Celery y Redis

Requisitos para desarrollar el tutorial

En particular se utilizarán los siguientes recursos:

1. Máquina (virtual) con sistema operativo Ubuntu 20.04
2. Permisos de administrador para instalar paquetes
3. Conocimientos en comandos de consola

El módulo venv

Al desarrollar y especialmente al probar cosas nuevas, se cometen errores y se rompen cosas. Aislar el ambiente de desarrollo facilita instalar paquetes, probar librerías y ejecutar tareas de modo que cuando inevitablemente se rompa algo, el daño quede aislado. Esta es una breve explicación de los entornos virtuales que utilizaremos para desarrollar aplicaciones Python y cómo configurarlo.

Los entornos virtuales facilitan mantener las dependencias de un proyecto aisladas y bien administradas cuando desarrolla aplicaciones en Python. Hay muchas herramientas diferentes para hacer esto. Los entornos virtuales crean una carpeta para la instalación y las dependencias de Python y modifican la variable de entorno PATH para apuntar a la instalación deseada.

Utilizaremos Python 3.8 o superior para nuestros proyectos, utilizaremos el paquete venv para administrar entornos virtuales. Está incluido en la biblioteca estándar de Python desde la versión 3.3, esto significa que no tiene que instalar ninguna herramienta adicional en el caso de sistemas Microsoft Windows, sin embargo, para el caso de Ubuntu/Debian es necesario instalar el paquete.

Si no ha instalado Python 3.x lo puede hacer ejecutando:

$ sudo apt-get install python3

Para instalar el gestor de ambientes virtuales de Python lo puede hacer ejecutando:

$ sudo apt-get install python3-venv

Para administrar un nuevo proyecto usando Python 3.x, debe crear un nuevo ambiente virtual y una nueva carpeta ejecutando:

$ python3 -m venv lab-celery

Una instalación de Python 3.x es creada de inmediato en la carpeta ambiente. Para activar en nuevo entorno virtual, utilice el comando source:

$ source lab-celery/bin/activate

Para cerrar el ambiente de ejecución, ejecute el comando deactivate:

$ deactivate

Instalación

Celery es una biblioteca de Python para gestionar colas de tareas distribuidas, es decir, nos permite trabajar con diferentes tareas y distribuirlas en procesos. Ventajas de Celery:

• Escalabilidad
• Eficiencia
• Orden de transacciones garantizado
• Tolerante a fallos
• Soporta picos de mensajes

¿Para qué sirve Celery? Para dividir el backend de una aplicación en receptor de tareas y el procesador de tareas, administrando de forma uniforme las respuestas del backend.

Celery requiere una solución para enviar y recibir mensajes; por lo general, esto se presenta en forma de un servicio separado llamado de un message brokers. Los message brokers son servicios de mensajería que actúan como intermediarios; los brokers enrutan, agregan y permiten crear servicios de publicación/subscripción, comúnmente llamados servicios pubsub. Por último, mencionar que Celery soporta de manera estable varios servidores de tareas.

Hay varias opciones disponibles, que incluyen:

• RabbitMQ
• Redis
• Amazon SQS

Los message brokers son fundamentales en las soluciones cloud actuales, necesarias en arquitecturas basadas en eventos y en arquitecturas para streaming. Para esta práctica utilizaremos Redis.

Celery está disponible Python Package Index (PyPI), y para instalarlo basta con utilizar pip:

$ pip3 install celery

Celery + Redis

El primer paso es desarrollar una instancia de Celery (a esto lo llamamos la aplicación). La instancia o aplicación se utiliza como punto de entrada para todo lo que desea hacer en Celery, cómo crear tareas y administrar workers.

Vamos a desarrollar un primer ejemplo; para esto, cree un nuevo archivo llamado tasks.py con el siguiente código:

from datetime import datetime
from celery import Celery

app = Celery( 'tasks' , broker = 'redis://localhost:6379/0' ) 

# Creamos una tarea llamada sumar_numeros usando el decorador @app.task
# Se imprime un mensaje con la fecha simulando un LOG
@app.task
def sumar_numeros(x, y):
    print ("-> Se generó una tarea [{}]: {} + {}".format(datetime.now(), x, y))
    return x + y

# Creamos una tarea llamada hola
@app.task
def hola(nombre):
        return 'Hola %s' % nombre

El primer argumento de la función Celery es el nombre del módulo actual (el archivo tasks omitiendo la extensión). Esto solo es necesario para que los nombres se puedan generar automáticamente cuando las tareas se definan en el módulo __main__.

El segundo argumento es para conectarse al broker de mensajes, que especifica la URL del broker de mensajes que desea utilizar. Para este ejemplo, estamos especificando Redis como broker. Para instalar Redis en su máquina Linux ejecute los siguientes comandos:

$ sudo apt-get install redis-server

$ sudo systemctl enable redis-server.service

El comando anterior, habilita Redis como servicio de arranque en una máquina GNU/Linux para que esté activo cada vez que se reinicie la máquina. Para ejecutar redis sin reiniciar la máquina ejecute el comando:

$ redis-server

Para validar que el servidor de redis está en ejecución, ejecute el siguiente comando (cierre presionando la tecla Q):

$ sudo systemctl status redis

Para utilizar Redis con Python es necesario instalar la librería correspondiente, para esto utilice pip:

$ pip3 install redis

Como se puede observar en el código (tasks.py); para crear una tarea solo hay que añadir el decorador @app.task a una función de Python.

Guarde el archivo tasks.py y ahora vamos a asignarle a Redis las tareas que acabamos de crear. Para ello, debe ejecutar el siguiente comando en una terminal:

$ celery -A tasks worker -l info

Donde:

• La opción -A tasks especifica el archivo donde se encuentra la aplicación de Celery.
• La opción worker le indica a Celery que queremos crear un worker con las tareas del módulo
• Y la opción -l info permite especificar el nivel de los mensajes de información que se presentan en pantalla.

Abra una nueva terminal. El siguiente paso es utilizar e invocar las tareas creadas en el paso anterior desde nuestro programa principal. Para esto vamos a crear un archivo llamado app.py:

from tasks import hola, sumar_numeros

# Para ejecutar una función en forma de tarea lo tenemos que hacer de esta forma:
hola.delay('Mundo!')

# Para ejecutar la función sumar:
sumar_numeros.delay(3, 2)

Guarde el archivo, abra una nueva terminal sin cerrar la anterior (dónde está ejecutando el worker Celery) y ejecute el programa utilizando la instrucción:

$ python app.py

En este punto del ejemplo deberíamos tener 2 terminales abiertos:

• El terminal con el worker.

En este terminal va a poder visualizar las tareas que va ejecutando el Worker.

• El terminal donde estamos ejecutando el archivo app.py

En este terminal va a poder visualizar las salidas de la aplicación que genera tareas (en este caso no se generan salidas en este terminal).

Compresión de archivos

Finalice el proceso de Celery en la primera terminal (Ctrl + C). Modifique el archivo tasks.py, importando la librería zipFile:

import zipfile

Ahora, agregue la siguiente función al finalizar el archivo. La función compress utiliza zipFile para comprimir a formato zip y lo ubica en otro directorio. Consulte la referencia de la librería zipFile si desea profundizar en el tema.

@app.task
def comprimir(filename, zipname, new_path):
    print ('\n-> Se va a comprimir el archivo: {}'.format(filename))
    zfile = zipfile.ZipFile(new_path + '/' + zipname, 'w')
    zfile.write(filename, compress_type = zipfile.ZIP_DEFLATED)
    zfile.close()
    print ('\n-> El archivo comprimido se copió a : {}'.format(new_path))

Guarde el archivo tasks.py y ahora vamos a pasar el nuevo archivo de tareas que acabamos de modificar a Redis. Para ello, debe ejecutar el siguiente comando en una terminal.

$ celery -A tasks worker -l info

En la segunda terminal, vamos a modificar el archivo app.py, para que invoque la nueva tarea. Agregue la siguiente instrucción y guarde el archivo.

Al inicio del archivo recuerde importar la función comprimir:

from tasks import hola, sumar_numeros, comprimir

Al final del archivo invoque la tarea comprimir:

comprimir.delay('sin_comprimir/un.pdf', 'unPDF.zip', 'comprimidos')

Descargue en su consola cualquier archivo, para esto puede usar el comando wget, ejemplo:

$ wget https://d1.awsstatic.com/whitepapers/architecture/AWS_Well-Architected_Framework.pdf

Para su comodidad puede modificar el nombre del archivo para llamarlo un.pdf y cree dos nuevos directos llamados: sin_comprimir y comprimidos. Finalmente, ubique el archivo un.pdf en el directorio sin_comprimir.

Ejecute el programa utilizando la instrucción:

$ python app.py

En este punto del ejemplo deberíamos tener 2 terminales abiertos:

• El terminal con el worker.

• El terminal donde estamos ejecutando el archivo app.py

Si utiliza un explorador de archivos o comandos de consola para navegar entre los directorios podrá validar que el archivo fue comprimido en ZIP.

El código completo

El código de la aplicación lo encuentra en el siguiente repositorio de GitHub:

https://github.com/jpadillaa/taller-celery

[1] https://docs.celeryproject.org/en/stable/

[2] https://docs.celeryproject.org/en/stable/getting-started/introduction.html

[3] https://docs.celeryproject.org/en/stable/userguide/tasks.html