Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Python API サーバ実践ガイド アウトライン

現代の Web サービス開発では、高スループットな API サーバと、重い処理を非同期に切り離すバックグラウンドワーカーの組み合わせが標準的な設計となっています。

Python においては、型ヒントを最大限に活用した高速な ASGI フレームワークである FastAPI、ASGI サーバとして広く利用される Uvicorn、そして非同期タスクキューとして実績のある Celery を組み合わせることで、堅牢でスケーラブルな Web API 基盤を構築できます。

本書では、Python の基礎知識（文法や uv / Ruff による環境構築等）を前提とし、FastAPI による API 設計、Uvicorn による ASGI サーバー運用、そして Celery と Redis による非同期タスク処理の連携までを体系的に解説します。

章構成

  • 1. Web API と非同期アーキテクチャの全体像

    Web API の基本概念、同期処理（WSGI）と非同期処理（ASGI）の違い、FastAPI・Uvicorn・Celery・Redis が担う役割の分担、およびシステム全体のデータフローを整理します。

  • 2. FastAPI による Web API の基本設計

    FastAPI アプリケーションの初期化、HTTP メソッドに応じたルーティング、Pydantic によるリクエストとレスポンスの型バリデーション、ステータスコードとエラーハンドリングの基本を学びます。

  • 3. Depends による依存性注入の実践

    FastAPI の強力な仕組みである依存性注入（Depends）の概念、共通認証処理の実装、データベースセッションや設定値の注入パターン、およびテスト容易性の向上方法を解説します。

  • 4. 非同期処理と Uvicorn 運用設計

    async def と通常の def の使い分け、イベントループとブロッキング I/O の注意点、Uvicorn の起動オプション、ワーカープロセスの設定、および本番環境向け運用のベストプラクティスを解説します。

  • 5. バックグラウンドタスクと Celery の使い分け

    FastAPI 標準の BackgroundTasks と分散タスクキュー Celery の決定的な違い、インメモリ処理の限界、永続化・リトライ・ワーカー分散が必要となる判断基準を整理します。

  • 6. Celery と Redis による非同期タスク基盤

    Message Broker と Result Backend の仕組み、Celery アプリケーションの構成、タスクの定義と呼び出し（delay / apply_async）、ワーカープロセスの起動、およびタスク状態・結果の取得方法を学びます。

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

    FastAPI、Uvicorn、Celery、Redis を連携させた実践的な非同期ジョブ API を構築します。重い処理のリクエスト受付、タスク ID の即時返却、進捗状態の確認、および完了結果の取得までの一連の処理フローを実装します。

第1章 Web API と非同期アーキテクチャの全体像

Web アプリケーションやスマートフォンアプリは、インターネットを通じてサーバーとデータをやり取りします。その中心的な役割を果たすのが Web API サーバーです。

本章では、Web API の基本的な仕組み、従来の同期通信とモダンな非同期通信の違い、そして FastAPI、Uvicorn、Celery、Redis を組み合わせた全体アーキテクチャを学びます。

1.1 Web API とクライアント・サーバーの関係

Web API（Application Programming Interface）は、プログラム同士がネットワーク経由でデータを受け渡すための約束事です。

ブラウザやスマートフォンなどのクライアントは、HTTP リクエストをサーバーに送信します。サーバーはその要求を処理し、JSON 形式などのデータを HTTP レスポンスとして返却します。ユーザーの画面表示とサーバーのデータ処理を分離できるため、柔軟なシステム開発が可能になります。

1.2 同期処理（WSGI）と非同期処理（ASGI）

従来の Python Web アプリケーションでは、WSGI（Web Server Gateway Interface）という標準規格が使われていました。

WSGI は 1 つのリクエストを 1 つのスレッドやプロセスが専有して処理する「同期型」です。データベースへの問い合わせや外部 API 通信など、通信の待ち時間（I/O 待ち）が発生している間もプロセスが塞がってしまい、大量の同時接続を処理するには限界がありました。

一方、ASGI（Asynchronous Server Gateway Interface）は「非同期処理（async / await）」に対応した新しい規格です。通信の待ち時間が発生した際に、別のリクエストの処理へ CPU を切り替えることができるため、少ないリソースで数万規模の同時アクセスを効率よく処理できます。

