FastAPIの非同期処理：BackgroundTasksとWebSocketの実装と運用ガイド

FastAPIで非同期処理を扱うときは、async/await、BackgroundTasks、WebSocketを同じ仕組みとして考えないことが出発点です。

async/awaitは待ち時間を別の処理に譲るための構文で、BackgroundTasksはレスポンス送信後に小さな処理を続ける機能です。

一方、WebSocketはクライアントとサーバーの接続を保ち、双方向にメッセージを送る通信方式です。

FastAPIの特徴とAPI設計の基本を理解している読者に向けて、三つの役割と実運用での境界をコード例で整理します。

三つの仕組みを使い分ける

選択基準は「いつ処理するか」と「クライアントとの接続を保つか」です。

asyncとawaitが解決する待ち時間

非同期処理とは、ネットワークやファイルなどの完了を待つあいだに、プログラムが別の処理を進められる実行方法です。

async defで定義した関数はコルーチンとなり、awaitに到達すると、待機中にほかの処理へ実行を譲れます。

import asyncio

async def fetch_summary(name: str, delay: int) -> str:
    await asyncio.sleep(delay)
    return f"{name} 完了"

async def main() -> None:
    results = await asyncio.gather(
        fetch_summary("task-a", 1),
        fetch_summary("task-b", 2),
    )
    print(results)

asyncio.run(main())

この例では、二つの待機を一つずつ終わらせてから次へ進むのではなく、asyncio.gather()が両方のコルーチンを並行して進めます。

ここでの並行処理は、二つの計算を複数のCPUで同時に実行する並列処理とは別です。

画像変換や大規模な数値計算のようにCPUを長く使う処理は、イベントループを占有しないように別プロセスやジョブ基盤へ分けます。

パス操作関数をdefとasync defのどちらで書くか

• awaitできるデータベースやHTTPクライアントを呼ぶ場合は、パス操作関数をasync defで定義します。
• 同期I/Oしか提供しないライブラリを使う場合は、パス操作関数を通常のdefで定義する選択肢があります。
• async defの中で長い同期処理を直接実行すると、その処理が終わるまでイベントループを塞ぐため避けます。

BackgroundTasksでレスポンス後の処理を分ける

BackgroundTasksは、レスポンスをクライアントへ送った後で、同じFastAPIアプリ内の関数を実行する機能です。

監査ログへの追記や小規模な通知など、クライアントが完了を待つ必要のない短い処理に向きます。

from fastapi import BackgroundTasks, FastAPI, status

app = FastAPI()

def write_audit_log(task_id: str) -> None:
    with open("task.log", "a", encoding="utf-8") as log_file:
        print(task_id, file=log_file)

@app.post("/process/{task_id}", status_code=status.HTTP_202_ACCEPTED)
async def process_data(
    task_id: str,
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    background_tasks.add_task(write_audit_log, task_id)
    return {"status": "accepted"}

このAPIが返すacceptedは、処理を受け付けたことを示します。

バックグラウンド処理の成功まで保証する応答ではないため、完了確認が必要な業務では状態を保存し、別のAPIから確認できる設計が必要です。

失敗時の扱いとタスクキューとの境界

BackgroundTasksはアプリ内で実行されるため、処理中にプロセスが停止すれば完了しない可能性があります。

複数のタスクは追加した順に実行され、途中のタスクが例外を出すと後続タスクは実行されません。

例外を記録するログと監視を用意し、応答済みのリクエストから失敗を追跡できるようにします。

長い計算、大量の処理、再試行、複数サーバーでの実行が必要なら、FastAPIとCelery、Redisでジョブ基盤を分ける方法を検討します。

WebSocketで双方向通信を続ける

WebSocketは、一度確立した接続を保ったまま、クライアントとサーバーのどちらからもメッセージを送れる通信方式です。

HTTPリクエストを繰り返さずに更新を届けられるため、チャットや処理状況の通知に使えます。

切断を処理するエコーサーバー

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"受信: {message}")
    except WebSocketDisconnect:
        # 接続一覧を管理する場合は、ここで削除する
        pass

accept()で接続を受け入れ、receive_text()とsend_text()で受信と送信を繰り返します。

クライアントが接続を閉じるとWebSocketDisconnectが発生するため、接続一覧や関連リソースの後始末をこのタイミングで行います。

ブラウザから接続する最小例

<form id="chat-form">
  <label for="message">メッセージ</label>
  <input id="message" autocomplete="off" required>
  <button type="submit">送信</button>
</form>
<div id="log" role="log" aria-live="polite"></div>

<script>
  const socket = new WebSocket("ws://127.0.0.1:8000/ws");
  const form = document.getElementById("chat-form");
  const input = document.getElementById("message");
  const log = document.getElementById("log");

  socket.addEventListener("message", (event) => {
    const line = document.createElement("p");
    line.textContent = event.data;
    log.appendChild(line);
  });

  form.addEventListener("submit", (event) => {
    event.preventDefault();
    if (socket.readyState !== WebSocket.OPEN) return;
    socket.send(input.value);
    input.value = "";
  });
</script>

入力欄にはラベルを関連付け、受信ログにはrole="log"とaria-live="polite"を指定しています。

これにより、画面へ追加されたメッセージを支援技術が穏やかに通知できます。

本番環境ではHTTPSに対応するwss://を使い、接続前の認証と権限確認、切断後の再接続方針も実装します。

複数クライアントへの配信やスケールアウトまで進める場合は、FastAPIとWebSocketの接続管理パターンも参照してください。

本番運用で確認する項目

サンプルが動いても、失敗時の挙動を決めなければ運用できません。

• 接続制御：同時接続数とメッセージサイズの上限を負荷試験から決めます。Uvicornの--wsはWebSocketプロトコル実装を選ぶ設定で、接続数を制限する設定ではありません。
• 切断と再接続：サーバーはWebSocketDisconnectで後始末を行い、クライアントは再接続の間隔と上限を決めます。
• 認証と認可：接続時に利用者を確認し、接続後も送受信できるデータの範囲を制限します。
• バックグラウンド処理の可観測性：成功と失敗、処理時間を記録し、応答後に発生した問題を追跡できるようにします。
• 複数ワーカー：プロセス内の接続一覧は別ワーカーと共有されません。複数プロセスから同じメッセージを配信する場合は、共有できるメッセージ基盤を用意します。

Uvicornでサーバー全体の同時接続またはタスク数を制限する設定は--limit-concurrencyです。

ただし、適切な値はアプリの処理内容、メモリ使用量、ワーカー数で変わるため、固定値を写すのではなく負荷試験から決めます。

実装を選ぶためのチェックリスト

• 待ち時間の中心がデータベースや外部APIなら、対応ライブラリとasync/awaitの組み合わせを確認する。
• レスポンス後の短い処理なら、失敗を追跡できる状態でBackgroundTasksを使う。
• 再試行や永続化が必要なら、APIプロセスからジョブキューへ処理を渡す。
• 双方向のライブ更新が必要なら、WebSocketの切断、再接続、認証を先に設計する。
• 公開前に、通常時だけでなくプロセス停止、通信断、過負荷時の挙動もテストする。

まとめ

async/awaitは入出力待ちを効率よく扱い、BackgroundTasksはレスポンス後の小さな処理を同じアプリ内で実行します。

WebSocketは双方向通信を続ける仕組みであり、バックグラウンド処理の代わりではありません。

処理の完了保証、再試行、切断、認証、監視まで要件に含めると、サンプルコードを本番で運用できる構成へ進められます。