FastAPI×Celery/Redisバックグラウンド処理完全ガイド：重い処理をAPIから切り離してサービスを安定させる方法

greeden

4時間前

FastAPI×Celery/Redisバックグラウンド処理完全ガイド：重い処理をAPIから切り離してサービスを安定させる方法

はじめに（この記事でできるようになること）

本記事では、FastAPIを使ったWeb APIに「バックグラウンド処理」「バッチ的な非同期ジョブ」を組み込む方法を、丁寧に解説します。

ユーザーからのリクエストにはすばやくレスポンスを返しつつ、時間のかかる処理は裏側で進める
メール送信・レポート生成・画像加工・外部APIへの大量リクエストなどを、安全に非同期化する
Celery＋Redisを使って、「FastAPI本体」「ワーカー」「メッセージブローカー」の三つ組みを構成する

といったことを、サンプルコード付きで理解していきます。

誰が読んで得をするか（具体的な読者像）

個人開発・学習者さん

「登録完了メール」や「PDFレポート生成」など、数秒〜数分かかる処理をFastAPIから呼びたい
BackgroundTasks ぐらいは知っているけれど、プロセスを分けた本格的な非同期処理はまだ使ったことがない
RedisやCeleryの名前は聞いたことがあるけれど、「どう組み合わせるの？」がふわっとしている

→ この記事を読み進めることで、CeleryワーカーとRedisブローカーを使った「王道のバックグラウンド処理」の流れを、一通り体験できます。

小規模チームのバックエンドエンジニアさん

社内システムや小〜中規模SaaSをFastAPIで開発していて、「重い処理」や「再試行が大事な処理」をどう設計するか悩んでいる
同期APIのタイムアウトや、ピーク時のスローダウンが気になり始めている
チームで「ここから先はジョブキューに投げよう」という共通パターンを作りたい

→ FastAPIとCeleryの責務分離、リトライ・スケジューリングの基本パターンを押さえることで、運用しやすい構成のイメージが掴めます。

SaaS開発チーム・スタートアップの皆さま

画像処理・バッチ送信・データ集計など、スケールする処理が多いプロダクトを手がけている
マイクロサービス化やイベント駆動を見据えつつ、まずはシンプルなワーカー構成から始めたい
将来的には、ワーカーのスケールアウトやキューモニタリングも視野に入れている

→ 「FastAPI＝同期API」「Celery＝非同期ワーカー」「Redis＝キューブローカー」という役割分担をベースに、段階的にスケールさせていくための足場作りに役立つと思います。

アクセシビリティ評価（読みやすさの観点）

記事構成は、「全体像 → なぜバックグラウンド処理が必要か → Celery/Redisの役割 → 実装例 → エラーハンドリングとリトライ → スケジューリング → 運用のコツ → ロードマップ」という流れで整理しています。
専門用語（ブローカー、ワーカー、ジョブ、タスクなど）は、初出時に短く説明し、その後は同じ用語を繰り返すことで認知負荷を下げています。
コードブロックは1つあたりを短めに保ち、コメントも必要最低限にして、視覚的な負担が増えないようにしています。
読者が自分のペースで読み進められるよう、段落はやや短めに区切り、箇条書きも活用しています。

全体として、多様な利用者にとって読みやすく、理解のステップが追いやすいよう、WCAG AA相当のテキスト構造を意識しています。

1. なぜバックグラウンド処理が必要なのか

まずは、「そもそもなぜAPIから処理を切り離す必要があるのか」を整理しておきましょう。

1.1 APIに重い処理を直接書くと何が起きるか

たとえば、次のようなエンドポイントを考えてみます。

ユーザーがフォームを送信する
サーバ側でPDFレポートを生成（10秒）
外部メールサービスに送信依頼（＋ネットワーク待ち）
結果を返す

この流れを1つのHTTPリクエストの中で完結させると、

クライアント側のタイムアウトに引っかかる
Uvicornのワーカーが「重い処理」で塞がり、他のリクエストが待たされる
外部サービスの遅延や一時的なエラーに、ユーザー体験が直撃する

といった問題が発生しやすくなります。

1.2 バックグラウンド処理の基本的な考え方

そこで登場するのが「バックグラウンド処理」の発想です。

HTTPリクエストでは、「ジョブを登録しました」とだけすばやく返す
実際の重い処理は、別プロセス（ワーカー）がキューからジョブを受け取って実行する
結果や進捗は、後から別のAPI・通知・ダッシュボードなどで参照する

このように、

FastAPI本体＝入り口の受付係
Celeryワーカー＝裏側で作業するスタッフ
Redisなどのブローカー＝「やることリスト」を置く場所

と役割を分けることで、耐障害性とスケーラビリティを高めることができます。

2. CeleryとRedisの役割をざっくり理解する

