python-background-jobs

出典に関する声明：本スキルの内容は、wshobson/agents （MIT License, Copyright (c) 2024 Seth Hobson）を出典としています。ライセンスの全文は同じディレクトリ内の LICENSE.txt を参照してください。 使用、変更、再配布は、当該 MIT ライセンスの条項に従い、著作権表示を保持する必要があります。

日本語の説明

このスキルは、フルスタック開発に関連するベストプラクティスとパターンに関するガイダンスを提供します。

---

name: python-background-jobs description: Python のバックグラウンドジョブのパターン（タスクキュー、ワーカー、イベント駆動型アーキテクチャなど）。非同期タスク処理、ジョブキュー、長時間実行される処理、またはリクエスト／レスポンスサイクルから処理を切り離す実装に使用します。

Python のバックグラウンドジョブとタスクキュー

長時間実行される処理や信頼性の低い処理を、リクエスト／レスポンスサイクルから切り離します。ユーザーにはすぐに応答し、バックグラウンドワーカーが非同期で重い処理を実行します。

このスキルを使う場面

• 数秒以上かかるタスクを処理する場合
• メール、通知、Webhook を送信する場合
• レポートの生成やデータのエクスポートを行う場合
• アップロードやメディア変換を処理する場合
• 信頼性の低い外部サービスと連携する場合
• イベント駆動型アーキテクチャを構築する場合

基本概念

1. タスクキューパターン

API がリクエストを受け付け、ジョブをキューに登録し、ジョブ ID をすぐに返します。ワーカーはジョブを非同期で処理します。

2. 冪等性

タスクは失敗時に再試行されることがあります。安全に再実行できるように設計してください。

3. ジョブの状態遷移

ジョブは次の状態に遷移します：保留中 → 実行中 → 成功／失敗。

4. 少なくとも 1 回の配信

ほとんどのキューは、少なくとも 1 回の配信を保証します。コードで重複を処理できるようにする必要があります。

クイックスタート

このスキルでは、広く採用されているタスクキューである Celery を例に使用します。RQ、Dramatiq、クラウドネイティブなソリューション（AWS SQS、GCP Tasks）などの代替手段も、同様に有効な選択肢です。

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

@app.task
def send_email(to: str, subject: str, body: str) -> None:
    # This runs in a background worker
    email_client.send(to, subject, body)

# In your API handler
send_email.delay("[email protected]", "Welcome!", "Thanks for signing up")

基本的なパターン

パターン 1：ジョブ ID をすぐに返す

数秒以上かかる処理では、ジョブ ID を返し、非同期で処理します。

from uuid import uuid4
from dataclasses import dataclass
from enum import Enum
from datetime import datetime

class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

@dataclass
class Job:
    id: str
    status: JobStatus
    created_at: datetime
    started_at: datetime | None = None
    completed_at: datetime | None = None
    result: dict | None = None
    error: str | None = None

# API endpoint
async def start_export(request: ExportRequest) -> JobResponse:
    """Start export job and return job ID."""
    job_id = str(uuid4())

    # Persist job record
    await jobs_repo.create(Job(
        id=job_id,
        status=JobStatus.PENDING,
        created_at=datetime.utcnow(),
    ))

    # Enqueue task for background processing
    await task_queue.enqueue(
        "export_data",
        job_id=job_id,
        params=request.model_dump(),
    )

    # Return immediately with job ID
    return JobResponse(
        job_id=job_id,
        status="pending",
        poll_url=f"/jobs/{job_id}",
    )

パターン 2：Celery タスクの設定

適切な再試行とタイムアウトの設定で Celery タスクを構成します。

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

# Global configuration
app.conf.update(
    task_time_limit=3600,          # Hard limit: 1 hour
    task_soft_time_limit=3000,      # Soft limit: 50 minutes
    task_acks_late=True,            # Acknowledge after completion
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,   # Don't prefetch too many tasks
)

@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
)
def process_payment(self, payment_id: str) -> dict:
    """Process payment with automatic retry on transient errors."""
    try:
        result = payment_gateway.charge(payment_id)
        return {"status": "success", "transaction_id": result.id}
    except PaymentDeclinedError as e:
        # Don't retry permanent failures
        return {"status": "declined", "reason": str(e)}
    except TransientError as e:
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)

パターン 3：タスクを冪等にする

ワーカーはクラッシュやタイムアウト時に再試行する場合があります。安全に再実行できるように設計してください。

