django-q2

Django Q2 バックグラウンドジョブ

タスクのキュー登録、スケジュール、ワーカーのデプロイ、Q_CLUSTER、または Django Q2 ワーカーがインポートするコードに手を加える前に、これを確認してください。

設定の確認

• 依存関係は django-q2 で、Python のインポートパスは django_q です。
• django_q が INSTALLED_APPS に含まれていることを確認してください。そのマイグレーションによって、タスク結果、スケジュール、ブローカーのモデルが提供されます。
• タスク、ワーカー、またはブローカーの動作を変更する前に、プロジェクトの設定モジュールで Q_CLUSTER を探してください。
• 設定されているブローカーを確認してください。Redis が一般的で、REDIS_URL などの環境変数を介して設定されていることがよくあります。ORM ブローカーは、スループットが低いデプロイや Redis を使わないデプロイに適しています。
• プロジェクトのワーカーコマンドを確認してください。基本のコマンドは python manage.py qcluster ですが、プロジェクトによっては uv、Poetry、Docker Compose、プロセスマネージャー、またはプラットフォーム固有のワーカー定義でラップされている場合があります。
• Redis をブローカーとして使わなくする場合は、Redis に依存するキャッシュ、ヘルスチェック、Docker、デプロイ、ドキュメントに関する記述も個別に確認してください。

基本的な仕組み

• Web コードから async_task(...) を呼び出すか、Schedule の行を作成します。
• ブローカーは、qcluster プロセスがタスクを予約するまで、キューに入ったタスクパッケージを保持します。
• ワーカープロセスはインポート可能な Python 関数を実行し、結果または失敗を保存します。
• スケジューラーはクラスター内で実行されます。スケジュールはデータベースの行であり、qcluster が実行されていなければ何も実行されません。

タスクの追加

1. タスク関数は、その処理を所有するアプリに配置します。通常は tasks.py モジュールか、プロジェクトですでに使われている別のインポート可能なモジュールです。
2. 関数はモジュールのインポート時にインポート可能な状態にしてください。リクエストオブジェクト、ローカルクロージャー、またはプロセス内だけで有効な状態に依存しないでください。
3. モデルインスタンス、開いているファイル、接続、大きなペイロードではなく、主キーなどの永続性のある識別子を渡してください。
4. タスクを冪等にしてください。Redis は厳密に一度だけ実行される保証を提供せず、受領確認を使うブローカーではタスクが再実行されることがあります。
5. タスクが保存直後のデータベース行に依存する場合は、transaction.on_commit(...) からキューに登録してください。

def send_welcome_email(user_id: int) -> None:
    from django.contrib.auth import get_user_model

    user = get_user_model().objects.get(pk=user_id)
    ...

from django.db import transaction
from django_q.tasks import async_task

transaction.on_commit(
    lambda: async_task("myapp.tasks.send_welcome_email", user.pk)
)

Django Q2 のオプションとタスクのキーワード引数が衝突する場合は、q_options を使用してください。

async_task(
    "myapp.tasks.rebuild_report",
    report_id,
    q_options={"timeout": 300, "group": "reports"},
)

作業のスケジュール設定

マイグレーション、管理アクション、またはセットアップコマンドで作成する、名前付きの冪等なスケジュールを優先してください。インポート時やアプリの起動時に無条件でスケジュールを作成するのは避けてください。

from django_q.models import Schedule

Schedule.objects.get_or_create(
    name="clear-expired-sessions",
    defaults={
        "func": "django.core.management.call_command",
        "args": "'clearsessions'",
        "schedule_type": Schedule.HOURLY,
    },
)

スケジュールの初期登録には Schedule.objects.get_or_create(name=..., defaults={...}) を使用し、セットアップを繰り返してもジョブが重複しないようにしてください。Cron スケジュールにはオプションの croniter 依存関係が必要です。プロジェクトに含まれていない場合は、Schedule.CRON を使用しないでください。