ここからは、よく使われる組み合わせ「Celery＋Redis」について、ざっくりと役割を整理します。

2.1 Celeryとは

Celeryは、Python製の分散タスクキュー／ジョブキューです。

「タスク」と呼ばれる関数を、別プロセスで非同期に実行してくれる
メッセージブローカー（RedisやRabbitMQなど）を使って、タスクをキューに積んだり取り出したりする
リトライやスケジューリング（定期実行）、結果の保存（結果バックエンド）など、多くの機能を備えています

FastAPI公式ドキュメントでも、Celeryとの連携例が紹介されています。

2.2 Redisとは

Redisは、インメモリのデータストアであり、メッセージブローカーとしても使われます。

Celeryにとって、「タスクのキューを保存する場所」として機能
軽量かつ高速で、ローカル・コンテナ・クラウドなど様々な場所で動かせる

もちろんRabbitMQなど他のブローカーを使うこともできますが、Redisは導入がシンプルなので、まずはRedisから始めるのがおすすめです。

3. 開発環境の準備：FastAPI＋Celery＋Redisを動かす

ここからは、実際に動かせる形でセットアップしていきます。

3.1 必要なパッケージ

例として、次のようなパッケージを使います。

pip install "fastapi[standard]" celery[redis] redis

fastapi[standard]：FastAPI本体＋Uvicornなどの基本ツール
celery[redis]：Celery本体とRedisサポート
redis：Python用Redisクライアント

（バージョンはプロジェクトに合わせて調整してください）

3.2 Redisの起動（ローカル開発）

ローカル開発では、Dockerを使うと手軽です。

docker run -d --name redis -p 6379:6379 redis:7

これで、localhost:6379 でRedisに接続できるようになります。

4. FastAPI側とCelery側の最小構成を作る

次に、FastAPIアプリとCeleryアプリを定義していきます。

4.1 プロジェクト構成の例

project/
  app/
    __init__.py
    main.py          # FastAPI本体
    celery_app.py    # Celeryアプリ定義
    tasks.py         # バックグラウンドタスク
  requirements.txt

4.2 Celeryアプリの定義

# app/celery_app.py
from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",      # ブローカー（Redis）
    backend="redis://localhost:6379/1",     # 結果バックエンド（任意）
)

celery_app.conf.update(
    task_routes={
        "app.tasks.send_email": {"queue": "emails"},
        "app.tasks.generate_report": {"queue": "reports"},
    },
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
)

ここでは単純に localhost を使っていますが、本番では環境変数や設定クラスからURLを読み込むようにします。

4.3 タスクの定義

# app/tasks.py
from time import sleep
from app.celery_app import celery_app

@celery_app.task(name="app.tasks.send_email")
def send_email(to: str, subject: str, body: str) -> str:
    # 実際にはメール送信ロジック
    sleep(5)  # 処理が重いと仮定
    # 成功したらログなどを記録するイメージ
    return f"Email sent to {to}"

@celery_app.task(name="app.tasks.generate_report")
def generate_report(user_id: int) -> str:
    # 重い集計やPDF生成など
    sleep(10)
    return f"Report generated for user {user_id}"

ポイントは、

@celery_app.task デコレータを付けた関数が「タスク」として扱われる
send_email.delay(...) のように呼び出すと、即時ではなく「キューに登録」される

という点です。

5. FastAPIからタスクをキューに投げる

次に、FastAPI側のエンドポイントからタスクを登録してみます。

5.1 FastAPI本体の定義

# app/main.py
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from app.tasks import send_email, generate_report

app = FastAPI(title="FastAPI + Celery Example")

class EmailRequest(BaseModel):
    to: str
    subject: str
    body: str

@app.post("/emails")
def create_email(req: EmailRequest):
    # Celeryタスクを非同期でキューへ
    task = send_email.delay(req.to, req.subject, req.body)
    return {"task_id": task.id}

@app.post("/reports/{user_id}")
def create_report(user_id: int):
    task = generate_report.delay(user_id)
    return {"task_id": task.id}

HTTPリクエストに対しては、すぐにtask_idだけ返し, 実際の処理はCeleryワーカーに任せる形になっています。

5.2 タスクの結果を問い合わせるAPI（任意）

結果バックエンドを設定している場合（先ほどbackendにRedisを指定しました）、タスクの状態を問い合わせることもできます。

from app.celery_app import celery_app

@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    result = celery_app.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": result.status,      # PENDING / STARTED / SUCCESS / FAILURE など
        "result": result.result,      # 完了時は戻り値、失敗時は例外情報
    }

フロントエンドからは、

/emails や /reports/{user_id} でtask_idを取得
/tasks/{task_id} をポーリングして状態を確認

という流れで進捗を見られるようになります。

6. ワーカーの起動と動作確認

