Présentation de Huey : une alternative au céleri pour Django

Arrière-plan

Aujourd'hui, je vais donc parler d'une alternative au céleri nommée Huey , qui est dotée d'une configuration beaucoup plus simple que le céleri et est beaucoup plus petite que le céleri.

La raison pour laquelle j'ai décidé d'essayer Huey est que j'ai parfois rencontré des problèmes avec Celery pour effectuer certaines tâches courantes, car la documentation n'est pas très complète.

Pour ceux qui ne savent pas ce qu'est le céleri ou qui ne l'ont jamais utilisé auparavant, Huey est une file d'attente de tâches asynchrone qui vous permet d'effectuer des tâches planifiées ou des tâches de longue durée en arrière-plan.

Conditions préalables

Nous allons installer les packages suivants :

• redis
• Django
• Huey
• requêtes (facultatif, nécessaire pour la démo)

Dépôt GitHub

Le blog suivant est accompagné d'un dépôt GitHub que vous pouvez utiliser pour tester le projet de démonstration que nous allons créer.

Cliquez ici pour voir le dépôt.

Configuration du projet

Créer le répertoire du projet

Ouvrez le terminal et tapez ce qui suit pour créer un répertoire : vous pouvez ignorer cette étape et le faire à partir de l'Explorateur de fichiers lui-même.

 mkdir huey_demo

Environnement virtuel

• Créons d'abord un virtualenv pour installer les dépendances de notre projet :

   python -m venv venv

  
  

• Activez le virtualenv (Linux) :

   source venv/bin/activate

Installation des dépendances

Tapez la commande suivante dans le terminal pour installer toutes les dépendances :

 pip install Django==4.0.4 redis==4.2.2 huey==2.4.3

Au moment de la rédaction de cet article, ce sont les versions avec lesquelles j'ai testé cette configuration, gardez un œil sur le dépôt Github pour toute mise à jour selon la dernière version à l'avenir.

Créer le projet

• Créez le projet Django en tapant la commande suivante dans le terminal :

   django-admin startproject django_huey_demo

  
  

• Changez de répertoire dans le répertoire du projet Django :

   cd django_huey_demo

  
  

• Créez l'application dans le cadre de notre projet :

   python manage.py startapp demo

  
  

• Incluez l'application créée dans le projet settings.py et apportez les modifications suivantes :

   INSTALLED_APPS = [ # Existing Apps "demo.apps.DemoConfig", # <== Add this line ]

  
  

• Définissez le mode de débogage sur False dans settings.py :

   DEBUG=False

  Nous définissons Debug sur False afin que nous puissions voir comment Huey fonctionne en production, nous en reparlerons plus tard.

Aperçu du projet

Maintenant que nous avons fini de mettre en place notre projet, c'est le bon moment pour vous présenter ce que nous allons construire aujourd'hui.

Nous récupérerons quotidiennement le "Mot du jour" à partir de l'API Wordnik . Ensuite, nous stockerons le mot, sa définition et un exemple du mot dans une phrase dans notre base de données.

Nous mettrons en place une tâche périodique en utilisant Huey qui récupérera le mot du jour et le stockera.

Pour stocker le mot, nous allons créer un modèle Django de celui-ci.

Obtenir la clé API Wordnik

Vous pouvez suivre ce guide pour obtenir la clé API.

Codage de notre projet

Ajoutez Huey à notre projet

Nous devons ajouter Huey aux applications installées de notre projet, alors apportez les modifications suivantes dans le fichier settings.py :

 INSTALLED_APPS = [ # Existing apps 'huey.contrib.djhuey', # <== Add this line ]

Installer Redis

Nous devons installer Redis pour Huey pour y stocker des informations sur les tâches en file d'attente, comme nous le faisions également avec Celery. Vous pouvez vous référer au lien suivant pour installer Redis en fonction de votre système d'exploitation spécifique.

Si vous êtes à l'aise avec Docker, vous pouvez utiliser la commande suivante :

 docker run --name redis_huey -p 6379:6379 -d redis

Par défaut, Huey essaiera de se connecter au serveur Redis exécuté sur localhost:6379 . S'il n'est pas présent, cela générera une erreur.

Définition du modèle

1. Ajoutez le code suivant à votre fichier demo/models.py :

    from django.db import models class Word(models.Model): word = models.CharField(max_length=200) part_of_speech = models.CharField(max_length=100) definition = models.TextField() example = models.TextField() def __str__(self): return self.word

   
   

2. Effectuer des migrations :

    python manage.py makemigrations demo

   
   

3. Appliquer les migrations :

    python manage.py migrate demo

Définition de la tâche

Créez un fichier nommé tasks.py dans le répertoire de l'application de démonstration. La raison pour laquelle nous avons nommé notre fichier tasks.py est d'aider Huey à découvrir automatiquement les tâches présentes dans nos applications enregistrées, si nous nommions notre fichier autrement que cela, nous devions enregistrer manuellement notre tâche. Si vous souhaitez en savoir plus, vous pouvez consulter la documentation Huey ici .

Avant d'écrire la définition de la tâche, nous devons installer une dépendance supplémentaire requests . Installez-le en tapant ce qui suit dans votre terminal :

 pip install requests==2.27.1

Maintenant, vient le code :

 import requests from django.conf import settings from huey import crontab from huey.contrib.djhuey import db_periodic_task from demo.models import Word @db_periodic_task(crontab(hour="18", minute="00")) def fetch_daily_word(): r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}") data = r.json() Word.objects.get_or_create( word=data["word"], part_of_speech=data["definitions"][0]["partOfSpeech"], definition=data["definitions"][0]["text"], example=data["examples"][0]["text"] )

Ajoutez la ligne suivante dans les paramètres de votre projet :

 WORDNIK_API_KEY = "api-key-here"

Ce bloc de code peut être long à prendre en compte, alors passons en revue les éléments qu'il contient un par un :

1. Huey Décorateur

    from huey.contrib.djhuey import db_periodic_task

   Il s'agit d'un décorateur fourni par Huey pour enregistrer les tâches périodiques qui impliquent de travailler avec la base de données, ce décorateur ferme automatiquement la connexion à la base de données une fois la tâche terminée, pour plus de détails, vous pouvez vous référer ici.

   
   

2. Calendrier Crontab

    @db_periodic_task(crontab(hour="18", minute="00"))

   
   

   Nous transmettons l'argument crontab(hour="18", minute="00") à notre décorateur de tâches périodiques, cela indique à Huey d'exécuter notre tâche à 18 heures tous les jours. Vous pouvez utiliser ce site pour créer vos plannings crontab, je l'utilise à chaque fois.

   
   

3. Clé API Wordnik

    from django.conf import settings # Usage ## settings.WORDNIK_API_KEY

   from django.conf import settings est le moyen standard d'importer des données à partir des paramètres de notre projet, il est utile dans les cas où nous avons plusieurs fichiers de paramètres configurés pour différents environnements afin qu'il sache quel fichier choisir sans que nous ayons à nous soucier de il. Il découvre quel fichier de paramètres nous utilisons à partir de la variable d'environnement DJANGO_SETTINGS_MODULE . Mais vous n'avez pas à vous soucier de ces détails.

   
   

   Ensuite, nous utilisons la clé dans notre appel API Wordnik.

   
   

4. Appel API Wordnik

    r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}")

   Ici, nous utilisons le module de requêtes pour effectuer une requête GET à l'API Wordnik tout en transmettant notre clé API pour l'authentification.

   
   

5. Stockage du mot dans la base de données

    data = r.json() Word.objects.get_or_create( word=data["word"], part_of_speech=data["definitions"][0]["partOfSpeech"], definition=data["definitions"][0]["text"], example=data["examples"][0]["text"] )

   Après avoir analysé la réponse de l'API, nous stockons la définition du mot dans notre base de données. Nous utilisons ici la méthode get_or_create au lieu de la méthode create afin de ne pas créer plusieurs copies du même mot dans notre base de données si ce mot est répété par l'API Wordnik.

   
   

6. Réponse de l'API Wordnik

   Voici à quoi ressemble la réponse de l'API Wordnik pour le point de terminaison Word of the Day. Certaines des sections non pertinentes de la réponse ont été tronquées par souci de concision.

    { "word": "stolon", "definitions": [ { "source": "ahd-5", "text": "A long thin stem that usually grows horizontally along the ground and produces roots and shoots at widely spaced nodes, as in a strawberry plant.", "note": null, "partOfSpeech": "noun" }, // More definitions here... ], "publishDate": "2022-05-08T03:00:00.000Z", "examples": [ { "title": "4.1 Nursery establishment", "text": "A stolon is a stem that grows along the ground, producing at its nodes new plants with roots and upright stems.", // Additional data here... }, // More examples here... ], // Additional fields here... }

Exécution d'un travailleur Huey

Vous pouvez démarrer le travailleur Huey en tapant la commande suivante dans votre terminal :

 python manage.py run_huey

Vous pouvez transmettre plusieurs indicateurs à la commande ci-dessus, ce qui modifiera ce qui est enregistré sur la console, par exemple :

• -v, --verbose - journalisation détaillée (inclut le niveau DEBUG)
• -q, --quiet - journalisation minimale
• -S, --simple - format de journalisation simple (« message horaire »)

Pour consulter diverses autres options de journalisation, consultez la documentation ici .

Que pouvez-vous faire d'autre avec Huey ?

Décorateurs de tâches

Huey est livré avec plusieurs décorateurs de tâches en fonction des opérations que vous effectuez dans le cadre de la tâche.

Je vais expliquer brièvement ce que font tous ces éléments ci-dessous.

Voici la déclaration d'importation pour tous les décorateurs :

 from huey.contrib.djhuey import task, periodic_task, db_task, db_periodic_task

• task : Une tâche régulière.
• periodic_task : lorsque vous souhaitez exécuter une tâche périodiquement en fonction d'un planning.
• db_task : lorsque vous souhaitez effectuer des opérations de base de données dans votre tâche.
• db_periodic_task : Lorsque vous souhaitez effectuer des opérations de base de données dans une tâche périodique.

Exemples de Crontab

Laissez-moi vous montrer quelques exemples supplémentaires de la façon dont vous pouvez utiliser crontab pour planifier vos tâches.

• crontab(minute='*/3') planifierait l'exécution de la tâche toutes les trois minutes.
• crontab(hour='*/3', minute='5') créerait une tâche qui s'exécuterait toutes les trois heures toutes les trois heures.
• crontab(minute='00', hour='10', month='*/2', day_of_week='*/5') créerait une tâche qui s'exécuterait tous les 5èmes jours de la semaine, tous les 2èmes mois à 10:00 DU MATIN.

Planification des tâches

Par exemple, la tâche suivante est définie dans tasks.py :

 from huey.contrib.djhuey import task @task() def count(): for i in range(10): print(i)

Maintenant, si vous souhaitez appeler cette tâche mais que vous souhaitez qu'elle s'exécute après 5 secondes, vous pouvez procéder comme suit :

 count.schedule(delay=5)

Le paramètre delay prend des valeurs en secondes, donc si vous souhaitez qu'il s'exécute après 5 minutes, spécifiez 300 secondes.

Nouvelle tentative de tâches qui échouent

Supposons que vous ajoutiez la logique suivante à notre tâche existante :

 @db_periodic_task(crontab(hour="18", minute="00"), retries=2) def fetch_daily_word(): r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}") if r.status_code != 200: raise Exception("Unable to fetch data from Wordnik API") ## Add this logic else: data = r.json() Word.objects.get_or_create( word=data["word"], part_of_speech=data["definitions"][0]["partOfSpeech"], definition=data["definitions"][0]["text"], example=data["examples"][0]["text"] )

Nous avons donc ajouté la logique pour vérifier le code d'état de la réponse, et s'il est autre que 200, il réessayera cette tâche jusqu'à 2 fois. Mais ces tentatives se produiraient sans aucun intervalle de temps entre les deux tentatives. Maintenant, que se passe-t-il si vous souhaitez retarder plusieurs tentatives de cette tâche ? Nous pouvons le faire en passant l'argument retry_delay , il accepte les valeurs en secondes.

 @db_periodic_task(crontab(hour="18", minute="00"), retries=2, retry_delay=10)

Cela entraînera un délai de 10 secondes entre plusieurs tentatives.

Mode de développement

Huey est livré avec un paramètre par défaut qui facilite le travail avec Huey pendant le développement dans Django. Ainsi, chaque fois que DEBUG=True est présent dans votre fichier settings.py , les tâches seront exécutées de manière synchrone, tout comme les appels de fonction normaux. Le but est d'éviter d'exécuter à la fois Redis et un processus consommateur supplémentaire lors du développement ou de l'exécution de tests. Vous pouvez en savoir plus à ce sujet ici .

Pour cela, nous devons ajouter la ligne suivante dans settings.py :

 HUEY = {}

Cependant, si vous souhaitez remplacer ce comportement, vous pouvez ajouter la configuration Huey suivante :

 HUEY = { "immediate": False }

Si vous disposez de la configuration ci-dessus mentionnée dans settings.py , tout en ayant DEBUG=True , Huey vous demandera de configurer Redis et d'exécuter Huey Worker à l'aide de la commande run_huey .

Céleri vs Huey

Voici quelques observations sur Huey par rapport au céleri :

• Moins d'empreinte de dépendances par rapport au céleri. Le céleri installe avec lui du kombu et du billard. Pendant ce temps, Huey n'a aucune dépendance.

• Des services moindres doivent être exécutés pour des tâches périodiques, Celery nécessite l'exécution d'un service beat et un service de travail pour travailler avec des tâches périodiques, tandis que nous n'avons besoin d'exécuter qu'un seul service à l'aide de la commande run_huey .

Les références

1. Huey Docs
2. API Wordnik
3. Dépôt Github associé