Django: tareas en background con celery y rabbitmq

Venga, con este post vamos a subir un poco el nivel de nuestras aplicaciones!

Una de las problemáticas que te encuentras al programar con django es que el envío de mails es horrorosamente lento, el proceso de enviar el mail es relativamente rápido, pero no termino de entender el porqué en general es lento (la conexión y sobre todo la desconexión). Para ello una de las soluciones que más o menos ya había aplicado a mi manera con command hacía que la ejecución cada minuto se solapase con la ejecución de minutos anteriores. Hice una ñapa hace unos días hasta encontrar una solución un poco más elegante. Aquí os la traigo! :D

Hay varios posts que hablan de celery, incluso el libro que uso de consulta “Django 3 by Example”, hablan de usar @tank, pero esta opción está descontinuada para la versión 5, así que aquí veremos como hacerlo con Celery 5. Toda la info la he sacado de éste post.

Configurar e instalar celery y rabbitmq en nuestro proyecto/sistema
Lo primero será instalar celery en nuestro proyecto

laura@melatonina:~/dev/whistleblowerbox$ source venv/bin/activate
(venv) laura@melatonina:~/dev/whistleblowerbox$ pip install celery

A continuación instalamos rabbitmq en nuestro sistema. Rabbitmq es el servicio que encolará las tareas que le pasemos y luego las ejecutará en background

root@melatonina:~# apt -y install rabbitmq-server

Ahora en el fichero settings.py de nuestro proyecto añadimos

(venv) laura@melatonina:~/dev/whistleblowerbox$ vi wbox/settings.py
CELERY_BROKER_URL = 'amqp://localhost'

A continuación, en el mismo directorio donde se encuentra el fichero settings.py vamos a crear uno que se llame celery.py con este contenido

(venv) laura@melatonina:~/dev/whistleblowerbox$ vi wbox/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'wbox.settings')

app = Celery('wbox')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

En el fichero __init__.py también del mismo directorio donde se encuentra el settings.py vamos a añadirle lo siguiente:

(venv) laura@melatonina:~/dev/whistleblowerbox$ vi wbox/__init__.py
from .celery import app as celery_app

__all__ = ['celery_app']

A partir de aquí comprobamos que todo está correcto y que nuestra aplicación sigue funcionando.

Configurar una tarea
A continuación, donde tengamos la aplicación, en mi caso, en web, vamos a crear un fichero llamado tasks.py con el siguiente contenido

(venv) laura@melatonina:~/dev/whistleblowerbox$ vi web/tasks.py
from celery import shared_task
from web import views_account

@shared_task(name="send_email")
def send_email(to, subject, body):
    mail_sent = views_account.send_email(to, subject, body)
    return mail_sent

Aquí creamos una simple tarea que en este caso llama a views_account.send_email que es la función que tengo actualmente para mandar mails (parte de la ñapa que hice que consiste en hacer un modelo (MailQueue) que contiene todos los mails que envía el sistema).

Aquí podremos ver también como crear tareas con el cron.

Lanzar una tarea
Ahora para llamar a la tarea, es tan simple como llamar la función que está dentro de taks pero añadirle “delay”

(venv) laura@melatonina:~/dev/whistleblowerbox$ web/views_cron.py
from . import views_account, tasks

def mailqueue():
    mails = MailQueue.objects.filter(sent=0)

    for mail in mails:
        if mail.retention == 0:
            # send email
            #views_account.send_email(mail.to, mail.subject, mail.body)
            tasks.send_email.delay(mail.to, mail.subject, mail.body)

            if mail.cc != "" and mail.cc != None:
                tasks.send_email.delay(mail.cc, mail.subject, mail.body)

            if mail.cco != "" and mail.cco != None:
                tasks.send_email.delay(mail.cco, mail.subject, mail.body)
            mail.sent = 1
            mail.save()
        elif mail.retention <= 1:
            mail.retention -= 1
            mail.save()

Lanzar el worker de tareas
Para que las tareas puedan pasarse a rabbitmq, es necesario ejecutar celery de esta forma

(venv) laura@melatonina:~/dev/whistleblowerbox$ celery -A wbox worker -l info

 -------------- celery@melatonina v5.2.6 (dawn-chorus)
--- ***** ----- 
-- ******* ---- Linux-5.10.0-13-amd64-x86_64-with-glibc2.31 2022-04-14 13:10:08
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         wbox:0x7f13446a49a0
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . send_email

[2022-04-14 13:10:09,255: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2022-04-14 13:10:09,260: INFO/MainProcess] mingle: searching for neighbors
[2022-04-14 13:10:10,312: INFO/MainProcess] mingle: all alone
[2022-04-14 13:10:10,348: WARNING/MainProcess] /home/laura/dev/whistleblowerbox/venv/lib/python3.9/site-packages/celery/fixups/django.py:203: UserWarning: Using settings.DEBUG leads to a memory
            leak, never use this setting in production environments!
  warnings.warn('''Using settings.DEBUG leads to a memory

[2022-04-14 13:10:10,348: INFO/MainProcess] celery@melatonina ready.
[2022-04-14 13:10:23,688: INFO/MainProcess] Task send_email[1c0e449e-a005-46a0-9bf5-2993d719fc29] received
[2022-04-14 13:11:27,758: INFO/ForkPoolWorker-2] Task send_email[1c0e449e-a005-46a0-9bf5-2993d719fc29] succeeded in 30.0677882960008s: None

Aquí podemos ver que al ejecutar el trozo de código que envía los mails aún no procesados, se recibe la tarea y la ejecuta sin afectar a la velocidad de la aplicación o el resto del código que se esté ejecutando.

Poner el worker en producción
Lo siguiente, una vez probado que todo funciona correctamente en nuestro entorno de desarrollo, nos quedará prepararlo para el entorno de producción y hacer que el worker trabaje como un servicio. Para ello vamos a hacer uso de supervisord.

Lo instalaremos y también instalaremos rabbit-mq

root@condor-prod:~# apt -y install supervisor rabbitmq-server

A continuación configuraremos el celery en el supervisord, añadiendo un fichero en /etc/supervisor/conf.d/

root@condor-viadenuncia:~# vi /etc/supervisor/conf.d/wbox-celery.conf
[program:wbox-celery]
command=/var/www/whistleblowerbox/venv/bin/celery -A wbox worker --loglevel=INFO
directory=/var/www/whistleblowerbox/
user=nobody
numprocs=1
stdout_logfile=/var/log/celery/wbox-celery.log
stderr_logfile=/var/log/celery/wbox-celery.log
autostart=true
autorestart=true
startsecs=10

; Need to wait for currently executing tasks to finish at shutdown.
; Increase this if you have very long running tasks.
stopwaitsecs = 600

stopasgroup=true

; Set Celery priority higher than default (999)
; so, if rabbitmq is supervised, it will start first.
priority=1000

Cargaremos la configuración a supervisord:

root@condor-viadenuncia:~# supervisorctl reread
root@condor-viadenuncia:~# supervisorctl update

En /var/log/celery/wbox-celery.log vamos a poder ver si está todo correcto. Si hay algún problema lo veremos al cargar el supervisorctl o en los logs.

A continuación tendremos que instalar con pip3 install celery, celery en nuestra aplicación de producción y listos!