graph TD
    subgraph WSGI["同期型 (WSGI)"]
        W_Req["リクエスト 1"] --> W_Wait["I/O 待ち (CPU 占有)"] --> W_Res["レスポンス 1"]
        W_Req2["リクエスト 2"] --> W_Block["待機 (前の処理が終わるまで待つ)"]
    end

    subgraph ASGI["非同期型 (ASGI)"]
        A_Req1["リクエスト 1"] --> A_Wait["I/O 待ち発生"]
        A_Wait -.->|"待ち時間に切り替え"| A_Req2["リクエスト 2 の処理開始"]
    end

1.3 FastAPI・Uvicorn・Celery・Redis の役割分担

高速で安定した Web サービスを構築するために、それぞれの得意分野に応じたコンポーネントを組み合わせて使用します。

  • Uvicorn: ASGI に準拠した Web サーバーです。クライアントからの HTTP 接続を直接受け付け、高速な非同期イベントループで FastAPI にリクエストを渡します。

  • FastAPI: Python の型ヒントを駆使した最新の Web フレームワークです。URL ルーティング、リクエストの型検証、レスポンスの生成、API 仕様書（Swagger UI）の自動生成を担当します。

  • Redis: メモリ上でデータを高速に保持するインメモリデータストアです。ここでは非同期タスクの待ち行列を管理する「メッセージブローカー」および結果の保存先として利用します。

  • Celery: 時間のかかる重い処理をバックグラウンドで非同期に実行する「分散タスクキュー」です。API サーバーとは別の独立したプロセス（ワーカー）として動作します。

1.4 システム全体のデータフローと構成図

ユーザーがアクセスしたときの全体の流れを整理します。

  1. クライアントが Uvicorn へ HTTP リクエストを送信します。
  2. Uvicorn がリクエストを FastAPI アプリケーションに渡します。
  3. 数秒以内に完了する軽量なデータ取得や更新は、FastAPI がそのまま処理して即座に応答します。
  4. 画像変換や一括メール送信、機械学習推論など数秒以上かかる重い処理は、FastAPI が Redis にタスクを投入し、タスク ID のみをクライアントに即座に返却します。
  5. 裏で待機している Celery ワーカーが Redis からタスクを取り出して実行し、完了結果を Redis に保存します。
sequenceDiagram
    autonumber
    actor Client as クライアント
    participant Uvicorn as Uvicorn (ASGI サーバー)
    participant FastAPI as FastAPI (API アプリ)
    participant Redis as Redis (Broker / Backend)
    participant Celery as Celery (ワーカー)

    Client->>Uvicorn: HTTP POST /tasks (重い処理依頼)
    Uvicorn->>FastAPI: リクエスト転送
    FastAPI->>Redis: タスクを登録 (タスク ID 発行)
    FastAPI-->>Client: 202 Accepted (タスク ID を即座に返却)
    
    par バックグラウンド実行
        Redis->>Celery: タスクを通知
        Celery->>Celery: 重い処理を実行
        Celery->>Redis: 実行結果を保存
    end

    Client->>FastAPI: HTTP GET /tasks/{task_id} (進捗確認)
    FastAPI->>Redis: 状態・結果を問い合わせ
    FastAPI-->>Client: 200 OK (進捗 or 完了結果)

第2章 FastAPI による Web API の基本設計

FastAPI は、現代的な Python の型ヒントを最大限に活用した、直感的で書きやすい Web フレームワークです。

本章では、FastAPI アプリケーションの基本的な作り方、URL ルーティング、Pydantic モデルを用いた入力検証、ステータスコードの制御、および自動生成されるドキュメント機能について学びます。

2.1 FastAPI アプリケーションの起動とルーティング

FastAPI の最小構成は、FastAPI クラスのインスタンスを作成し、デコレータを用いて関数にパス（URL）と HTTP メソッドを紐付けることです。

以下のコードは、ルートパス / に GET リクエストが届いた際に JSON を返す最小限のアプリケーション例です。

from fastapi import FastAPI

app = FastAPI(title="Sample API", version="1.0.0")


@app.get("/")
def read_root() -> dict[str, str]:
    return {"message": "Hello, FastAPI"}

このファイルを main.py として保存した場合、ターミナルから uvicorn main:app --reload を実行するとサーバーが起動し、http://127.0.0.1:8000/ でアクセスできるようになります。

2.2 パスパラメータとクエリパラメータ

