576 測定値

Huey の紹介: Django の Celery 代替品

に Udit11m2024/06/14

長すぎる; 読むには

Huey は、Celery 非同期バックグラウンドタスクキューの軽量で簡単に構成できる代替手段です。複雑さを軽減しながら信頼性の高いタスク管理を実現する、よりシンプルなソリューションをお探しの場合は、Huey がプロジェクトに最適です。Huey が機能性を損なうことなくプロセスを合理化する方法をご覧ください。

featured image - Huey の紹介: Django の Celery 代替品

背景

そこで今日は、Celery の代替品であるHueyについてお話します。これは、 Celeryよりもセットアップが簡単で、サイズも Celery に比べてずっと小さいです。

Huey を試してみることにした理由は、ドキュメントがあまり良くないため、いくつかの一般的なタスクを実行するときに Celery で時々問題が発生することがあったからです。

Celery が何であるかを知らない人や、これまで使用したことがない人のために説明すると、Huey は、スケジュールされたタスクや長時間実行されるタスクをバックグラウンドで実行できる非同期タスクキューです。

前提条件

以下のパッケージをインストールします。

レディス
ジャンゴ
ヒューイ
リクエスト（オプション、デモに必要）

GitHub リポジトリ

次のブログには、これから作成するデモプロジェクトをテストするために使用できる GitHub リポジトリが付属しています。

リポジトリを表示するにはここをクリックしてください。

プロジェクトのセットアップ

プロジェクトディレクトリを作成する

ターミナルを開き、次のように入力してディレクトリを作成します。この手順をスキップして、ファイルエクスプローラー自体から実行することもできます。

 mkdir huey_demo

仮想環境

まず、プロジェクトの依存関係をインストールするために virtualenv を作成しましょう。
```
 python -m venv venv
```
virtualenv をアクティブ化します (Linux):
```
 source venv/bin/activate
```

依存関係のインストール

すべての依存関係をインストールするには、ターミナルで次のコマンドを入力します。

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

この記事の執筆時点では、このセットアップをテストしたバージョンはこれらでした。今後最新バージョンに関する更新があるかどうかは、Github リポジトリで確認してください。

プロジェクトを作成する

ターミナルで次のコマンドを入力して、Django プロジェクトを作成します。
```
 django-admin startproject django_huey_demo
```
Django プロジェクトディレクトリにディレクトリを変更します。
```
 cd django_huey_demo
```
プロジェクトの下にアプリを作成します。
```
 python manage.py startapp demo
```
作成したアプリをプロジェクトsettings.pyに含め、次の変更を加えます。
```
 INSTALLED_APPS = [ # Existing Apps "demo.apps.DemoConfig", # <== Add this line ]
```
settings.pyでデバッグモードをFalseに設定します。
```
 DEBUG=False
```
Huey が本番環境でどのように実行されるかを確認できるように、Debug を False に設定しています。これについては後で詳しく説明します。

プロジェクト概要

プロジェクトの設定が完了したので、今日構築する内容について説明するのにちょうど良いタイミングです。

Wordnik APIから毎日「今日の単語」を取得します。次に、単語、その定義、および文中の単語の例をデータベースに保存します。

Huey を使用して、今日の単語を取得して保存する定期的なタスクを設定します。

単語を保存するために、同じ Django モデルを作成します。

Wordnik API キーの取得

API キーを取得するには、このガイドに従ってください。

プロジェクトのコーディング

私たちのプロジェクトにHueyを追加する

プロジェクトのインストール済みアプリに Huey を追加する必要があるため、 settings.pyファイルで次の変更を加えます。

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

Redisをインストールする

Celery で行っていたのと同様に、キューに入れられたタスクに関する情報を Huey に保存するには、Redis をインストールする必要があります。特定のオペレーティングシステムに基づいて Redis をインストールするには、次のリンクを参照してください。

Docker の使用に慣れている場合は、次のコマンドを使用できます。

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

デフォルトでは、Huey はlocalhost:6379で実行されている Redis サーバーに接続しようとします。存在しない場合は、エラーが発生します。

モデル定義

次のコードを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

移行を行う:
```
 python manage.py makemigrations demo
```
移行を適用します。
```
 python manage.py migrate demo
```

タスク定義

デモアプリディレクトリに、 tasks.pyという名前のファイルを作成します。ファイルにtasks.pyという名前を付けた理由は、登録したアプリに存在するタスクを Huey が自動検出できるようにするためです。ファイルにこれ以外の名前を付けると、タスクを手動で登録する必要があります。詳細については、こちらのHuey ドキュメントを参照してください。

タスク定義を記述する前に、追加の依存関係requestsをインストールする必要があります。ターミナルで次のように入力してインストールします。

 pip install requests==2.27.1

さて、コードは次のようになります:

 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"] )

プロジェクト設定に次の行を追加します。

 WORDNIK_API_KEY = "api-key-here"

