Celery & Redis : exécution de tâches en différé / asynchrones
Tags: django, celery, tasks, redis
Published: 2023-03-12 10:46:22

## Sommaire
  * [Préambule](#intro)
  * [Architecture](#archi)
  * [Code](#code)
    * [Code tâche / API et configuration Celery](#snippet)
    * [Docker](#docker)
  * [Sources](#sources)

## Préambule TL;DR <a name="intro"></a>

Lorsque l'on développe des applications Web, il peut arriver que l'on doive exécuter des tâches qui dépassent le _timeout_ du serveur Web : génération de gros fichiers, envoi de mails en masse, exécution d'une machine learning, traitement de fichiers importés, génération de rapports, ...tout un tas de cas d'utilisation qui arrivent régulièrement.

L'exécution de ce type de tâches est difficilement réalisable directement par le serveur d'applications Web, qui a un _timeout_ de configuré et qui doit être réduit (quelques secondes) sous peine d'effondrement (j'exagère...un peu 🥵😁).

Nous allons voir, grâce à [Django](https://www.djangoproject.com/), [Celery](https://docs.celeryq.dev/en/stable/) et [Redis](https://redis.com/) comment exécuter des tâches en différé, elles ne seront alors pas exécutées par le serveur d'application mais par un _processus_ à part : un _worker_ Celery.

## Architecture <a name="archi"></a>

Dans ce contexte, les différentes briques utilisées seront :

* **Django** pour le serveur d'applications Web avec le module Celery : il enverra une tâche à exécuter en _asynchrone_
* **Celery _worker_** : les processus applicatifs qui exécuteront les tâches qui leur seront envoyées
* **Redis** : servira à être un [_message broker_](https://fr.wikipedia.org/wiki/Agent_de_messages). En plus d'être un serveur de cache de données et/ou NoSql, Redis implémente le _pattern_ de messages [Pub/Sub](https://fr.wikipedia.org/wiki/Publish-subscribe) : il y a des producteurs / clients (serveur d'applications) qui envoient des messages et des consommateurs de messages qui vont les lire (celery workers) : Celery django va envoyer l'ordre d'exécuter une tâche et un des _workers_ va lire la tâche à exécuter et le faire. Le lancement des tâches est **atomique**, c'est à dire que la tâche ne va être prise que par un seul _worker_.

On peut représenter cette communication entre applications et processus avec ces différentes briques de la façon suivante :

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5fsg36mhiy09ezg6enrd.png)

(source : https://excalidraw.com/#json=MpX5kp_uYfL17ugUS6a-S,CgyUY0Aqanul0DCksVDzuA)

## Un peu de code ! <a name="code"></a>

On va s'amuser à créer une tâche qui va attendre 5 secondes avant de nous dire qu'elle démarre, cette tâche ne sera pas exécuter dans le serveur d'applications mais dans le _worker_ celery, en différé : le serveur d'applications rend la main immédiatement, la tâche s'exécute en temps voulu

### Modules

Il nous faudra a minima comme modules, pour le côté client :

- django & djangorestframwork [pour une API de lancement d'une tâche]
- celery
- redis

Pour le côté serveurs :

- un serveur d'applications Web avec une API qui lance une tâche
- un _worker_ qui réceptionnera notre tâche à exécuter
- un redis pour l'échange de messages entre le client Web et le _worker_ Celery

### Docker <a name="docker"></a>

4 images ont été préparées pour lancer 4 conteneurs : nginx pour le reverse proxy host -> conteneur, serveur d'application **django**, un _worker_ **celery**, le _message broker_ **redis**

au lancement de ces services via `docker-compose up` on aura 

```bash
$ docker ps
CONTAINER ID   IMAGE                    COMMAND                  CREATED          STATUS          PORTS                                            NAMES
6caf76fca92e   api-and-worker           "celery -A core.asyn…"   11 minutes ago   Up 11 minutes   8100/tcp                                         celery-worker
9cadb4c717a5   nginx                    "nginx -g 'daemon of…"   11 minutes ago   Up 11 minutes   0.0.0.0:80->80/tcp                               nginx-api-server
b83db3cf05eb   redis:7.0.8              "docker-entrypoint.s…"   11 minutes ago   Up 11 minutes   0.0.0.0:6379->6379/tcp                           redis-broker
654b6d1222de   api-and-worker           "python3 manage.py r…"   11 minutes ago   Up 11 minutes   8100/tcp                                         api-server
```

avec le `no-trunc` pour voir les paramètres des services lancés que l'on peut retrouver dans le docker-compose :

```bash
$ docker ps --no-trunc
CONTAINER ID                                                       IMAGE                    COMMAND                                       CREATED          STATUS          PORTS
                               NAMES
6caf76fca92e5b4e83663019f34f57b5780fd3a9244d07daeb431f5f0e57b255   api-and-worker           "celery -A core.async_tasks worker -l info"   11 minutes ago   Up 11 minutes   8100/tcp
                               celery-worker
9cadb4c717a535d609dbd43ee2bb65667efd1b74f873e6e5b93efd7be365ab3a   nginx                    "nginx -g 'daemon off;'"                      11 minutes ago   Up 11 minutes   0.0.0.0:80->80/tcp                               nginx-api-server
b83db3cf05eb7c9d9aaa4fb3ce120e98f2820f1040434d336ca024e0d1f0a5b6   redis:7.0.8              "docker-entrypoint.sh redis-server"           11 minutes ago   Up 11 minutes   0.0.0.0:6379->6379/tcp                           redis-broker
654b6d1222de36b56f3e3d7bc9d13376ec7e54b527c316bbac7649576a82263f   api-and-worker           "python3 manage.py runserver 0.0.0.0:8110"    11 minutes ago   Up 11 minutes   8100/tcp
                               api-server
```

### Code <a name="snippet"></a>

On aura :

- un module configuration celery : broker de message à utiliser, paramétrage Celery (tâches à prendre en compte, serialisation, timeout, etc)
- un module "tasks", commun à l'application et au _worker_ sera nécessaire pour définir la tâche et l'appeler ensuite, une API qui déclenchera cette tâche

Déclaration de la tâche, on met une temporisation pour démontrer l'intérêt :

```python
from time import sleep
import logging
logger = logging.getLogger('django')
from core.async_tasks.celery import app

@app.task
def execute_long_task(title: str, delay: int, wordings: list):
    logger.info(f"** On s'endort {delay} secondes avant de lancer la tâche ZZzzzZZzzzzzzzzz **")

    sleep(delay)

    logger.info(f'** On se réveille ! Démarrage de la tâche {title} avec {wordings} **')
    logger.info(f'** Terminée !')
```

API d'envoi et d'ordre d'exécution de la tâche:

```python
class ExecuteTaskApi(APIView):
    permission_classes = [AllowAny]
    renderer_classes = (JSONRenderer,)

    def get(self, request, *args, **kwargs):
        execute_long_task.delay('Exécute ma longue tâche !', 5,
                                ['Vais attendre', 'le super pouvoir de Celery'])
        return Response({'data': 'je réponds immédiatement sans attendre la fin de la tâche !'})
```

Configuration de Celery

```python
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings")

app = Celery("tasks")
app.config_from_object("core.async_tasks.celery_config")

# par défaut, il recherche les tâches dans le module "tasks.py" https://docs.celeryq.dev/en/stable/reference/celery.html#celery.Celery.autodiscover_tasks
app.autodiscover_tasks(['api'])
```

Logs de l'exécution de http://plateform/api/execute_task/ sur le navigateur ou par curl 

```bash
$ curl http://plateform/api/execute_task/
{"data":"je réponds immédiatement sans attendre la fin de la tâche !"}
``` 

Le serveur d'application envoie l'ordre d'exécuter la tâche et rend immédiatement la main au navigateur, le _worker_ réceptionne la tâche à exécuter, il y a une temporisation de 5 secondes pour démontrer l'aspect différé / asynchrone :

```bash
api-server        | INFO basehttp "GET /api/execute_task/ HTTP/1.0" 200 73
celery-worker     | [2023-03-12 11:19:00,851: INFO/MainProcess] Received task: api.tasks.execute_long_task[3f1b8339-4192-45da-9607-23246b532af7]
celery-worker     | INFO tasks ** On s'endort 5 secondes avant de lancer la tâche ZZzzzZZzzzzzzzzz **
celery-worker     | [2023-03-12 11:19:00,854: INFO/ForkPoolWorker-8] ** On s'endort 5 secondes avant de lancer la tâche ZZzzzZZzzzzzzzzz **
celery-worker     | INFO tasks ** On se réveille ! Démarrage de la tâche Exécute ma longue tâche ! avec ['Vais attendre', 'le super pouvoir de Celery'] **
celery-worker     | [2023-03-12 11:19:05,860: INFO/ForkPoolWorker-8] ** On se réveille ! Démarrage de la tâche Exécute ma longue tâche ! avec ['Vais attendre', 'le super pouvoir de Celery'] **
celery-worker     | INFO tasks ** Terminée !
celery-worker     | [2023-03-12 11:19:05,861: INFO/ForkPoolWorker-8] ** Terminée !

```

La tâche a duré **5 secondes**, entre 11:19:00 et 11:19:05, indépendamment du serveur Web et de son _timout_ ! 

On peut dès lors imaginer des tâches qui pourraient durer quelques minutes pour effectuer des traitements particuliers.

# Sources <a name="sources"></a>

Celery permet d'autres paramétrages comme la reprise de tâche en cas d'échec ou de s'assurer qu'une tâche ne sera pas exécutée 2 fois, etc

Il existe aussi un outil de monitoring pour visualiser l'état des tâches : [Flower](https://flower.readthedocs.io/en/latest/)

Ou encore, la possibilité d'avoir des tâches récurrentes et programmées, "à la crontab", grâce à [Celery Beat](https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html).

Retrouvez les **sources sur le Github** : https://github.com/zorky/celery-redis-django-tasks