Celery と Docker を使用して Django で定期的なタスクを処理する方法

Django アプリケーションを構築して拡張していくと、必然的に特定のタスクをバックグラウンドで自動的かつ定期的に実行する必要が生じます。

例:

定期的なレポートを生成する

キャッシュをクリアする

一括メール通知を送信する

夜間メンテナンスを実行する

これは、Django コアの一部ではない、Web アプリケーションの構築と拡張に必要な数少ない機能の 1 つです。幸いなことに、Celery は Celery Beat と呼ばれる、実装が非常に簡単な強力なソリューションを提供しています。

次の記事では、Docker を使用して Django、Celery、Redis を設定し、Celery Beat 経由でカスタム Django Admin コマンドを定期的に実行する方法を説明します。

依存関係:

ジャンゴ v3.0.5

ドッカーv19.03.8

Python v3.8.2

セロリ v4.4.1

レディス v5.0.8

Django + Celery シリーズ:

Django と Celery による非同期タスク

Celery と Docker を使用して Django で定期的なタスクを処理する (この記事!)

ターゲット

このチュートリアルを終了すると、次のことができるようになります。

Docker で Django、Celery、Redis をコンテナ化する

CeleryをDjangoアプリケーションに統合し、タスクを作成する

カスタム Django Admin コマンドの作成

Celery Beat 経由で定期的に実行されるカスタム Django Admin コマンドをスケジュールする

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

django-celery-beatリポジトリからベースプロジェクトをクローンし、ベースブランチをチェックアウトします。

$ gitクローン
https://github.com/testdrivenio/django-celery-beat
--ブランチベース --シングルブランチ
$ cd django-celery-beat

管理するプロセスは合計 4 つ (Django、Redis、ワーカー、スケジューラ) あるため、Docker を使用してこれらのプロセスを接続し、単一のターミナルウィンドウから 1 つのコマンドですべて実行できるようにすることで、ワークフローを簡素化します。

プロジェクトのルートディレクトリからイメージを作成し、Docker コンテナを起動します。

$ docker-compose up -d --build
$ docker-compose exec web python manage.py migrate

ビルドが完了したら、http://localhost:1337 に移動して、アプリが期待どおりに実行されることを確認します。次のテキストが表示されます。

注文
注文が見つかりません!

プロジェクト構造:

├── .gitignore
├── docker-compose.yml
└── プロジェクト
├── Dockerファイル
├── コア
│ ├── __init__.py
│ ├── asgi.py
│ ├── 設定.py
│ ├── urls.py
│ └── wsgi.py
├── エントリポイント.sh
├── manage.py
├── 注文
│ ├── __init__.py
│ ├── admin.py
│ ├── アプリ.py
│ ├── 移行
│ │ ├── 0001_initial.py
│ │ └── __init__.py
│ ├── モデル.py
│ ├── テスト.py
│ ├── urls.py
│ └── views.py
├── 要件.txt
└── テンプレート
└── 注文
└── 注文リスト.html

セロリとRedis

ここで、Celery、Celery Beat、Redis のコンテナを追加する必要があります。

まず、依存関係を requirements.txt ファイルに追加します。

ジャンゴ==3.0.5
セロリ==4.4.1
レディス==3.4.1

docker-compose.ymlファイルの内容:

レディス:
 画像: redis:alpine