ここまでで、

FastAPIアプリ（HTTP入口）
Celeryワーカー（バックグラウンド処理本体）
Redis（キューブローカー）

が揃いました。次は実際に動かしてみましょう。

6.1 FastAPIの起動

uvicorn app.main:app --reload

6.2 Celeryワーカーの起動

プロジェクトルートで、次のコマンドを実行します。

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

-A には「Celeryアプリのインポートパス」を指定します。
ここでは app/celery_app.py に celery_app を定義しているので、このような形になります。

6.3 動作確認

別のターミナルから、

curl -X POST "http://127.0.0.1:8000/emails" \
  -H "Content-Type: application/json" \
  -d '{"to": "test@example.com", "subject": "Hello", "body": "Hi"}'

と呼ぶと、即座に

{"task_id": "xxxxxxxx-xxxx-...."}

のようなレスポンスが返り、ワーカー側のログに「Email sent to …」といったメッセージが表示されるはずです。

/tasks/{task_id} にGETリクエストを送れば、タスクの状態も確認できます。

7. 失敗時のリトライ・タイムアウトなどの制御

実務では、外部サービスの一時的な失敗やネットワークエラーが付きものです。
Celeryのタスクには、リトライやタイムアウトなどの設定を行うことができます。

7.1 自動リトライの例

# app/tasks.py（抜粋）
from celery import shared_task
import requests

@shared_task(
    bind=True,
    max_retries=5,
    default_retry_delay=10,  # 秒
)
def send_notification(self, endpoint: str, payload: dict) -> str:
    try:
        r = requests.post(endpoint, json=payload, timeout=5)
        r.raise_for_status()
    except requests.RequestException as exc:
        # self.retry でリトライをスケジュール
        raise self.retry(exc=exc)
    return "ok"

max_retries：最大リトライ回数
default_retry_delay：リトライ間隔（秒）
self.retry()：次のリトライをキューに登録します

7.2 タスクの実行時間制限

time_limit や soft_time_limit を使うと、タスクの最大実行時間も制御できます。

@celery_app.task(
    name="app.tasks.heavy_task",
    time_limit=60,       # 秒
    soft_time_limit=50,
)
def heavy_task():
    # 時間のかかる処理
    ...

時間制限を決めておくことで、暴走したタスクからワーカーを守り、他のタスクへの影響を減らすことができます。

8. 定期実行（スケジューリング）を行う

「毎日0時にレポートを生成」「5分ごとに集計タスクを走らせる」といった定期実行も、Celeryビート（スケジューラ）で実現できます。

8.1 Celeryビートの設定

# app/celery_app.py（ビート設定を追加）
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "generate-daily-reports": {
        "task": "app.tasks.generate_report",
        "schedule": crontab(hour=0, minute=0),
        "args": (1,),  # 例としてuser_id=1のレポート
    },
}

8.2 ビートの起動

ワーカーとは別に、スケジューラ用プロセスを起動します。

celery -A app.celery_app.celery_app beat --loglevel=info

この構成で、

beatプロセスがスケジュールに従ってタスクをキューに登録
workerプロセスがキューからタスクを取り出して実行

という流れが自動的に回り続けます。

9. FastAPIの`BackgroundTasks`との違いと使い分け

FastAPIにも BackgroundTasks という軽量なバックグラウンド機能があります。

from fastapi import BackgroundTasks

def send_email_sync(to: str):
    # 同期のメール送信処理
    ...

@app.post("/signup")
def signup(..., background_tasks: BackgroundTasks):
    # ユーザー登録処理...
    background_tasks.add_task(send_email_sync, user.email)
    return {"status": "ok"}

これはとても使いやすいのですが、

FastAPIの同一プロセス内で実行される
プロセスが落ちるとタスクも失われる
リトライやスケジューリングなどの高度な機能はない

という特徴があります。

一方、Celeryは

別プロセス（別コンテナ）でタスクを実行
キューに乗ったタスクは、ワーカーが一時的に死んでも残り続ける
リトライ・スケジュール・結果保存など、豊富な制御が可能

という違いがあります。

目安としては：

「ごく短時間で終わり、失敗しても痛手が少ない処理」→ BackgroundTasks
「重く、失敗時のリトライや確実な実行が大事な処理」→ Celeryなどのジョブキュー

といった使い分けを意識すると、設計が整理しやすくなります。

10. 運用を意識した設計のポイント

最後に、実務でCeleryを運用していく際に意識したいポイントを、いくつか挙げておきます。

10.1 ログと監視

タスクの成功・失敗ログは、FastAPIとは別に「ワーカーのログ」として集約する
タスクの失敗率や平均実行時間をメトリクスとして観測すると、ボトルネックの特定に役立つ
一部のタスクについては、「このエラーが3回連続で発生したらアラート」といった閾値を設定しておく

