python-background-jobs

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

実行時間の長い処理や不安定な処理を、リクエスト/レスポンスのサイクルから切り離します。バックグラウンドワーカーが重い処理を非同期で担う間、ユーザーには即座に応答を返します。

このスキルを使用する場面

• 数秒以上かかるタスクの処理
• メール、通知、Webhook の送信
• レポートの生成やデータのエクスポート
• アップロードやメディア変換の処理
• 不安定な外部サービスとの連携
• イベント駆動アーキテクチャの構築

中核となる概念

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

API はリクエストを受け取り、ジョブをエンキューし、ジョブ ID を返して即座に応答します。ワーカーはジョブを非同期で処理します。

2. 冪等性

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

3. ジョブの状態遷移

ジョブは pending → running → succeeded/failed という状態を遷移します。

4. At-Least-Once 配信

ほとんどのキューは at-least-once 配信を保証します。コードは重複を処理できる必要があります。

クイックスタート

このスキルでは、広く採用されているタスクキューである 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. キューの深さを監視する - バックログの増加時にアラートを出す