Python asyncioとFastAPIによる高並行Webサービス設計：ブロッキングIO事故を防ぐ鉄則

FastAPIの普及により、Pythonでも非同期処理（async/await）を活用した高スループットなAPIサーバーを簡単に構築できるようになった。しかし、非同期の仕組みを正しく理解しないまま「とりあえず関数にasyncを付ける」開発を行うと、たった1つの重い同期処理がイベントループ全体を停止させ、全リクエストが詰まる大事故を引き起こす。本稿では、FastAPIにおけるブロッキングIOの分離パターンと高並行設計の鉄則を解説する。

陥りがちな罠：async def 内でのブロッキング呼び出し

# 🚨 危険なアンチパターン：全ユーザーのリクエストがここで停止する！
@app.get("/data")
async def get_data():
    time.sleep(2) # 同期ブロッキング関数（requests.get()なども同様）
    return {"status": "ok"}

async関数の中でブロッキング関数を実行すると、シングルスレッドで動くイベントループが完全にブロックされ、並行処理の恩恵がゼロになる。

正しい対処法：FastAPIの自動スレッドプール委譲

同期ブロッキングな関数（外部ライブラリなど）を呼び出す場合は、通常の `def` でエンドポイントを定義するのがFastAPIの基本だ。FastAPIは `def` で定義されたエンドポイントを自動的に外部スレッドプール（`anyio.to_thread`）へオフロードして並行実行してくれる。

# ✅ 正しいアプローチ：FastAPIが別スレッドで安全に並行処理してくれる
@app.get("/data")
def get_data():
    time.sleep(2)
    return {"status": "ok"}

どうしてもasync関数内から同期関数を呼び出す必要がある場合は、明示的にスレッドプールへ逃がす：

import asyncio

@app.get("/heavy")
async def heavy_task():
    result = await asyncio.to_thread(sync_heavy_computation)
    return result

まとめ

非同期フレームワークの真価を引き出すには、I/Oバウンドな非同期ライブラリ（`httpx`, `asyncpg`, `aiofiles`）の採用と、CPUバウンド・同期処理の適切なスレッドプール分離が不可欠だ。