API では、URL の一部に変数を埋め込む「パスパラメータ」と、末尾に ?key=value の形式で付与する「クエリパラメータ」を使い分けます。

FastAPI では、関数の引数に型アノテーションを記述するだけで、自動的に文字列から数値への変換と妥当性チェックが行われます。

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None) -> dict[str, int | str | None]:
    return {"item_id": item_id, "query": q}

パス内に記述された {item_id} はパスパラメータとして int 型に変換されます。パスに含まれない引数 q は自動的にクエリパラメータとして扱われます。もし item_id に文字（例: /items/abc）を指定すると、FastAPI は自動的にエラー（422 Unprocessable Entity）を返します。

2.3 Pydantic によるリクエストボディの検証

POST や PUT メソッドで JSON データを受け取る場合は、Pydantic の BaseModel を継承したクラスを定義します。

Pydantic モデルを使うことで、期待するフィールドの名前、型、デフォルト値、必須チェックを宣言的に定義できます。

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50, description="商品名")
    price: int = Field(..., gt=0, description="単価（1円以上）")
    tax_rate: float = Field(default=0.1, ge=0.0, le=1.0, description="消費税率")


@app.post("/items")
def create_item(item: ItemCreate) -> dict[str, str | int | float]:
    total_price = int(item.price * (1 + item.tax_rate))
    return {
        "name": item.name,
        "price": item.price,
        "total_price": total_price,
    }

クライアントから送られた JSON は自動的に解析され、型と制約条件（文字数や数値範囲）が検証されます。ルール違反がある場合、どのフィールドにどんな不備があったのかを詳細な JSON エラーとして返却してくれます。

2.4 レスポンスモデルと HTTP ステータスコード

API の設計では、クライアントへ返すデータの構造を明確にし、適切な HTTP ステータスコード（成功時は 200 または 201、エラー時は 400 や 404 など）を設定することが不可欠です。

response_model 引数をデコレータに指定すると、余計な内部データの漏洩を防ぎ、出力データの型を保証できます。

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()


class UserOut(BaseModel):
    id: int
    username: str
    email: str


users_db: dict[int, dict[str, str | int]] = {
    1: {"id": 1, "username": "alice", "email": "alice@example.com", "hashed_password": "secret"},
}


@app.get("/users/{user_id}", response_model=UserOut, status_code=status.HTTP_200_OK)
def get_user(user_id: int):
    user = users_db.get(user_id)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User with id {user_id} was not found",
        )
    return user

上記コードでは、データベース内に hashed_password が存在していても、response_model=UserOut で定義されたフィールドのみが JSON としてクライアントに返されるため、意図しない機密情報の漏洩を防ぐことができます。

2.5 自動生成される対話型 API ドキュメント

FastAPI を起動すると、ブラウザからアクセスできる対話型ドキュメントが追加の設定なしで自動生成されます。

  • Swagger UI: http://127.0.0.1:8000/docs にアクセスすると利用できます。ブラウザ上で各エンドポイントの仕様を確認し、実際にリクエストを送信して動作をテストできます。
  • ReDoc: http://127.0.0.1:8000/redoc にアクセスすると利用できます。読みやすく整形された API ドキュメントを閲覧できます。

コード中に書いた型ヒント、Pydantic モデル、および説明文（description や関数の docstring）がそのまま仕様書に反映されるため、コードとドキュメントの乖離が発生しません。

第3章 Depends による依存性注入の実践

実務の Web API 開発では、認証・権限チェック、データベース接続の取得と破棄、共通設定の読み込みなど、多くのエンドポイントで共有される処理が発生します。

FastAPI は、これらを美しく解決する「依存性注入（Dependency Injection: DI）」の強力なシステムを備えています。本章では、Depends を活用した共通処理の設計手法を学びます。

3.1 依存性注入（Dependency Injection）とは何か

依存性注入とは、ある処理が必要とする機能やデータ（依存関係）を、その処理自身が直接生成するのではなく、外部から渡してもらう設計手法です。

例えば、ユーザー情報を取得するエンドポイントが自前で認証トークンの検証やデータベースの接続作成を行うと、コードが重複し、テストも困難になります。FastAPI の DI 機構を使うことで、共通処理を独立した関数として切り離し、エンドポイントの引数に注入することができます。

3.2 Depends の基本的な使い方

FastAPI では、fastapi.Depends を使用して依存関係を宣言します。