セロリ：
 ビルド: ./project
 コマンド: celery -A core worker -l info
 ボリューム:
 - ./project/:/usr/src/app/
 環境：
 -デバッグ=1
 - SECRET_KEY=dbaa1_i7%*3r9-=z-+_mz4r-!qeed@(-a_r(g@k8jo8y3r27%m
 - DJANGO_ALLOWED_HOSTS=ローカルホスト 127.0.0.1 [::1]
 依存:
 - レディス
セロリビート:
 ビルド: ./project
 コマンド: celery -A core beat -l info
 ボリューム:
 - ./project/:/usr/src/app/
 環境：
 -デバッグ=1
 - SECRET_KEY=dbaa1_i7%*3r9-=z-+_mz4r-!qeed@(-a_r(g@k8jo8y3r27%m
 - DJANGO_ALLOWED_HOSTS=ローカルホスト 127.0.0.1 [::1]
 依存:
 - レディス

また、Web サービスのdepends_on セクションを更新する必要があります。

ウェブ：
 ビルド: ./project
 コマンド: python manage.py runserver 0.0.0.0:8000
 ボリューム:
 - ./project/:/usr/src/app/
 ポート:
 - 1337:8000
 環境：
 -デバッグ=1
 - SECRET_KEY=dbaa1_i7%*3r9-=z-+_mz4r-!qeed@(-a_r(g@k8jo8y3r27%m
 - DJANGO_ALLOWED_HOSTS=ローカルホスト 127.0.0.1 [::1]
 依存:
 - redis # 新着

完全な docker-compose ファイルは次のとおりです。

バージョン: '3.7'
 
サービス:
 ウェブ：
 ビルド: ./project
 コマンド: python manage.py runserver 0.0.0.0:8000
 ボリューム:
 - ./project/:/usr/src/app/
 ポート:
 - 1337:8000
 環境：
 -デバッグ=1
 - SECRET_KEY=dbaa1_i7%*3r9-=z-+_mz4r-!qeed@(-a_r(g@k8jo8y3r27%m
 - DJANGO_ALLOWED_HOSTS=ローカルホスト 127.0.0.1 [::1]
 依存:
 - レディス
 レディス:
 画像: redis:alpine
 セロリ：
 ビルド: ./project
 コマンド: celery -A core worker -l info
 ボリューム:
 - ./project/:/usr/src/app/
 環境：
 -デバッグ=1
 - SECRET_KEY=dbaa1_i7%*3r9-=z-+_mz4r-!qeed@(-a_r(g@k8jo8y3r27%m
 - DJANGO_ALLOWED_HOSTS=ローカルホスト 127.0.0.1 [::1]
 依存:
 - レディス
 セロリビート:
 ビルド: ./project
 コマンド: celery -A core beat -l info
 ボリューム:
 - ./project/:/usr/src/app/
 環境：
 -デバッグ=1
 - SECRET_KEY=dbaa1_i7%*3r9-=z-+_mz4r-!qeed@(-a_r(g@k8jo8y3r27%m
 - DJANGO_ALLOWED_HOSTS=ローカルホスト 127.0.0.1 [::1]
 依存:
 - レディス

新しいコンテナを構築する前に、Django アプリケーションで Celery を構成する必要があります。

セロリの設定

設定

「core」ディレクトリに celery.py ファイルを作成し、次のコードを追加します。

インポートOS
セロリ輸入セロリ
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")
 
アプリ = Celery("core")
app.config_from_object("django.conf:settings", 名前空間="CELERY")
app.autodiscover_tasks()

ここで何が起こっているのですか?

まず、Celery が Django プロジェクトを見つける方法を認識できるように、 DJANGO_SETTINGS_MODULE 環境変数のデフォルト値を設定します。

次に、core という名前の新しい Celery インスタンスを作成し、その値を app という変数に割り当てました。

次に、django.conf の設定オブジェクトから Celery 設定値を読み込みました。他の Django 設定との競合を防ぐために、namespace="CELERY" を使用します。つまり、Celery のすべての構成設定には、 CELERY_ というプレフィックスを付ける必要があります。

最後に、 app.autodiscover_tasks（） 、 settings.INSTALLED_APPS で定義されたアプリケーションから Celery タスクを検索するように Celery に指示します。

core/__init__.py に次のコードを追加します。

.celeryからアプリをcelery_appとしてインポートします
 
__all__ = ("celery_app",)

最後に、Redis に接続できるように、次の Celery 設定で core/settings.py ファイルを更新します。

CELERY_BROKER_URL = "redis://redis:6379"
CELERY_RESULT_BACKEND = "redis://redis:6379"

建てる：

$ docker-compose up -d --build

ログを表示します:

$ docker-compose は 'web' をログに記録します
$ docker-compose は 'celery' をログに記録します
$ docker-compose は 'celery-beat' をログに記録します
$ docker-compose は 'redis' をログに記録します

すべてがうまくいけば、それぞれ異なるサービスを提供する 4 つのコンテナーが作成されます。

これで、サンプルタスクを作成して、正しく動作するかどうかを確認する準備が整いました。

タスクを作成する

新しいファイル core/tasks.py を作成し、コンソールに出力するだけのサンプルタスクの次のコードを追加します。

セロリからshared_taskをインポート

@共有タスク
sample_task() を定義します:
 print("サンプルタスクが実行されました。")

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

settings.py ファイルの最後に、Celery Beat を使用して sample_task が 1 分ごとに実行されるようにスケジュールするための次のコードを追加します。

CELERY_BEAT_SCHEDULE = {
 "サンプルタスク": {
 "タスク": "core.tasks.sample_task",
 "スケジュール": crontab(分="*/1"),
 },
}

ここでは、CELERY_BEAT_SCHEDULE 設定を使用して定期的なタスクを定義します。タスクに sample_task という名前を付け、2 つの設定を宣言しました。

Task は実行するタスクを宣言します。

スケジュールは、タスクを実行する時間間隔を設定します。これは整数、時間デルタ、または crontab になります。タスクでは crontab モードを使用し、1 分ごとに実行するように指示しました。 Celery のスケジュールの詳細については、こちらをご覧ください。

必ずインポートを追加してください:

celery.schedulesからcrontabをインポートする
 
core.tasksをインポートする

変更を適用するにはコンテナを再起動します。

$ docker-compose up -d --build

ログを表示します:

$ docker-compose ログ -f 'セロリ'
celery_1 | -------------- [キュー]
celery_1 | .> celery exchange=celery(direct) key=celery
セロリ_1 |
セロリ_1 |
celery_1 | [タスク]
セロリ_1 | . core.tasks.sample_task

Celery がサンプルタスク core.tasks.sample_task を取得したことがわかります。

1 分ごとに、ログに「The example task just ran」で終わる行が表示されます。

celery_1 | [2020-04-15 22:49:00,003: INFO/メインプロセス]
受信したタスク: core.tasks.sample_task[8ee5a84f-c54b-4e41-945b-645765e7b20a]
celery_1 | [2020-04-15 22:49:00,007: WARNING/ForkPoolWorker-1] サンプルタスクが実行されました。

Django 管理コマンドのカスタマイズ

Django には、次のような多くの組み込み django-admin コマンドが用意されています。

移住する

プロジェクトを開始する

スタートアプリ

データのダンプ

移民

組み込みコマンドに加えて、Django では独自のカスタムコマンドを作成するオプションも提供されています。

カスタム管理コマンドは、スタンドアロンスクリプトや、UNIX crontab または Windows のスケジュールされたタスクコントロールパネルから定期的に実行されるスクリプトを実行する場合に特に便利です。

したがって、最初に新しいコマンドを設定し、Celery Beat を使用してそれを自動的に実行します。

まず、orders/management/commands/my_custom_command.py という新しいファイルを作成します。次に、実行に必要な最小限のコードを追加します。

django.core.management.base から BaseCommand、CommandError をインポートします
 
 
クラス Command(BaseCommand):
 help = "コマンドの説明"
 
 def ハンドル(self, *args, **options):
 合格

BaseCommand にはオーバーライドできるメソッドがいくつかありますが、必須のメソッドは handle だけです。ハンドルはカスタムコマンドのエントリポイントです。つまり、コマンドを実行すると、このメソッドが呼び出されます。

テスト目的では、通常は簡単な print ステートメントを追加するだけです。ただし、Django のドキュメントによると、代わりに stdout.write を使用することをお勧めします。

管理コマンドを使用していてコンソール出力を提供する場合は、stdout と stderr に直接出力するのではなく、self.stdout と self.stderr に書き込む必要があります。これらのプロキシを使用すると、カスタムコマンドのテストがはるかに簡単になります。また、メッセージを改行で終了する必要はありません。終了パラメータを指定しない限り、改行は自動的に追加されます。

そこで、self.stdout.write コマンドを追加します。

django.core.management.base から BaseCommand、CommandError をインポートします
 
 
クラス Command(BaseCommand):
 help = "コマンドの説明"
 
 def ハンドル(self, *args, **options):
 self.stdout.write("サンプルコマンドが実行されました。") # NEW

テスト：

$ docker-compose exec web python manage.py my_custom_command
サンプルコマンドが実行されました。

それでは、すべてをまとめてみましょう。

Celery Beat でカスタムコマンドをスケジュールする

コンテナを起動して実行し、タスクを定期的に実行するようにスケジュールできることをテストし、カスタム Django Admin サンプルコマンドを記述したので、次はカスタムコマンドを定期的に実行するように設定します。

設定

このプロジェクトには、「Orders」と呼ばれる非常に基本的なアプリケーションがあります。 Product と Order の 2 つのモデルが含まれています。現在の日付からの注文を確認する電子メールレポートを送信するカスタムコマンドを作成しましょう。

まず、このプロジェクトに含まれるフィクスチャを使用して、いくつかの製品と注文をデータベースに追加します。

$ docker-compose exec web python manage.py loaddata products.json

スーパーユーザーを作成します。

$ docker-compose exec web python manage.py スーパーユーザーを作成します

プロンプトが表示されたら、ユーザー名、メールアドレス、パスワードを入力します。次に、Web ブラウザで http://127.0.0.1:1337/admin に移動します。作成したスーパーユーザーでログインし、いくつかの注文を作成します。少なくとも 1 つの日付が今日であることを確認してください。

電子メールレポート用の新しいカスタムコマンドを作成しましょう。

orders/management/commands/email_report.py というファイルを作成します。

datetime から timedelta、time、datetime をインポートします
 
django.core.mail から mail_admins をインポートします
django.core.management から BaseCommand をインポートします
django.utilsからタイムゾーンをインポートする
django.utils.timezone から make_aware をインポートします
 
orders.models から Order をインポート
 
今日 = タイムゾーン.now()
明日 = 今日 + 時間デルタ(1)
today_start = make_aware(datetime.combine(today, time()))
today_end = make_aware(datetime.combine(tomorrow, time()))
 
 
クラス Command(BaseCommand):
 help = "今日の注文レポートを管理者に送信する"
 
 def ハンドル(self, *args, **options):
 注文 = Order.objects.filter(confirmed_date__range=(today_start, today_end))
 
 注文の場合:
 メッセージ = ""
 
 注文の順序:
 メッセージ += f"{order} \n"
 
 件名 = (
 f"{today_start.strftime('%Y-%m-%d')} の注文レポート"
 f"to {today_end.strftime('%Y-%m-%d')}"
 ）
 
 mail_admins(subject=件名、message=メッセージ、html_message=なし)
 
 self.stdout.write("電子メールレポートが送信されました。")
 それ以外：
 self.stdout.write("本日は注文が確認されていません。")

コードでは、confirmed_date の日付の注文をデータベースで照会し、注文を電子メール本文の 1 つのメッセージに結合し、Django の組み込み mail_admins コマンドを使用して管理者に電子メールを送信しました。

ダミーの管理者メールを追加し、コンソールバックエンドを使用するように EMAIL_BACKEND を設定して、メールが設定ファイルの stdout に送信されるようにします。

EMAIL_BACKEND = "django.core.mail.backends.console.EmailBackend"
DEFAULT_FROM_EMAIL = "[email protected]"
管理者 = [("testuser", "[email protected]"), ]

走る：

$ docker-compose exec web python manage.py email_report
コンテンツタイプ: text/plain; charset="utf-8"
MIME バージョン: 1.0
コンテンツ転送エンコーディング: 7 ビット
件名: [Django] 2020-04-15 から 2020-04-16 までの注文レポート
送信元: root@localhost
宛先: [email protected]
日付: 2020 年 4 月 15 日 (水) 23:10:45 -0000
メッセージ ID: <158699224565.85.8278261495663971825@5ce6313185d3>

注文: 337ef21c-5f53-4761-9f81-07945de385ae - 製品: 米

---------------------------------------------------------------------------------
電子メールレポートが送信されました。

セロリビート

ここで、このコマンドを毎日実行する定期的なタスクを作成する必要があります。

core/tasks.py に新しいタスクを追加します。

セロリからshared_taskをインポート
django.core.management から call_command をインポート # 新規
 
 
@共有タスク
sample_task() を定義します:
 print("サンプルタスクが実行されました。")
 
 
＃新しい
@共有タスク
send_email_report() を定義します:
 call_command("email_report", )

そこでまず、django-admin コマンドをプログラムで呼び出すために使用される call_command インポートを追加しました。新しいタスクでは、カスタムコマンドの名前とともに call_command を引数として使用します。

このタスクをスケジュールするには、core/settings.py ファイルを開き、CELERY_BEAT_SCHEDULE 設定を更新して新しいタスクを含めます。

CELERY_BEAT_SCHEDULE = {
 "サンプルタスク": {
 "タスク": "core.tasks.sample_task",
 "スケジュール": crontab(分="*/1"),
 },
 「電子メールレポートを送信」: {
 "タスク": "core.tasks.send_email_report",
 "スケジュール": crontab(hour="*/1"),
 },
}

ここで、CELERY_BEAT_SCHEDULE に send_email_report という新しいエントリを追加しました。前のタスクと同様に、このタスクを実行するタスク (例: core.tasks.send_email_report) を宣言し、crontab モードを使用して繰り返しを設定します。

新しい設定が有効になっていることを確認するためにコンテナを再起動します。

$ docker-compose up -d --build
ログを見てください:
$ docker-compose ログ -f 'セロリ'
celery_1 | -------------- [キュー]
celery_1 | .> celery exchange=celery(direct) key=celery
セロリ_1 |
セロリ_1 |
celery_1 | [タスク]
セロリ_1 | . core.tasks.sample_task
celery_1 | . core.tasks.send_email_report

1分後、メールが送信されました。

celery_1 | [2020-04-15 23:20:00,309: 警告/ForkPoolWorker-1] コンテンツタイプ: text/plain; charset="utf-8"
celery_1 | MIME バージョン: 1.0
celery_1 | コンテンツ転送エンコーディング: 7 ビット
celery_1 | 件名: [Django] 2020-04-15 から 2020-04-16 までの注文レポート
celery_1 | 送信元: root@localhost
celery_1 | 宛先: [email protected]
celery_1 | 日付: 2020 年 4 月 15 日 (水) 23:20:00 -0000
celery_1 | メッセージ ID: <158699280030.12.8934112422500683251@42481c198b77>
セロリ_1 |
celery_1 | 注文: 337ef21c-5f53-4761-9f81-07945de385ae - 製品: 米
celery_1 | [2020-04-15 23:20:00,310: 警告/ForkPoolWorker-1] -------------------------------------------------------------------------------
celery_1 | [2020-04-15 23:20:00,312: WARNING/ForkPoolWorker-1] 電子メールレポートが送信されました。

結論は

この記事では、Celery、Celery Beat、Redis 用の Docker コンテナの設定方法について説明しました。次に、Celery Beat を使用してカスタム Django Admin コマンドと、そのコマンドを自動的に実行する定期的なタスクを作成する方法を示しました。

元記事: https://testdriven.io/blog/django-celery-periodic-tasks/

Celery と Docker を使用して Django で定期的なタスクを処理する方法についてはこれで終わりです。Celery Docker で Django の定期的なタスクを処理する方法の詳細については、123WORDPRESS.COM の以前の記事を検索するか、次の関連記事を引き続き参照してください。今後とも 123WORDPRESS.COM をよろしくお願いいたします。

以下もご興味があるかもしれません: