FastAPIにおける非同期 vs 同期関数：どちらを選ぶべきか

はじめに

効率的でスケーラブルなWebアプリケーションを構築することは、開発者にとって絶え間ない追求です。Pythonエコシステムにおいて、FastAPIは、その非同期機能のおかげで、高性能APIの作成における強力なツールとして登場しました。しかし、FastAPIへの移行において、新参者や経験豊富な開発者にとって共通の混乱の種は、いつasync defを使用し、いつ従来のdef関数に留まるべきかを理解することです。この決定は単なるスタイル上の問題ではなく、アプリケーションの応答性、リソース利用、および全体的なパフォーマンスに大きな影響を与えます。この記事では、FastAPIにおける非同期関数定義と同期関数定義の違いを解明し、それぞれの使用時期についての明確なガイダンスを提供し、最終的に、より堅牢で効率的なWebサービスを構築できるようにします。

基本概念の理解

FastAPIの具体例に入る前に、Pythonにおける同期プログラミングと非同期プログラミングの基本的な概念を把握することが不可欠です。

同期関数 (def)

defを使用して関数を定義すると、それは同期関数と見なされます。これは、関数が呼び出されたときに、その操作を順番に、一つずつ実行することを意味します。同期関数が完了までに時間がかかる操作（データベースクエリ、外部API呼び出し、ファイルI/Oの待機など）に遭遇した場合、プログラム全体がその時点でブロックされ、次のコード行に進む前に操作が完了するのを待ちます。Webサーバーのコンテキストでは、これは、1つのリクエストがブロックI/O操作を実行している間、サーバーはその同じワーカープロセスで他の着信リクエストを処理できないことを意味します。

import time

def synchronous_task(task_id: int):
    print(f"Synchronous Task {task_id}: Starting CPU-bound work...")
    # CPUバウンドな操作をシミュレート
    count = 0
    for _ in range(1_000_000_000):
        count += 1
    print(f"Synchronous Task {task_id}: CPU-bound work finished.")
    
    print(f"Synchronous Task {task_id}: Starting I/O-bound wait...")
    # ブロッキングI/O操作をシミュレート
    time.sleep(2)  # これはスレッドをブロックします
    print(f"Synchronous Task {task_id}: I/O-bound wait finished.")
    return f"Result from synchronous task {task_id}"

# 実行すると、synchronous_task(1) は synchronous_task(2) が開始する前に完全に完了します。

非同期関数 (async def)

async defで定義された関数は非同期であり、メイン実行スレッドをブロックすることなく、並行して操作を実行するように設計されています。awaitキーワードは、async def関数内で重要です。async def関数が「awaitable」オブジェクト（asyncio.sleep、httpxを使用した非同期HTTPリクエスト、または非同期データベースドライバー呼び出しなど）のawait式に遭遇すると、その時点で実行を一時停止し、イベントループに制御を返します。イベントループは、実行準備ができている別のタスクに切り替えることができます。awaitされた操作が完了すると、async def関数は中断したところから実行を再開できます。このノンブロッキング動作は、I/Oバウンドタスクにとって特に有利です。

import asyncio

async def asynchronous_task(task_id: int):
    print(f"Asynchronous Task {task_id}: Starting I/O-bound wait...")
    # 非ブロッキングI/O操作をシミュレート
    await asyncio.sleep(2)  # これはイベントループに制御を譲ります
    print(f"Asynchronous Task {task_id}: I/O-bound wait finished.")
    
    print(f"Asynchronous Task {task_id}: Starting CPU-bound work...")
    # CPUバウンドな操作をシミュレート（オフロードしない限り、依然としてブロックします）
    count = 0
    for _ in range(1_000_000_000):
        count += 1
    print(f"Asynchronous Task {task_id}: CPU-bound work finished.")
    return f"Result from asynchronous task {task_id}"

# 非同期コンテキストでは、asynchronous_taskの複数の呼び出しは、await期間中に「並行して」実行される可能性があります。

FastAPIと関数実行

StarletteとPydantic上に構築されたFastAPIは、Pythonのasyncioライブラリを活用して非同期リクエスト処理を可能にします。これにより、比較的少数のワーカープロセスで高い同時実行性を達成できます。