関数の引数に Depends(依存関数) を指定すると、FastAPI はエンドポイントが実行される直前にその依存関数を実行し、戻り値を引数に渡します。

from fastapi import Depends, FastAPI

app = FastAPI()


def pagination_params(skip: int = 0, limit: int = 10) -> dict[str, int]:
    return {"skip": skip, "limit": limit}


@app.get("/items")
def list_items(params: dict[str, int] = Depends(pagination_params)):
    return {
        "message": "Items fetched",
        "skip": params["skip"],
        "limit": params["limit"],
    }


@app.get("/users")
def list_users(params: dict[str, int] = Depends(pagination_params)):
    return {
        "message": "Users fetched",
        "skip": params["skip"],
        "limit": params["limit"],
    }

共通のクエリパラメータ処理（ページネーション情報）を 1 つの関数に集約でき、複数のエンドポイントで手軽に再利用できます。

3.3 認証・認可処理の共通化

DI の最も代表的な用途の 1 つが、API キーやトークンによる認証処理です。

HTTP ヘッダーから認証情報を取得し、不正な場合に即座に例外を発生させる依存関数を定義します。

from fastapi import Depends, FastAPI, HTTPException, Header, status

app = FastAPI()

VALID_API_KEYS = {"secret-token-12345": "AdminUser"}


def verify_api_key(x_api_key: str = Header(...)) -> str:
    user_name = VALID_API_KEYS.get(x_api_key)
    if not user_name:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API Key",
        )
    return user_name


@app.get("/protected-data")
def get_protected_data(current_user: str = Depends(verify_api_key)):
    return {
        "status": "success",
        "data": "This is sensitive data",
        "operator": current_user,
    }

エンドポイント側は current_user: str = Depends(verify_api_key) と宣言するだけで、認証に合格したユーザー名のみを受け取ることができ、ビジネスロジックに集中できます。

3.4 データベースセッションのライフサイクル管理

リクエストごとにデータベース接続を開き、処理が終了したら確実にクローズする処理には、yield を用いたジェネレータ関数を利用します。

from collections.abc import Generator
from fastapi import Depends, FastAPI


class DatabaseSession:
    def __init__(self):
        print("DB接続をオープンしました")

    def close(self):
        print("DB接続をクローズしました")


def get_db() -> Generator[DatabaseSession, None, None]:
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()


app = FastAPI()


@app.get("/records")
def list_records(db: DatabaseSession = Depends(get_db)):
    return {"message": "Records fetched using active DB session"}

yield の前にあるコードがエンドポイントの実行前に走り、レスポンスがクライアントへ返された後（または途中で例外が発生した場合でも）、finally ブロックのクリーンアップ処理が確実に実行されます。

3.5 テスト時の依存性オーバーライド

DI 機構を採用する大きな利点は、単体テストや結合テストが極めて容易になる点です。

FastAPI の app.dependency_overrides 辞書を使うと、テスト実行時のみ本物の依存関係をモック（ダミー）関数に差し替えることができます。

from fastapi.testclient import TestClient

client = TestClient(app)


def mock_verify_api_key():
    return "TestMockUser"


# テスト用に依存関係を差し替え
app.dependency_overrides[verify_api_key] = mock_verify_api_key

response = client.get("/protected-data")
assert response.status_code == 200
assert response.json()["operator"] == "TestMockUser"

# テスト終了後に元に戻す
app.dependency_overrides.clear()

本物の認証サーバーや外部データベースを用意することなく、確実で高速なテストを実行できます。

第4章 非同期処理と Uvicorn 運用設計

FastAPI が高いパフォーマンスを発揮できる理由は、Python の非同期処理機能（async / await）と、超高速な ASGI サーバー Uvicorn の組み合わせにあります。

本章では、非同期処理の正しい使い分け、やってはいけないアンチパターン、そして Uvicorn を本番環境で安定して稼働させるための運用設計を解説します。

4.1 async def と通常の def の使い分け

FastAPI では、エンドポイント関数を async def で定義することも、通常の def で定義することも可能です。

FastAPI は、関数の宣言方法に応じて内部での実行方式を自動的に切り替えます。

  • async def: 非同期 I/O ライブラリ（例: httpx による外部 API 通信、非同期 DB ドライバ asyncpg など）を await する場合に使用します。メインのイベントループ上で高速に並行実行されます。
  • 通常の def: 同期ライブラリ（例: 従来の同期型 DB ドライバ、ファイル読み書き、重い計算処理など）を扱う場合に使用します。FastAPI は自動的にスレッドプールへ処理を逃がし、メインのイベントループがブロックされるのを防ぎます。