@app.task(bind=True)
def process_order(self, order_id: str) -> None:
    """Process order idempotently."""
    order = orders_repo.get(order_id)

    # Already processed? Return early
    if order.status == OrderStatus.COMPLETED:
        logger.info("Order already processed", order_id=order_id)
        return

    # Already in progress? Check if we should continue
    if order.status == OrderStatus.PROCESSING:
        # Use idempotency key to avoid double-charging
        pass

    # Process with idempotency key
    result = payment_provider.charge(
        amount=order.total,
        idempotency_key=f"order-{order_id}",  # Critical!
    )

    orders_repo.update(order_id, status=OrderStatus.COMPLETED)

冪等性を実現する方法：

1. 書き込み前の確認：処理の前に状態を確認する
2. 冪等性キー：外部サービスで一意のトークンを使用する
3. アップサートパターン：INSERT ... ON CONFLICT UPDATE
4. 重複排除期間：処理済み ID を N 時間追跡する

パターン 4：ジョブの状態管理

可視性とデバッグのために、ジョブの状態遷移を永続化します。

class JobRepository:
    """Repository for managing job state."""

    async def create(self, job: Job) -> Job:
        """Create new job record."""
        await self._db.execute(
            """INSERT INTO jobs (id, status, created_at)
               VALUES ($1, $2, $3)""",
            job.id, job.status.value, job.created_at,
        )
        return job

    async def update_status(
        self,
        job_id: str,
        status: JobStatus,
        **fields,
    ) -> None:
        """Update job status with timestamp."""
        updates = {"status": status.value, **fields}

        if status == JobStatus.RUNNING:
            updates["started_at"] = datetime.utcnow()
        elif status in (JobStatus.SUCCEEDED, JobStatus.FAILED):
            updates["completed_at"] = datetime.utcnow()

        await self._db.execute(
            "UPDATE jobs SET status = $1, ... WHERE id = $2",
            updates, job_id,
        )

        logger.info(
            "Job status updated",
            job_id=job_id,
            status=status.value,
        )

詳細な実例とパターン

詳細なセクション（## Advanced Patterns から始まる内容）は references/details.md にあります。上記の概要だけでは不十分な場合は、そのファイルを読んでください。

ベストプラクティスのまとめ

1. すぐに応答する - 長時間の処理でリクエストをブロックしない
2. ジョブの状態を永続化する - ステータスのポーリングとデバッグを可能にする
3. タスクを冪等にする - あらゆる失敗時に安全に再試行できるようにする
4. 冪等性キーを使用する - 外部サービスを呼び出す場合に使用する
5. タイムアウトを設定する - ソフトリミットとハードリミットの両方を設定する
6. DLQ を実装する - 恒久的に失敗したタスクを捕捉する
7. 状態遷移をログに記録する - ジョブの状態変化を追跡する
8. 適切に再試行する - 一時的なエラーには指数バックオフを使用する
9. 恒久的な失敗は再試行しない - 検証エラー、無効な認証情報など
10. キューの深さを監視する - バックログの増加を検知して通知する

能力の範囲

✅ 得意なこと

• フルスタック開発に関連するベストプラクティス
• コードアーキテクチャの設計ガイダンス
• パフォーマンス最適化の提案

⚠️ 素材が必要なこと

• 具体的なプロジェクトのコンテキスト
• 技術スタックの情報

❌ 対応範囲外

• 技術以外の分野におけるビジネス上の意思決定
• 具体的なコードの実装（ガイダンスのみを提供）

ワークフロー

ステップ 1：要件分析

プロジェクトの要件と技術的な制約を理解します

ステップ 2：アーキテクチャ設計

システムアーキテクチャとコンポーネントの分割を設計します

ステップ 3：実装ガイダンス

具体的な実装の提案とコードパターンを提供します

ステップ 4：コードレビュー

コードの品質とベストプラクティスへの準拠状況を確認します

ステップ 5：最適化の提案

パフォーマンスの最適化と改善の提案を行います

注意点

1. 過剰な設計

単純な状況で複雑な設計パターンを使わないようにします

2. パフォーマンスの軽視

特に本番環境では、パフォーマンスの最適化を軽視しないでください

3. セキュリティへの配慮

セキュリティとデータ保護を常に考慮してください

4. テストカバレッジ

特に境界条件について、十分なテストカバレッジを確保してください

5. ドキュメントの保守

ドキュメントをコードと同期させて更新してください