10.2 キューの設計

メール・レポート・集計など、性質の異なるタスクはキューを分ける（emails / reports / defaultなど）
キューごとにワーカーの数を変えることで、重要度の高いタスクにリソースを優先的に割り当てる
バックプレッシャーを意識し、あまりにタスクが溜まっている場合はAPI側で受付制限をすることも検討する

10.3 エラー時のユーザー体験

タスクが失敗したときに、ユーザーにどう伝えるか（通知・リトライ・お問い合わせ誘導など）
「処理が完了したらメールを送る」「ダッシュボードのステータスを更新する」といった設計を、API側と合わせて考える

11. 導入ロードマップ（少しずつ段階を踏むために）

一気にすべてを導入する必要はありません。段階的に進めるためのステップ案をまとめます。

BackgroundTasksで軽い処理から非同期化する
- まずはメール送信など、本当に軽いものをFastAPI内のバックグラウンド処理に移してみる。
Celery＋Redisをローカルで動かしてみる
- 本記事の「最小構成」を真似して、小さなPoCプロジェクトを作る。
1種類のタスクをCeleryに移行する
- レポート作成や重い集計など、時間のかかる処理を1つだけCeleryタスク化してみる。
タスク状態を問い合わせるAPIやダッシュボードを整える
- task_idベースで状態確認を行い、ユーザーに進捗や結果を見えるようにする。
リトライ・キュー分け・スケジューリングなどを追加していく
- サービスの成長に合わせて、必要なタスクにだけ少しずつ高度な機能を適用。
本番環境（Dockerやオーケストレーション）に展開する
- FastAPIコンテナ・Celeryワーカーコンテナ・Redisコンテナを分け、Docker ComposeやKubernetesで運用する。

12. まとめ

FastAPIだけで重い処理を抱え込むと、レスポンスの遅延やタイムアウト、全体のスループット低下につながりやすくなります。
Celery＋Redisを組み合わせることで、重い処理や失敗時にリトライしたい処理を別プロセスのワーカーに任せる構成を簡単に構築できます。
FastAPIは「タスクをキューに登録してすばやくレスポンスを返す窓口」として、Celeryは「裏側で淡々と処理をこなすエンジン」として役割分担するのが基本です。
BackgroundTasks は「軽い処理」、Celery は「より信頼性やスケールが必要な処理」と考え、段階的に使い分けていくと移行がスムーズです。
いきなり大規模なジョブキューシステムを目指す必要はありません。まずは1つのタスクから少しずつ切り出していくことで、無理なくサービスを「落ちにくく、重くなりにくい」構成へと育てていけます。

ここまで読んでくださって、ありがとうございます。
あなたのFastAPIアプリの裏側で、静かにたくさんの仕事をこなしてくれる「頼れるワーカーさんたち」が育っていきますように。
どうぞ、あせらず一歩ずつ、試してみてくださいね。

FastAPI×Celery/Redisバックグラウンド処理完全ガイド：重い処理をAPIから切り離してサービスを安定させる方法

はじめに（この記事でできるようになること）

誰が読んで得をするか（具体的な読者像）

個人開発・学習者さん

小規模チームのバックエンドエンジニアさん

SaaS開発チーム・スタートアップの皆さま

アクセシビリティ評価（読みやすさの観点）

1. なぜバックグラウンド処理が必要なのか

1.1 APIに重い処理を直接書くと何が起きるか

1.2 バックグラウンド処理の基本的な考え方

2. CeleryとRedisの役割をざっくり理解する

2.1 Celeryとは

2.2 Redisとは

3. 開発環境の準備：FastAPI＋Celery＋Redisを動かす

3.1 必要なパッケージ

3.2 Redisの起動（ローカル開発）

4. FastAPI側とCelery側の最小構成を作る

4.1 プロジェクト構成の例

4.2 Celeryアプリの定義

4.3 タスクの定義

5. FastAPIからタスクをキューに投げる

5.1 FastAPI本体の定義

5.2 タスクの結果を問い合わせるAPI（任意）

6. ワーカーの起動と動作確認

6.1 FastAPIの起動

6.2 Celeryワーカーの起動

6.3 動作確認

7. 失敗時のリトライ・タイムアウトなどの制御

7.1 自動リトライの例

7.2 タスクの実行時間制限

8. 定期実行（スケジューリング）を行う

8.1 Celeryビートの設定

8.2 ビートの起動

9. FastAPIのBackgroundTasksとの違いと使い分け

10. 運用を意識した設計のポイント

10.1 ログと監視

10.2 キューの設計

10.3 エラー時のユーザー体験

11. 導入ロードマップ（少しずつ段階を踏むために）

12. まとめ

共有:

9. FastAPIの`BackgroundTasks`との違いと使い分け