import asyncio
from fastapi import FastAPI
import httpx

app = FastAPI()


# 非同期I/O: await可能な処理がある場合
@app.get("/async-fetch")
async def fetch_external_data():
    async with httpx.AsyncClient() as client:
        res = await client.get("https://httpbin.org/delay/1")
        return {"status": res.status_code}


# 同期処理: 外部スレッドプールで安全に実行される
@app.get("/sync-calculation")
def compute_heavy_math():
    result = sum(i * i for i in range(1_000_000))
    return {"result": result}

4.2 イベントループをブロックするアンチパターン

非同期プログラミングで最も注意しなければならないのが、async def の中でブロッキング処理（同期的な待ち時間）を実行してしまうことです。

例えば、async def の中で標準の time.sleep() や同期的な requests.get() を呼び出すと、イベントループ全体が停止し、その間に届いた他のすべてのユーザーのリクエストが完全に待たされてしまいます。

# 悪い例: イベントループ全体を止めてしまう
@app.get("/bad")
async def bad_endpoint():
    import time

    time.sleep(3)  # 全員のリクエストが止まる！
    return {"message": "bad"}


# 良い例: 非同期スリープを使用する
@app.get("/good")
async def good_endpoint():
    import asyncio

    await asyncio.sleep(3)  # 他のリクエストは並行して処理される
    return {"message": "good"}

4.3 Uvicorn の起動オプションと基本設定

Uvicorn は uvloop（C言語で書かれた超高速イベントループ）と httptools を内蔵した ASGI サーバーです。

開発環境と本番環境では、起動オプションを適切に使い分けます。

# 開発時: コードの変更を検知して自動再起動 (--reload)
uvicorn main:app --host 127.0.0.1 --port 8000 --reload

# 本番時: 複数ワーカープロセスで起動 (--workers)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

主なオプションの意味は以下のとおりです。

  • --host: 待ち受ける IP アドレスを指定します。コンテナ環境（Docker）では外部からの通信を受け付けるため 0.0.0.0 を指定します。
  • --port: ポート番号（既定値は 8000）を指定します。
  • --workers: 起動するワーカープロセスの数です。Python は GIL（Global Interpreter Lock）があるため、CPU コア数を有効活用するには複数プロセスで起動します。

4.4 本番環境でのプロセス管理とワーカー数設計

本番サーバーのスペックに応じて、適切なワーカー数を設計します。

推奨される一般的なワーカー数の計算式は以下のとおりです。

$$ \text{ワーカー数} = (2 \times \text{CPU コア数}) + 1 $$

例えば 2 コアのサーバーであれば 4〜5 ワーカーが目安です。

また、本番環境でプロセスの異常終了検知やグレースフルリスタートを行いたい場合は、プロセス管理機構として Gunicorn を親プロセスにし、ワーカーとして UvicornWorker を動作させる構成が広く採用されています。

gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

4.5 リバースプロキシとヘルスチェック

本番運用では、Uvicorn の前段に Nginx や Cloudflare、AWS ALB などのリバースプロキシを配置し、SSL/TLS 暗号化解除や静的ファイル配信を任せます。

ロードバランサーからの死活監視に対応するため、必ず専用のヘルスチェック用エンドポイントを用意します。

@app.get("/healthz", tags=["System"])
def health_check():
    return {"status": "ok"}

第5章 バックグラウンドタスクと Celery の使い分け

Web サービスを開発していると、メール送信、PDF 出力、画像のリサイズ、機械学習モデルの推論など、時間のかかる処理を実行したい場面が頻繁に訪れます。

FastAPI には軽量な BackgroundTasks が組み込まれていますが、システムの規模や信頼性の要件によって、本格的な分散タスクキューである Celery が必要になります。本章では、両者の違いと使い分けの基準を学びます。

5.1 なぜ Web リクエスト内で重い処理をしてはいけないのか

Web クライアント（ブラウザやアプリ）は、API リクエストを送った後、レスポンスが返ってくるまで待機します。

