第7章実践：FastAPI と Celery を組み合わせた非同期処理 API

これまでに学んだ FastAPI、Uvicorn、Celery、および Redis を統合し、実践的な「非同期ジョブ実行 API」を構築します。

重い処理（例: レポート集計や画像生成）をリクエストとして受け付け、即座にタスク ID を返却し、クライアントが進捗確認および結果取得を行える完全なサンプルを完成させます。

7.1 プロジェクト構成

シンプルなディレクトリ構成で、API サーバーと Celery タスクを整理します。

my_api_project/
├── main.py          # FastAPI アプリケーション (エンドポイント定義)
├── worker.py        # Celery インスタンスと非同期タスク定義
└── requirements.txt # 依存パッケージ一覧

必要なパッケージは以下のとおりです。

pip install fastapi uvicorn celery redis httpx

7.2 Celery タスクの実装 (worker.py)

時間のかかる処理を担当する Celery アプリケーションとタスクを作成します。

処理中の進捗率を更新できるように、self.update_state() を使用します。

import time
from celery import Celery

REDIS_URL = "redis://localhost:6379/0"

celery_app = Celery(
    "worker",
    broker=REDIS_URL,
    backend=REDIS_URL,
)


@celery_app.task(bind=True)
def generate_report_task(self, report_name: str, total_steps: int = 5) -> dict:
    """疑似的に時間のかかるレポート作成タスク（進捗を更新しながら実行）"""
    print(f"[Task {self.request.id}] レポート作成開始: {report_name}")

    for step in range(1, total_steps + 1):
        time.sleep(1)  # 1秒ごとのステップ処理
        percent = int((step / total_steps) * 100)
        # クライアントへ進捗を伝えるために状態を PROGRESS に更新
        self.update_state(
            state="PROGRESS",
            meta={"current_step": step, "total_steps": total_steps, "percent": percent},
        )

    print(f"[Task {self.request.id}] レポート作成完了")
    return {
        "report_name": report_name,
        "status": "completed",
        "download_url": f"/files/reports/{self.request.id}.pdf",
    }

7.3 FastAPI エンドポイントの実装 (main.py)

FastAPI 側で、タスクの投入受付（POST）と状態照会（GET）の 2 つのエンドポイントを実装します。

from celery.result import AsyncResult
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

# worker.py から Celery アプリとタスク関数を読み込む
from worker import celery_app, generate_report_task

app = FastAPI(title="Async Job API")


class JobRequest(BaseModel):
    report_name: str = Field(..., min_length=1, description="作成するレポートの名前")
    steps: int = Field(default=5, ge=1, le=20, description="処理ステップ数")


class JobAcceptedResponse(BaseModel):
    job_id: str
    status: str
    message: str


@app.post(
    "/jobs",
    response_model=JobAcceptedResponse,
    status_code=status.HTTP_202_ACCEPTED,
)
def submit_job(req: JobRequest):
    """重いジョブを受け付け、即座にタスクIDを返す"""
    task = generate_report_task.delay(req.report_name, req.steps)
    return {
        "job_id": task.id,
        "status": "ACCEPTED",
        "message": "ジョブの受付が完了しました。進捗は /jobs/{job_id} で確認してください。",
    }


@app.get("/jobs/{job_id}")
def get_job_status(job_id: str):
    """タスク ID から現在の状態・進捗・結果を取得する"""
    task_result = AsyncResult(job_id, app=celery_app)

    response = {
        "job_id": job_id,
        "status": task_result.status,
    }

    if task_result.status == "PENDING":
        response["detail"] = "タスクは待機中または未開始です"
    elif task_result.status == "PROGRESS":
        # self.update_state() で渡した meta 情報が info に格納される
        response["progress"] = task_result.info
    elif task_result.status == "SUCCESS":
        response["result"] = task_result.result
    elif task_result.status == "FAILURE":
        response["error"] = str(task_result.result)

    return response

7.4 サービスの起動手順

この構成を動かすには、3 つのコンポーネントを起動します。

ステップ 1: Redis の起動

docker run -d --name test-redis -p 6379:6379 redis:alpine

ステップ 2: Celery ワーカーの起動

別のターミナルを開き、ワーカーを起動します。

celery -A worker.celery_app worker --loglevel=info

ステップ 3: FastAPI (Uvicorn) の起動

さらに別のターミナルを開き、Web サーバーを起動します。

uvicorn main:app --reload --port 8000

7.5 動作確認と実行フロー

curl コマンドを使って API の動作をテストします。

1. ジョブを登録する (POST /jobs)

curl -X POST http://127.0.0.1:8000/jobs \
  -H "Content-Type: application/json" \
  -d '{"report_name": "月次売上レポート", "steps": 5}'

レスポンス（即時返却）:

{
  "job_id": "c1f7a83d-9b1e-450a-8671-89c0a6b57731",
  "status": "ACCEPTED",
  "message": "ジョブの受付が完了しました。進捗は /jobs/{job_id} で確認してください。"
}

HTTP ステータス 202 Accepted とともに、待たされることなく即座にタスク ID が返ってきます。

2. 進捗状況を確認する (GET /jobs/{job_id})

2 秒後に進捗を確認します。

curl http://127.0.0.1:8000/jobs/c1f7a83d-9b1e-450a-8671-89c0a6b57731

レスポンス（実行中）:

{
  "job_id": "c1f7a83d-9b1e-450a-8671-89c0a6b57731",
  "status": "PROGRESS",
  "progress": {
    "current_step": 2,
    "total_steps": 5,
    "percent": 40
  }
}

3. 完了結果を取得する

6 秒後に再度確認します。

curl http://127.0.0.1:8000/jobs/c1f7a83d-9b1e-450a-8671-89c0a6b57731

レスポンス（完了）:

{
  "job_id": "c1f7a83d-9b1e-450a-8671-89c0a6b57731",
  "status": "SUCCESS",
  "result": {
    "report_name": "月次売上レポート",
    "status": "completed",
    "download_url": "/files/reports/c1f7a83d-9b1e-450a-8671-89c0a6b57731.pdf"
  }
}

このように、ユーザーに待ち時間を感じさせることなく、安全にバックグラウンドで重い処理を遂行できます。

7.6 本番化に向けた推奨事項

本番環境で運用する際は、以下のツールと設定を導入することで、さらなる安定性を確保できます。

Flower によるダッシュボード監視: Celery 公式のリアルタイム監視ツール Flower を起動することで、Web ブラウザからアクティブなタスク数、ワーカーの死活状態、失敗したタスクのリトライ履歴を一目で把握できます。
結果の有効期限設定 (result_expires): Redis に無制限に結果が溜まり続けるとメモリが圧迫されるため、celery_app.conf.result_expires = 3600（1時間後に自動削除）などを設定します。
エラー通知の自動化: Sentry などのエラー監視ツールを Celery ワーカーにも導入し、未処理例外が発生した際にアラートを受け取れるようにします。

Keyboard shortcuts