デフォルトでは、実行されなかったスケジュールは後から実行されます。停止中に逃した各間隔分を再実行するのではなく、再開後に一度だけ実行したいジョブでは、Q_CLUSTER["catch_up"] = False を設定してください。

ブローカーの選択肢

Redis ブローカー

プロジェクトがすでにワーカーやデプロイで Redis に依存している場合は、Redis を使用してください。

Q_CLUSTER = {
    "name": "...",
    "timeout": 3600,
    "workers": 4,
    "redis": REDIS_URL,
}

Redis は高速で、キャッシュ、Docker、またはデプロイ用のワーカーにすでに Redis を使っているプロジェクトに適しています。デフォルトの Redis ブローカーは受領確認に対応していません。タスクの実行中にワーカーホストが致命的な障害で停止すると、実行中のパッケージが失われる可能性があります。タスクコードで例外が発生した場合、Django Q2 は失敗を記録します。タスクを冪等に設計し、必要に応じてタスクコードで明示的な再試行を行い、失敗を監視してください。

ORM ブローカー

Django データベースブローカーは、スループットの低いデプロイ、ローカルでの簡便さ、または Redis を利用できない環境でのみ使用してください。

Q_CLUSTER = {
    "name": "...",
    "timeout": 3600,
    "retry": 4800,
    "workers": 4,
    "max_attempts": 2,
    "orm": "default",
}

ORM に切り替える場合：

• "redis" ブローカーキーを削除してください。カスタムクラスターを意図的に使用する場合を除き、クラスターごとにブローカーを1つ設定してください。
• django_q のマイグレーションを実行してください。ブローカーがデフォルト以外のデータベースエイリアスを使用する場合は、--database <alias> を指定してマイグレーションを実行してください。
• データベースへのポーリング負荷を軽減しつつ、キューの取得遅延を許容できる場合は、"poll": 2.0 のように "poll" をデフォルトの 0.2 秒より大きくしてください。
• ORM ブローカーを有効にすると、「Queued Tasks」の管理テーブルが使えるようになります。
• Redis に依存するキャッシュ、ヘルスチェック、Docker、デプロイの設定は個別に確認してください。スケジュールは常にデータベースの行として保存されます。ブローカー設定が制御するのはキューに登録されたタスクパッケージであり、スケジュールテーブルではありません。

テスト

• 関数を直接呼び出して、タスクのビジネスロジックをテストしてください。
• 同期モードでキューへの登録をテストします。
  • 呼び出しごとに設定する場合：async_task("myapp.tasks.fn", arg, sync=True)
  • テストごとに設定する場合：Q_CLUSTER["sync"] = True をオーバーライドします。
• ワーカーとブローカーの統合テストでは、プロジェクトの qcluster コマンドを別プロセスで実行し、result(task_id, 200) などの上限付き待機を使用してください。任意の時間のスリープには依存しないでください。
• 実際のワーカープロセスからコミット済みのデータベース行を参照する必要がある場合は、pytest.mark.django_db(transaction=True) を使用してください。

デバッグチェックリスト

• Web プロセスと同じ設定モジュール、SECRET_KEY、ブローカー URL、クラスター名を使って qcluster プロセスが実行されていますか？
• ワーカーはドット区切りのタスクパスをインポートできますか？
• django_q のマイグレーションを含め、データベースのマイグレーションは実行済みですか？
• Redis は Web コンテナーとワーカーコンテナーの両方から到達できますか？または、ORM ブローカーは想定したデータベースをポーリングしていますか？
• セットアップ時に安定した名前を付けずに別の Schedule 行を作成したことで、スケジュールされたタスクが重複していませんか？
• ダウンタイムによってスケジュールの追いつき処理が発生しましたか？
• タスクの失敗は Django 管理画面、ログ、または設定済みのエラーレポーターで確認できますか？

参考資料

• 公式ドキュメント：https://django-q2.readthedocs.io/en/master/
• 上流リポジトリ：https://github.com/django-q2/django-q2