もしリクエスト処理の中で数秒から数十秒かかる重い処理を実行すると、ユーザーの画面がフリーズするだけでなく、以下の深刻な問題が発生します。

  1. クライアントタイムアウト: ブラウザやロードバランサーが通信を切断してしまう。
  2. API サーバーの枯渇: サーバーのリソース（スレッドやワーカー）が重い処理に占有され、他のユーザーの簡単なアクセスすら捌けなくなる。

したがって、「リクエストを受け付けたら即座に応答を返し、重い処理は裏（バックグラウンド）で実行する」という設計が基本原則となります。

5.2 FastAPI 標準の BackgroundTasks

FastAPI には、標準機能として BackgroundTasks が用意されています。

レスポンスをクライアントに返却した直後に、同じサーバープロセスの裏で関数を実行させることができます。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification_log(email: str, message: str):
    with open("notifications.log", "a", encoding="utf-8") as f:
        f.write(f"Sent to {email}: {message}\n")


@app.post("/send-notification")
def send_notification(email: str, background_tasks: BackgroundTasks):
    # レスポンス返却後に実行するタスクを登録
    background_tasks.add_task(write_notification_log, email, message="ようこそ！")
    return {"status": "accepted", "message": "通知送信を受け付けました"}

追加のミドルウェアやインフラを用意する必要がなく、手軽に利用できるのが最大のメリットです。

5.3 BackgroundTasks の限界

非常に便利な BackgroundTasks ですが、実務の大規模システムやクリティカルな要件では以下のような重大な限界があります。

  • 同一プロセス・インメモリ実行: API サーバーと同じプロセス内で実行されるため、タスクが大量のメモリや CPU を消費すると API サーバー全体の動作が重くなります。
  • プロセスの再起動によるデータ消失: タスクの実行待ちキューはサーバーのメモリ上にしか存在しません。サーバーが再起動したりクラッシュしたりすると、待機中・実行中のタスクは永久に失われます。
  • リトライ機能の欠如: ネットワークエラー等でタスクが失敗した場合に、数分後に自動再試行する仕組みが標準では存在しません。
  • 複数サーバーへの分散不可: API サーバーを水平スケール（台数を増やす）しても、タスク処理を別台の専用ワーカーマシンに任せることができません。

5.4 分散タスクキュー（Celery）が必要になる判断基準

上記の問題を解決するのが、分散タスクキューである Celery です。

以下の条件に 1 つでも当てはまる場合は、BackgroundTasks ではなく Celery を選択するべきです。

  1. タスクのロストが許されない: 決済処理、注文確定後の在庫引き当て、重要通知など、サーバー再起動時でも確実に実行したい処理。
  2. 処理時間が長い（目安として 1 秒以上）: 音声・動画エンコード、大規模データの一括集計、機械学習モデルの推論。
  3. 自動リトライが必要: 外部 API のレート制限や一時的なネットワーク障害時に、指数バックオフで再試行したい場合。
  4. CPU / メモリ消費が激しい: API サーバーとワーカーサーバーをインフラレベルで物理的に分離してスケールさせたい場合。

5.5 機能比較まとめ

項目FastAPI BackgroundTasksCelery (+ Redis)
追加インフラ不要（Python のみ）必要（Redis または RabbitMQ）
実行場所API サーバーと同じプロセス独立した専用ワーカープロセス
キューの永続化なし（メモリ上、再起動で消失）あり（Broker に安全に保持）
自動リトライなし（自前実装が必要）標準装備（回数・間隔・例外指定可）
進捗・結果管理なし標準装備（タスク ID で状態照会可能）
適した用途簡易ログ記録、軽微な通知重い非同期バッチ、決済、画像・機械学習処理

第6章 Celery と Redis による非同期タスク基盤

Celery は、Python で最も広く利用されている実績豊富な分散タスクキューフレームワークです。

本章では、Celery のコアアーキテクチャ、Redis を使ったメッセージングの仕組み、タスクの定義方法、ワーカーの起動、および障害に強いリトライ設計を学びます。

6.1 Celery の基本アーキテクチャ

Celery は主に以下の 4 つの要素で構成されます。

  1. クライアント（Producer）: タスクの実行を依頼する側です（FastAPI アプリケーションなど）。
  2. メッセージブローカー（Broker）: タスクの指示書（メッセージ）を安全にキューイングして蓄積する中間システムです（Redis や RabbitMQ）。
  3. ワーカー（Consumer）: ブローカーからタスクを取り出してバックグラウンドで実際に処理を実行する Python プロセスです。
  4. 結果バックエンド（Result Backend）: タスクの実行結果や成否、進捗状態を保存しておくストレージです（Redis やデータベース）。