このコードブロックは理解するのが難しいかもしれませんので、その内容を 1 つずつ確認してみましょう。

ヒューイ・デコレーター
```
from huey.contrib.djhuey import db_periodic_task
```
これは、データベースの操作を伴う定期的なタスクを登録するために Huey によって提供されるデコレータです。このデコレータは、タスクの完了時にデータベース接続を自動的に閉じます。詳細については、ここを参照してください。
Crontab スケジュール
```
@db_periodic_task(crontab(hour="18", minute="00"))
```
引数crontab(hour="18", minute="00")を定期タスクデコレータに渡します。これにより、Huey は毎日午後 6 時にタスクを実行します。crontab スケジュールを作成するには、この Web サイトを利用できます。私はいつもこれを使用しています。
Wordnik API キー
```
from django.conf import settings # Usage ## settings.WORDNIK_API_KEY
```
from django.conf import settings 、プロジェクト設定からデータをインポートする標準的な方法です。これは、異なる環境用に複数の設定ファイルを設定している場合に便利です。どのファイルを選択するかがわかるので、心配する必要はありません。DJANGO_SETTINGS_MODULE 環境変数から、どの設定ファイルを使用しているかがわかります。ただしDJANGO_SETTINGS_MODULEこれらの詳細について心配する必要はありません。

次に、Wordnik API 呼び出しでキーを使用します。
Wordnik API 呼び出し
```
r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}")
```
ここでは、認証用の API キーを渡しながら、リクエストモジュールを使用して Wordnik API に GET リクエストを送信します。
単語をデータベースに保存する
```
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"] )
```
API 応答を解析した後、単語の定義をデータベースに保存します。ここでは、 createメソッドではなくget_or_createメソッドを使用しているため、Wordnik API によって同じ単語が繰り返される場合に、データベースに同じ単語の複数のコピーが作成されません。

Wordnik API レスポンス

Word of the Day エンドポイントに対する Wordnik API 応答は次のようになります。応答の無関係なセクションの一部は、簡潔にするために切り捨てられています。

 { "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... }

ヒューイワーカーの実行

ターミナルで次のコマンドを入力すると、Huey ワーカーを起動できます。

 python manage.py run_huey

上記のコマンドに複数のフラグを渡すと、コンソールに記録される内容を変更できます。たとえば、次のようになります。

-v, --verbose - 詳細なログ出力（DEBUG レベルを含む）
-q, --quiet - 最小限のログ
-S, --simple - シンプルなログ形式 (“時間メッセージ”)

ログ記録に関するその他のさまざまなオプションを確認するには、ここにあるドキュメントを参照してください。

Huey で他に何ができるでしょうか?

タスクデコレータ

Huey には、タスク内で実行する操作に応じて複数のタスクデコレータが付属しています。

これらすべてが何をするのかを以下で簡単に説明します。

すべてのデコレータのインポートステートメントは次のとおりです。

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

task : 通常のタスク。
periodic_task : スケジュールに基づいて定期的にタスクを実行する場合。
db_task : タスク内で DB 操作を実行する場合。
db_periodic_task : 定期的なタスクで DB 操作を実行する場合。

Crontab の例

crontab を使用してタスクをスケジュールする方法の例をいくつか紹介します。

crontab(minute='*/3')タスクを 3 分ごとに実行するようにスケジュールします。
crontab(hour='*/3', minute='5') 3 時間ごとに 5 分後に実行されるタスクを作成します。
crontab(minute='00', hour='10', month='*/2', day_of_week='*/5') 、2 か月ごとに 5 日目の午前 10 時に実行されるタスクを作成します。

タスクのスケジュール設定

たとえば、 tasks.py内に次のタスクが定義されているとします。

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

ここで、このタスクを呼び出して 5 秒後に実行したい場合は、次のようにします。

 count.schedule(delay=5)

delayパラメータは秒単位で値を取るため、5 分後に実行したい場合は 300 秒を指定します。

失敗したタスクの再試行

既存のタスクに次のロジックを追加するとします。

 @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"] )

そこで、レスポンスのステータスコードをチェックするロジックを追加しました。ステータスコードが 200 以外の場合は、そのタスクを最大 2 回再試行します。ただし、これらの再試行は、2 回の試行の間に時間差なしで行われます。では、このタスクの複数の試行を遅らせたい場合はどうすればよいでしょうか。これは、秒単位で値を受け入れるretry_delay引数を渡すことで実行できます。

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

これにより、複数回の試行の間に 10 秒の遅延が発生します。

開発モード

Huey には、Django での開発中に Huey を使いやすくするデフォルト設定が付属しています。そのため、 settings.pyファイルにDEBUG=True存在する場合は常に、通常の関数呼び出しと同様にタスクが同期的に実行されます。この目的は、開発中またはテスト実行中に Redis と追加のコンシューマープロセスの両方が実行されないようにすることです。詳細については、ここを参照してください。

このためには、 settings.pyに次の行を追加する必要があります。