FastAPIでのasync defの処理

FastAPIがリクエストを受信し、それをasync defエンドポイント関数にルーティングすると、イベントループ内で直接この関数を実行します。async def関数がI/Oバウンドなawait可能な操作に遭遇した場合、制御を譲り、イベントループが他の着信リクエストや実行準備ができている他のタスクの処理に切り替わることを可能にします。ここでasync defの真価が発揮されます。I/Oバウンドな操作（データベース呼び出し、外部API呼び出し、ファイル読み書き、ネットワークリクエスト）の場合、アプリケーションは、各待機操作に専用のプロセスを割り当てることなく、多数の同時クライアントを効率的に処理できます。

例：I/Oバウンド操作のためのasync def

外部サービスからデータを取得するAPIエンドポイントを考えてみましょう。

from fastapi import FastAPI
import httpx # 非同期HTTPクライアント
import asyncio

app = FastAPI()

@app.get("/items_async/{item_id}")
async def get_item_async(item_id: int):
    print(f"Request for item {item_id}: Starting external API call asynchronously...")
    async with httpx.AsyncClient() as client:
        # 時間のかかる外部API呼び出しをシミュレート
        response = await client.get(f"https://jsonplaceholder.typicode.com/todos/{item_id}")
        data = response.json()
    print(f"Request for item {item_id}: External API call finished.")
    return {"item_id": item_id, "data": data}

# これをテストするには、/items_async/1、/items_async/2などに複数の同時リクエストを作成できます。
# それらが厳密に一つずつではなく、インターリーブされた方法で完了するのを観察できるでしょう。

この例では、await client.get(...)は、メインイベントループをブロックすることなくget_item_asyncの実行を一時停止します。FastAPIはその後、他の着信リクエストを処理したり、他のタスクを実行したりできます。

FastAPIでのdefの処理

FastAPIがdefエンドポイント関数を検出すると、それを同期関数として賢く認識します。同期関数がメインの非同期イベントループをブロックするのを防ぐために、FastAPIは自動的に同期エンドポイント関数を別のスレッドプールで実行します。これは、def関数がブロックI/O操作や長時間のCPUバウンド計算を実行する場合、このスレッドプールからスレッドをブロックすることになりますが、メインイベントループ自体はブロックしないことを意味します。

例：同期、潜在的にブロックする操作のためのdef

非同期化が容易ではない複雑でCPU集約的な計算を実行するエンドポイントを想像してみてください。

from fastapi import FastAPI
import time

app = FastAPI()

def perform_heavy_computation(number: int):
    print(f"Synchronous Computation for {number}: Starting CPU-bound work...")
    # CPUバウンドな操作をシミュレート
    result = 0
    for i in range(number * 10_000_000): # 大きなループ
        result += i
    print(f"Synchronous Computation for {number}: CPU-bound work finished.")
    return result

@app.get("/compute_sync/{number}")
def compute_sync(number: int):
    print(f"Request for computation {number}: Received.")
    computation_result = perform_heavy_computation(number)
    return {"input_number": number, "result": computation_result}

# 複数のリクエストが同時に/compute_syncにヒットした場合、それぞれがFastAPIのスレッドプールから別々のスレッドで実行されます。
# 同期操作の同時実行数は、このスレッドプールのサイズによって制限されます。

この場合、perform_heavy_computationはブロッキング関数です。FastAPIはこれをバックグラウンドスレッドで実行し、メインイベントループをブロックするのを防ぎます。ただし、そのような同時ブロッキング操作の数は、スレッドプールのサイズ（デフォルトはuvicornで最大40スレッド程度）によって制限され、スレッドの作成と管理にはオーバーヘッドが発生します。

async defとdefの使い分け

async defとdefの選択は、主にエンドポイントが実行する操作の性質にかかっています。

async defを使用する場合：

• **関数がawait可能なI/Oバウンドな操作を含む場合。**これが主なユースケースです。例：
  • httpxを使用して外部APIへのHTTPリクエストを行う。
  • 非同期データベースドライバー（例：PostgreSQL用のasyncpg、aioodbc、asyncioを使用したSQLModel）と対話する。
  • 非同期でファイルを読み書きする（例：aiofiles）。
  • 非同期キューからメッセージを待機する。
  • CPU使用率が高くない外部リソースを待機するあらゆる操作。
