背景 そこで今日は、Celery の代替品である についてお話します。これは、 よりもセットアップが簡単で、サイズも Celery に比べてずっと小さいです。 Huey 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 は で実行されている Redis サーバーに接続しようとします。存在しない場合は、エラーが発生します。 localhost:6379 モデル定義 次のコードを ファイルに追加します。 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 タスク定義 デモ アプリ ディレクトリに、 という名前のファイルを作成します。ファイルに という名前を付けた理由は、登録したアプリに存在するタスクを Huey が自動検出できるようにするためです。ファイルにこれ以外の名前を付けると、タスクを手動で登録する必要があります。詳細については、 Huey ドキュメントを参照してください。 tasks.py tasks.py こちら の タスク定義を記述する前に、追加の依存関係 をインストールする必要があります。ターミナルで次のように入力してインストールします。 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")) 引数 を定期タスク デコレータに渡します。これにより、Huey は毎日午後 6 時にタスクを実行します。crontab スケジュールを作成するには、 を利用できます。私はいつもこれを使用しています。 crontab(hour="18", minute="00") この Web サイト Wordnik API キー from django.conf import settings # Usage ## settings.WORDNIK_API_KEY 、プロジェクト設定からデータをインポートする標準的な方法です。これは、異なる環境用に複数の設定ファイルを設定している場合に便利です。どのファイルを選択するかがわかるので、心配する必要はありません。DJANGO_SETTINGS_MODULE 環境変数から、どの設定ファイルを使用しているかがわかります。ただし これらの詳細について心配する必要はありません。 from django.conf import settings 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 応答を解析した後、単語の定義をデータベースに保存します。ここでは、 メソッドではなく メソッドを使用しているため、Wordnik API によって同じ単語が繰り返される場合に、データベースに同じ単語の複数のコピーが作成されません。 create get_or_create 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 上記のコマンドに複数のフラグを渡すと、コンソールに記録される内容を変更できます。たとえば、次のようになります。 - 詳細なログ出力（DEBUG レベルを含む） -v, --verbose - 最小限のログ -q, --quiet - シンプルなログ形式 (“時間メッセージ”) -S, --simple ログ記録に関するその他のさまざまなオプションを確認するには、 ドキュメントを参照してください。 ここにある Huey で他に何ができるでしょうか? タスクデコレータ Huey には、タスク内で実行する操作に応じて複数のタスク デコレータが付属しています。 これらすべてが何をするのかを以下で簡単に説明します。 すべてのデコレータのインポート ステートメントは次のとおりです。 from huey.contrib.djhuey import task, periodic_task, db_task, db_periodic_task : 通常のタスク。 task : スケジュールに基づいて定期的にタスクを実行する場合。 periodic_task : タスク内で DB 操作を実行する場合。 db_task : 定期的なタスクで DB 操作を実行する場合。 db_periodic_task Crontab の例 crontab を使用してタスクをスケジュールする方法の例をいくつか紹介します。 タスクを 3 分ごとに実行するようにスケジュールします。 crontab(minute='*/3') 3 時間ごとに 5 分後に実行されるタスクを作成します。 crontab(hour='*/3', minute='5') 、2 か月ごとに 5 日目の午前 10 時に実行されるタスクを作成します。 crontab(minute='00', hour='10', month='*/2', day_of_week='*/5') タスクのスケジュール設定 たとえば、 内に次のタスクが定義されているとします。 tasks.py from huey.contrib.djhuey import task @task() def count(): for i in range(10): print(i) ここで、このタスクを呼び出して 5 秒後に実行したい場合は、次のようにします。 count.schedule(delay=5) パラメータは秒単位で値を取るため、5 分後に実行したい場合は 300 秒を指定します。 delay 失敗したタスクの再試行 既存のタスクに次のロジックを追加するとします。 @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 を使いやすくするデフォルト設定が付属しています。そのため、 ファイルに 存在する場合は常に、通常の関数呼び出しと同様にタスクが同期的に実行されます。この目的は、開発中またはテスト実行中に Redis と追加のコンシューマー プロセスの両方が実行されないようにすることです。詳細については 参照してください。 settings.py DEBUG=True 、ここ を このためには、 に次の行を追加する必要があります。 settings.py HUEY = {} ただし、この動作をオーバーライドする場合は、代わりに次の Huey 構成を追加できます。 HUEY = { "immediate": False } 上記の設定が に記載されており、 なっている場合、Huey では Redis をセットアップし、 コマンドを使用して Huey Worker を実行する必要があります。 settings.py DEBUG=True run_huey セロリ対ヒューイ Celery と比較した Huey に関するいくつかの観察事項は次のとおりです。 Celery と比較して依存関係のフットプリントが小さくなります。Celery は kombu と billiards も一緒にインストールします。一方、Huey には依存関係がありません。 定期的なタスクにはより少ないサービスを実行する必要があります。Celery では、定期的なタスクを処理するために beat サービスとワーカー サービスを実行する必要がありますが、 コマンドを使用して 1 つのサービスを実行するだけで済みます。 run_huey 参考文献 ヒューイドキュメント ワードニックAPI 関連 Github リポジトリ
