django-celery-expert

Django Celery エキスパート

手順

ステップ 1：リクエストを分類する

リクエストのタスクカテゴリを特定します。

• Django との統合 — トランザクションの安全性、ORM のパターン、テスト、リクエストの相関関係 → references/django-integration.md を読みます
• タスクの設計 — 新しいタスク、呼び出しパターン、チェーン／グループ／コーラス、冪等性 → references/task-design-patterns.md を読みます
• 設定 — ブローカーの設定、結果バックエンド、ワーカーの設定、キューのルーティング → references/configuration-guide.md を読みます
• エラー処理 — リトライ、バックオフ、デッドレターキュー、タイムアウト → references/error-handling.md を読みます
• 定期タスク — Celery Beat、crontab のスケジュール、動的スケジュール、タイムゾーンの処理 → references/periodic-tasks.md を読みます
• モニタリング — Flower、Prometheus、ログ記録、停止したタスクのデバッグ → references/monitoring-observability.md を読みます
• 本番環境へのデプロイ — スケーリング、プロセス監視、コンテナー、ヘルスチェック → references/production-deployment.md を読みます

リクエストが複数のカテゴリにまたがる場合は、続行する前に関連する参照ファイルをすべて読みます。

ステップ 2：参照ファイルを読む

ステップ 1 で特定した参照ファイルをそれぞれ読みます。関連する参照ファイルを読む前に、実装へ進んではいけません。

ステップ 3：実装する

参照ファイルのパターンを適用します。解決策を提示する前に、次の点を確認します。

• タスクの引数がシリアライズ可能である（モデルのインスタンスではなく ID を渡す）
• リトライが有効なタスクが冪等である
• エラーがコンテキストとともにログに記録される
• 長時間実行されるタスクにタイムアウトが設定されている

例

基本的なバックグラウンドタスク

リクエスト：「ユーザー登録後、バックグラウンドでウェルカムメールを送信する」

# tasks.py
from celery import shared_task
from django.core.mail import send_mail

@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id):
    from users.models import User

    try:
        user = User.objects.get(id=user_id)
        send_mail(
            subject="Welcome!",
            message=f"Hi {user.name}, welcome to our platform!",
            from_email="[email protected]",
            recipient_list=[user.email],
        )
    except User.DoesNotExist:
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))


# views.py — queue only after the transaction commits
from django.db import transaction

def register(request):
    user = User.objects.create(...)
    transaction.on_commit(lambda: send_welcome_email.delay(user.id))
    return redirect("dashboard")

進捗状況を追跡するタスク

リクエスト：「大きな CSV インポートを進捗状況の更新とともに処理する」

@shared_task(bind=True)
def import_csv(self, file_path, total_rows):
    from myapp.models import Record

    with open(file_path) as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader):
            Record.objects.create(**row)
            if i % 100 == 0:
                self.update_state(
                    state="PROGRESS",
                    meta={"current": i, "total": total_rows},
                )

    return {"status": "complete", "processed": total_rows}


# Poll progress
result = import_csv.AsyncResult(task_id)
if result.state == "PROGRESS":
    progress = result.info.get("current", 0) / result.info.get("total", 1)

チェーンを使ったワークフロー

リクエスト：「注文を処理する。在庫を確認し、支払いを処理してから、確認メールを送信する」

from celery import chain

@shared_task
def validate_inventory(order_id):
    order = Order.objects.get(id=order_id)
    if not order.items_in_stock():
        raise ValueError("Items out of stock")
    return order_id

@shared_task
def charge_payment(order_id):
    order = Order.objects.get(id=order_id)
    order.charge()
    return order_id

@shared_task
def send_confirmation(order_id):
    Order.objects.get(id=order_id).send_confirmation_email()

def process_order(order_id):
    chain(
        validate_inventory.s(order_id),
        charge_payment.s(),
        send_confirmation.s(),
    ).delay()