graph LR
    Client["FastAPI (Client)"] -->|"1. タスク発行"| Broker[("Redis (Broker)")]
    Broker -->|"2. タスク取得"| Worker["Celery Worker"]
    Worker -->|"3. 結果保存"| Backend[("Redis (Backend)")]
    Client -.->|"4. 結果問い合わせ"| Backend

6.2 Redis のセットアップ

Redis は、軽量・高速でセットアップが容易なため、Celery のメッセージブローカーおよび結果バックエンドとして最適です。

ローカル開発環境では、Docker を使用して数秒で起動できます。

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

6.3 Celery アプリケーションの初期化とタスク定義

Celery インスタンスを作成し、タスクとなる関数に @celery_app.task デコレータを付与します。

以下は、celery_tasks.py というファイルにタスクを定義する例です。

import time
from celery import Celery

# Redis への接続 URL を指定
REDIS_URL = "redis://localhost:6379/0"

# Celery アプリケーションの初期化
celery_app = Celery(
    "tasks",
    broker=REDIS_URL,
    backend=REDIS_URL,
)

# シリアライザやタイムゾーンの設定
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Tokyo",
    enable_utc=True,
)


# バックグラウンドで実行したいタスク
@celery_app.task(bind=True)
def long_running_task(self, x: int, y: int) -> int:
    print(f"タスク開始: ID={self.request.id}")
    time.sleep(5)  # 5秒かかる重い計算をシミュレート
    result = x + y
    print(f"タスク完了: 結果={result}")
    return result

bind=True を指定すると、関数の第 1 引数にタスク自身のインスタンス（self）が渡され、タスク ID の参照や進捗の更新が可能になります。

6.4 タスクの呼び出し（delay と apply_async）

定義したタスクを呼び出す方法は 2 種類あります。

  • task.delay(*args, **kwargs): 最もシンプルな呼び出し方です。引数を渡して即座にブローカーに投入します。
  • task.apply_async(args, kwargs, ...): 詳細な実行オプションを指定できます。カウントダウン（例: 60 秒後に実行）、優先度、有効期限、特定キューへのルーティングなどを指定できます。
# 1. シンプルな即時投入
result = long_running_task.delay(10, 20)
print(f"発行されたタスクID: {result.id}")

# 2. 30秒後に実行する遅延投入
result_delayed = long_running_task.apply_async(args=[10, 20], countdown=30)

どちらのメソッドも、呼び出した瞬間に AsyncResult オブジェクトを返します。この時点ではタスクはワーカーで処理中であり、待たされることはありません。

6.5 ワーカープロセスの起動

定義したタスクを実行するためには、ターミナルでワーカープロセスを起動します。

# celery_tasks.py の celery_app を指定してワーカーを起動
celery -A celery_tasks.celery_app worker --loglevel=info --concurrency=4
  • -A celery_tasks.celery_app: アプリケーションの場所を指定します。
  • --loglevel=info: ログの詳細度を指定します。
  • --concurrency=4: 同時に並行処理するタスク数（ワーカープロセス数）を指定します。

6.6 タスクの状態確認と自動リトライ

発行されたタスク ID を使って、結果バックエンドから現在の状態や戻り値を取得できます。

from celery.result import AsyncResult
from celery_tasks import celery_app

res = AsyncResult("発行されたタスクID", app=celery_app)

print("状態:", res.status)  # PENDING, STARTED, SUCCESS, FAILURE など
if res.ready():
    print("結果:", res.result)

また、ネットワークエラーや一時的な障害に備えて、自動リトライを宣言的に設定できます。

import httpx


@celery_app.task(
    bind=True,
    autoretry_for=(httpx.RequestError,),  # 通信エラー時に自動再試行
    retry_kwargs={"max_retries": 3, "countdown": 5},  # 最大3回、5秒間隔
    retry_backoff=True,  # 失敗するごとに間隔を延ばす（指数バックオフ）
)
def fetch_remote_webhook(self, url: str):
    res = httpx.post(url, json={"ping": "pong"}, timeout=5.0)
    res.raise_for_status()
    return res.status_code

第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 ワーカーにも導入し、未処理例外が発生した際にアラートを受け取れるようにします。