• **他のawait可能なユーティリティやライブラリを活用する必要がある場合。**本質的に非同期であるライブラリと統合している場合、それらの操作をawaitするにはasync defが必要です。
• I/Oバウンドタスクの同時実行性を最大化したい場合。async defを使用すると、リクエストがI/O待機に大部分の時間を費やす限り、アプリケーションは大量の同時リクエストを効率的に処理できます。

経験則： 関数にawaitキーワードが含まれている場合、それはasync defである必要があります。

defを使用する場合：

• **関数が純粋にCPUバウンドな操作を実行する場合。**関数が計算の実行、メモリ内データの処理、または外部リソースを待機せずに長時間ループすることに大部分の時間を使っている場合、それはCPUバウンドです。それをasync defにしても、CPU計算がノンブロッキングになるという魔法のような効果はありません。実際の計算は、イベントループをブロックし続けるか（awaitしない場合）、または（FastAPIがdef関数をオフロードする場合）バックグラウンドスレッドをブロックします。
  • 例：複雑な数学的計算、重いデータ変換、画像処理、ビデオエンコーディング。
• **同期専用ライブラリやドライバーと対話する場合。**多くの古いPythonライブラリ、特にデータベースドライバー（PostgreSQL用のpsycopg2や従来のSQLAlchemy ORMなど）は同期です。エンドポイントがこれらを使用する必要がある場合、defとして定義することで、FastAPIはスレッドプールで実行してブロッキングの性質を処理できます。
• **シンプルさと親しみやすさ。**I/Oや複雑なロジックを含まない非常にシンプルなエンドポイントの場合、特にasyncioのベストプラクティスに精通していない場合、def関数の方が記述や理解がわずかに簡単かもしれません。ただし、常に将来のスケーラビリティを考慮してください。

CPUバウンドなasync defからの重要な注意点： async def関数に、何も積極的にawaitしないCPUバウンドなコード片が含まれている場合、そのCPUバウンドなコードはイベントループをブロックし続けます。イベントループをブロックすることなくasync defエンドポイント内でCPUバウンドな作業を処理するには、通常、loop.run_in_executor()（またはそれに基づくライブラリ、例えばstarlette.concurrency.run_in_threadpool）を使用して、別のプロセスまたはスレッドプールにオフロードする必要があります。FastAPIはdef関数に対してこれを自動的に行いますが、async def関数では、長時間実行されるCPUコードがある場合は明示的に管理する必要があります。

from concurrent.futures import ThreadPoolExecutor
from functools import partial

# ... (app = FastAPI() と上記の perform_heavy_computation 関数を想定)

executor = ThreadPoolExecutor(max_workers=4) # CPUバウンドタスク用スレッドプール

@app.get("/compute_async_offloaded/{number}")
async def compute_async_offloaded(number: int):
    print(f"Request for computation {number}: Received, offloading CPU work...")
    # 集中的なCPU計算をスレッドプールにオフロードする
    loop = asyncio.get_event_loop()
    computation_result = await loop.run_in_executor(
        executor, partial(perform_heavy_computation, number)
    )
    return {"input_number": number, "result": computation_result}

これはより高度なパターンであり、場合によってはasync def関数でもCPUバウンドな作業を明示的に管理する必要があることを示しています。

結論

FastAPIでasync defとdefのどちらを選択するかは、アプリケーションのパフォーマンス特性に影響を与える重要な決定です。I/Oバウンドなタスクの場合、async defはほぼ常に優れた選択肢であり、Pythonのasyncioイベントループを活用することで、高い同時実行性と効率的なリソース利用を可能にします。逆に、CPUバウンドな操作や同期ライブラリとの対話の場合、defが適切であり、FastAPIはこれをインテリジェントにスレッドプールにオフロードして、メインイベントループのブロッキングを防ぎます。根本的な仕組みを理解し、操作の性質を考慮することで、両方のパラダイムを効果的に活用して、高性能でスケーラブルなFastAPIアプリケーションを構築できます。迷った場合は、関数のいずれかの部分がI/Oバウンドであり、非同期にawaitできる場合は、async defを優先してください。