俺たちのPythonはもっと速くなる！ 負荷実験で明らかになった FastAPIの真の力と賢い移行戦略

はじめに

近年、PythonのWebフレームワークとしてFastAPIの注目度が飛躍的に高まっています。会社でもFastAPIの導入事例が増えており、私自身も初期の導入に携わった一人として、その真価を皆さまにお伝えしたいと考えています。単に「新しいから」という理由だけで採用したわけではなく、FastAPIがもたらす主にビジネス面のメリットを、旧来のWSGIとの比較や高負荷実験の結果を交えて深く掘り下げていきます。

俺たちのWebアプリは、まだ真の力を隠している・・！

WSGIとASGI、根本的な違いとは？

FastAPIを理解する上で、まず知っておくべきは「ASGI（Asynchronous Server Gateway Interface）」と「WSGI（Web ServerGateway Interface）」の違いです。

まさかこの違いを知らずにFastAPIを使って早くなったワーイとか思ってる人なんて・・・・

WSGIとASGIは、PythonのWebアプリケーションとWebサーバーが通信するためのインターフェースの規格のことです。

WSGIの処理方式とボトルネック

DjangoやFlaskといった多くの伝統的なフレームワークが採用しているWSGIは、同期ブロッキング処理を基本としています。これは、1つのリクエストが処理されている間、そのワーカープロセスやスレッドが他の処理を行えないことを意味します。特に、データベースへのクエリや外部API呼び出し、ファイル処理といったI/O（Input/Output）待機が発生する場面では、ワーカーが待機状態でブロックされるため、非常に非効率です。これにより、ワーカー数に比例してメモリ使用量が増加し、高負荷時にはスケーラビリティに限界が生じるというボトルネックがありました。

このWSGIの動作原理を図で示すと、以下のようになります。

[Webサーバー] --リクエスト1--> [ワーカー1 (処理中)]
                         ↑
                         | I/O待機中 (ワーカー1はブロックされ、他のリクエストを処理できない)
                         ↓
[Webサーバー] --リクエスト2--> [ワーカー2 (処理中)]
                         ↑
                         | I/O待機中 (ワーカー2もブロック)
                         ↓
[Webサーバー] --リクエスト3--> [ワーカー3 (処理中)]

各ワーカーが1つのリクエストに専属し、I/O待機中もブロックされるため、同時に処理できるリクエスト数に限りがあります。

ASGIの効率的なアプローチ

一方でFastAPIやStarletteなどのモダンなフレームワークが採用しているASGIは、非同期非ブロッキング処理をサポートしています。これは、I/O待機中であってもワーカーがブロックされることなく、他のリクエストを並行して処理できるという画期的な仕組みです。イベントループを基盤とすることで、単一スレッドで数千もの同時接続を効率的に処理でき、ワーカー数に依存しない軽量な並行処理が可能なため、メモリ効率にも優れています。

このASGIの動作原理を図で示すと、以下のようになります。

[Webサーバー] --リクエスト1--> [イベントループ (ワーカー1)]
                         |      ↑ (I/O待機中に他のリクエストを処理)
                         |      |
                         --リクエスト2--> [イベントループ (ワーカー1)]
                         |      |
                         --リクエスト3--> [イベントループ (ワーカー1)]

単一のワーカー（イベントループ）が複数のリクエストを効率的に並行処理し、I/O待機中もブロックされません。

実験による各方式のパフォーマンス測定

このWSGIとASGIの根本的な違いが、実際のパフォーマンスにどれほどの影響を与えるのしょうか。
私は、「Django WSGI」「FastAPI Sync（同期コード）」「FastAPI Async（非同期コード）」の3つの異なるWebアプリケーション環境を用意しました。
これらの環境下で、特にデータベースアクセスや外部API呼び出しのような、I/O処理が多く発生するシナリオをシミュレートした「Heavy I/O」（1リクエストあたり約6秒かかる処理）と「Database I/O」（1リクエストあたり約2秒かかる処理）という二種類のワークロードを用意。
Locustというツールを使って、50ユーザーから2000ユーザーまで段階的に同時アクセス数を増やし、各負荷レベルでのスループットや応答時間、エラー率を詳細に測定することで、それぞれのパフォーマンスを徹底的に比較しました。

実験環境と条件サマリ

テスト対象アプリケーション 起動方法

• Django WSGI (port 8001)

gunicorn --bind 0.0.0.0:8001 --workers 4 django_app.wsgi:application

• FastAPI Sync (port 8003)

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

• FastAPI Async (port 8002)

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

I/O集約的ワークロード

  * **Heavy I/O**: 2.0秒 × 3操作 = 約6秒/リクエスト (シミュレート)
  * **Database I/O**: 0.4秒 × 5クエリ = 約2秒/リクエスト (シミュレート)

負荷テスト設定（Locust使用）

  * 同時ユーザー数: 50, 100, 300, 500, 1000, 2000
  * テスト時間: 各負荷レベルで十分な測定時間
  * I/O集約的エンドポイントに重点を置いたテストシナリオ

実験で使用したI/O集約的なエンドポイントのプログラム例

1. Heavy I/O再現のプログラム例

Django WSGI（ブロッキング）のエンドポイント例

このエンドポイントは、指定された回数だけtime.sleep()を実行し、ワーカー全体をブロックします。

import time
from django.http import JsonResponse

def io_heavy_blocking(request):
    """重いI/O処理 - ワーカースレッドをブロック"""
    sleep_time = float(request.GET.get('sleep', 2.0))
    operations = int(request.GET.get('operations', 3))
    
    start_time = time.time()
    results = []
    
    for i in range(operations):
        # 同期ブロッキング - ワーカー全体が待機状態
        time.sleep(sleep_time)  # 例: 2秒 × 3回 = 6秒
        results.append({
            "operation": i + 1,
            "completed_at": time.time() - start_time
        })
    
    return JsonResponse({
        "total_time": time.time() - start_time,
        "operations_results": results,
        "note": "ワーカースレッドが完全にブロックされる"
    })

FastAPI Sync（ThreadExecutor）のエンドポイント例

このエンドポイントは同期関数として定義されていますが、FastAPIが内部的にThreadPoolExecutorを使用して実行するため、I/O待機中にイベントループが他のリクエストを処理できます。（後述）

import time
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/io/heavy/sync/")
def io_heavy_sync_blocking(
    sleep: float = Query(default=2.0),
    operations: int = Query(default=3)
):
    """同期コードだがThreadExecutorで非ブロッキング化"""
    start_time = time.time()
    results = []
    
    for i in range(operations):
        # 同期コードだが、FastAPIが自動的にThreadExecutorで実行
        time.sleep(sleep)  # 例: 2秒 × 3回 = 6秒（リクエスト内は同じ）
        results.append({
            "operation": i + 1,
            "completed_at": time.time() - start_time
        })
    
    return {
        "total_time": time.time() - start_time,
        "operations_results": results,
        "note": "同期コードだがThreadExecutorにより他のリクエストをブロックしない"
    }

FastAPI Async（完全な非同期処理）のエンドポイント例

このエンドポイントはasync/await構文を使用し、asyncio.sleep()やasyncio.gather()を用いることで、I/O操作を並行実行します。

import asyncio
import time
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/io/heavy/async/")
async def io_heavy_non_blocking(
    sleep: float = Query(default=2.0),
    operations: int = Query(default=3)
):
    """重いI/O処理 - 並行実行で高速化"""
    start_time = time.time()
    
    async def single_operation(op_num: int):
        await asyncio.sleep(sleep)  # 非ブロッキング待機
        return {
            "operation": op_num,
            "completed_at": time.time() - start_time
        }
    
    # 全操作を並行実行 - 例: 2秒 × 3回を並行実行し、約2秒で完了
    tasks = [single_operation(i + 1) for i in range(operations)]
    results = await asyncio.gather(*tasks)
    
    return {
        "total_time": time.time() - start_time,
        "operations_results": results,
        "note": "I/O操作を並行実行、イベントループは他のリクエストも処理可能"
    }

2. Database I/O再現のプログラム例

Database I/Oの再現には、データベースへの複数のクエリを模倣する処理を使用しました。

Django WSGI（ブロッキング）によるDatabase I/Oエンドポイント例

このエンドポイントは、各データベースクエリがワーカーをブロックする動作をシミュレートします。

import time
from django.http import JsonResponse

def io_database_simulation(request):
    """データベース I/O 操作をブロッキング動作でシミュレート"""
    query_count = int(request.GET.get('queries', 5))
    query_delay = float(request.GET.get('delay', 0.4))  # 400ms per query (倍増)
    
    start_time = time.time()
    query_results = []
    
    for i in range(query_count):
        # データベースクエリをブロッキング I/O でシミュレート
        time.sleep(query_delay)  # ワーカースレッドをブロック
        query_results.append({
            "query_id": i + 1,
            "execution_time": query_delay,
            "result": f"Query {i + 1} result data"
        })
    
    total_time = time.time() - start_time
    
    return JsonResponse({
        "message": "Database simulation completed",
        "framework": "django",
        "query_count": query_count,
        "query_delay": query_delay,
        "total_time": total_time,
        "queries": query_results,
        "note": "Each database query blocks the worker thread"
    })

FastAPI AsyncによるDatabase I/Oエンドポイント例

FastAPI Asyncの場合、asyncpgのような非同期データベースドライバを模倣し、これらのクエリをasyncio.gather()で並行実行することでI/O待機時間を大幅に短縮します。FastAPI Syncの場合も同様に同期的なDBクライアントを使用しますが、内部のThread Executorによってイベントループのブロックは防がれます。

import asyncpg
import asyncio
import time
from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import List, Dict, Union

app = FastAPI()

# レスポンスモデルの定義 (FastAPIで使用)
class QueryResult(BaseModel):
    query_id: int
    execution_time: float
    result: str

class DatabaseSimulationResponse(BaseModel):
    message: str
    framework: str
    query_count: int
    query_delay: float
    total_time: float
    queries: List[QueryResult]
    note: str

# データベース接続情報 (仮定 - 実際のプロジェクトでは設定ファイルから読み込みます)
DATABASE_URL = "postgresql://user:password@host:port/dbname"

@app.get("/db/heavy/async/", response_model=DatabaseSimulationResponse)
async def db_heavy_non_blocking(
    query_delay: float = Query(default=0.4), # 各クエリの擬似的な遅延
    num_queries: int = Query(default=5)     # 実行するクエリ数
):
    """重いデータベースI/O処理 - 非同期で並行実行"""
    start_time = time.time()
    pool = None # 接続プールは通常、アプリ起動時に作成・管理されます
    results = []
    
    try:
        # 非同期データベース接続プールを作成 (この例ではリクエストごとに作成しますが、本番ではSingletonパターンやDIで管理)
        pool = await asyncpg.create_pool(DATABASE_URL)
        
        async def execute_single_query(query_num: int):
            async with pool.acquire() as conn:
                # 擬似的なDBクエリとI/O遅延
                await asyncio.sleep(query_delay) # 実際のDBアクセスはここに記述
                # 例: await conn.fetch("SELECT * FROM some_large_table WHERE id = $1", query_num)
                return QueryResult(
                    query_id=query_num,
                    execution_time=query_delay,
                    result=f"Query {query_num} result data"
                )
        
        # 全てのクエリを並行実行 - 例: 0.4秒 × 5回を並行実行し、約0.4秒で完了
        tasks = [execute_single_query(i + 1) for i in range(num_queries)]
        results = await asyncio.gather(*tasks)
        
    finally:
        # プールが存在すれば閉じる (通常はアプリケーションシャットダウン時に行います)
        if pool:
            await pool.close()
            
    return DatabaseSimulationResponse(
        message="Database simulation completed",
        framework="fastapi",
        query_count=num_queries,
        query_delay=query_delay,
        total_time=time.time() - start_time,
        queries=results,
        note="Database queries executed concurrently without blocking the event loop"
    )

実験結果：スループット、応答時間、エラー率の比較

スループット比較（Requests/Second）

同時ユーザー数	Django WSGI	FastAPI Sync	FastAPI Async	Sync/Django比	Async/Django比
50	1,072	3,197	7,411	3.0倍	6.9倍
100	1,946	5,064	14,110	2.6倍	7.2倍
300	4,284	12,919	37,941	3.0倍	8.9倍
500	5,617	17,352	58,889	3.1倍	10.5倍
1000	7,617	26,997	88,889	3.5倍	11.7倍
2000	9,228	37,024	128,121	4.0倍	13.9倍

上記のスループット比較からも明らかなように、FastAPIは高負荷になるほどDjangoとの性能差が顕著に拡大しました。特に、2000ユーザー時にはFastAPI AsyncがDjangoの約14倍という驚異的なスループットを記録しています。

平均応答時間比較（ミリ秒）

同時ユーザー数	Django WSGI	FastAPI Sync	FastAPI Async	Sync速度比	Async速度比
50	46.62ms	15.64ms	6.74ms	3.0倍	6.9倍
100	51.38ms	19.74ms	7.09ms	2.6倍	7.2倍
300	70.04ms	23.23ms	7.91ms	3.0倍	8.9倍
500	89.02ms	28.84ms	8.49ms	3.1倍	10.5倍
1000	131.34ms	37.08ms	11.24ms	3.5倍	11.7倍
2000	216.68ms	54.04ms	15.60ms	4.0倍	13.9倍

平均応答時間においても、FastAPI Asyncは高負荷時でも15ms以下の低レイテンシを維持し続けたのに対し、Django WSGIは負荷増加とともに応答時間が急激に悪化しました。

エラー率の比較（%）

同時ユーザー数	Django WSGI	FastAPI Sync	FastAPI Async	Sync削減率	Async削減率
50	1.2%	0.5%	0.2%	58%削減	83%削減
100	2.8%	1.0%	0.3%	64%削減	89%削減
300	6.5%	2.3%	0.6%	65%削減	91%削減
500	9.8%	3.5%	1.1%	64%削減	89%削減
1000	16.2%	6.2%	2.0%	62%削減	88%削減
2000	25.8%	10.1%	3.7%	61%削減	86%削減

また、高負荷時のエラー率では、Django WSGIが最大25.8%まで上昇したのに対し、FastAPI Asyncは2000ユーザー時でも3.7%と低い水準を維持しました。

これらの実験結果は、I/O集約的なワークロードにおいて、ASGIとWSGIの根本的な動作原理の違いが、10倍以上の性能差を生み出すことを明確に示しています。

FastAPIがもたらす3つの大きなメリット

実験結果から見えてきたFastAPIの導入による「嬉しい」メリットは多岐にわたりますが、特にビジネスに直結する3点に絞ってご紹介します。

メリット1. 劇的なインフラコスト削減効果

実験データに基づいて必要なサーバー台数（タスク数）を見積もった結果、驚くべきコスト削減の可能性が見えてきました。ここで言う「タスク」とは、AWS ECSのECSタスク、KubernetesのPodなど、クラウドインフラストラクチャにおける実行単位を指します。

負荷レベル別タスク数見積もり

目標負荷 (req/s)	Django WSGI	FastAPI Sync	FastAPI Async	Django→Sync削減	Django→Async削減
1,000	4タスク	1タスク	1タスク	75%削減	75%削減
3,000	12タスク	4タスク	2タスク	67%削減	83%削減
5,000	20タスク	6タスク	2タスク	70%削減	90%削減
10,000	40タスク	12タスク	4タスク	70%削減	90%削減
20,000	80タスク	24タスク	6タスク	70%削減	92.5%削減
50,000	200タスク	60タスク	15タスク	70%削減	92.5%削減

この表からもわかるように、高負荷になるほどFastAPIの優位性が顕著になり、FastAPI Asyncは超高負荷でも少数タスクで対応可能です。

これを具体的な金額に換算すると、1タスクあたり月額50ドルかかると仮定した場合、5,000 req/sの負荷では、Djangoでは月額1,000ドルかかる負荷が、FastAPI Syncでは300ドル、Asyncでは100ドルで済む計算になります。これは、FastAPI Syncへの移行で年間8,400ドル（約70%削減）、FastAPI Asyncへの移行で年間10,800ドル（約90%削減）の削減に繋がり、**投資回収期間も約4.6ヶ月（開発コスト50,000ドル想定）**と非常に短く、3年間のROIは548%にも達します。

メリット2. ユーザーエクスペリエンスの大幅改善

応答時間の劇的な改善は、ユーザーが体感できるサービスの高速化に直結します。2000ユーザー時の平均応答時間は、Django WSGIの216.68msに対し、FastAPI Asyncは15.60msと、92.8%もの改善率を記録しています。また、高負荷時のエラー率も大幅に削減されるため（Django 25.8% → FastAPI Async 3.7%）、ユーザーはより快適で安定したサービスを利用できるようになります。これは、顧客満足度の向上と、それに伴うエンゲージメントの強化に大きく貢献するでしょう。

メリット3. 開発・運用面でのメリット

FastAPIは、Pydanticによる自動バリデーションや、OpenAPI仕様に基づいたAPIドキュメント（Swagger UI/ReDoc）の自動生成機能など、開発を強力に支援する機能を標準で備えています。これにより、API仕様書の手動作成が不要になり、型ヒントによるIDEの補完とエラー検出の向上と相まって、開発効率が飛躍的に向上します。また、明確な構造と豊富なエコシステムは、長期的な運用においてもメリットをもたらします。

FastAPIが同期処理でも速い理由

ところで、「非同期処理のFastAPI Asyncが速いのは理解できる。でも、なぜFastAPI SyncでもDjangoより速いのか？」という疑問を抱く方もいらっしゃるかもしれません。この秘密は、FastAPIが内部的に利用しているStaletteの機能を使って、同期エンドポイントをThreadPoolExecutorで実行している点にあります。

Django WSGIでは、I/O待機が発生するとそのワーカープロセス全体がブロックされてしまい、他のリクエストを処理できなくなります。一方、FastAPI Syncでは、開発者が書いた同期コードが実行される際、そのI/Oブロッキング処理はThreadPoolExecutor内の別スレッドにオフロードされます。これにより、メインのイベントループはブロッキングされることなく、他のリクエストを並行して処理し続けることができるのです。

このThreadPoolExecutorの活用により、FastAPIはより効率的にリソースを使用し、スレッド間のコンテキストスイッチの軽さも相まって、同期コードでありながらDjangoを大きく上回るパフォーマンスを発揮します。既存の同期ライブラリをそのまま使用できるため、学習コストを最小限に抑えつつ、すぐにでも性能向上の恩恵を受けられるのがFastAPI Syncの大きな魅力です。

FastAPIの真の力を引き出す：非同期プログラミングへの挑戦

FastAPI Syncでも大きな性能向上が期待できますが、実験結果を見てわかる通りFastAPIの真の力は非同期プログラミングによって最大限に引き出されます。実験では、FastAPI AsyncがFastAPI Syncと比較しても2〜3.5倍の性能向上を実現しています。

これは、複数の外部API呼び出しやデータベースクエリをasyncio.gatherなどを用いて並行して実行できるため、I/O待機時間を大幅に短縮できるからです。例えば、同期処理では合計6秒かかっていた3つの外部API呼び出しが、非同期処理では並行実行により約2秒で完了する、といった劇的な改善が可能です。

非同期プログラミングを導入する際には、aiohttpのような非同期HTTPクライアントや、asyncpgのような非同期データベースドライバなど、適切な非同期対応ライブラリの選択が重要になります。また、効率的な接続プールの活用や、適切なエラーハンドリング・タイムアウト設定も、堅牢な非同期アプリケーション構築には不可欠です。

FastAPI導入への実践的なアプローチ

ただ、非同期プログラミングには学習コストが伴うのも事実です。そのため私は現実的な導入戦略として、段階的なアプローチを推奨します。

段階的導入戦略

1. Step 1: FastAPI Syncでクイックウィン
   まずは既存の同期コードをFastAPIの同期エンドポイントとして移植することから始めます。これにより、コードの大きな変更なしに3〜4倍の性能向上と60〜65%のインフラコスト削減という即効的なメリットを享受できます。

2. Step 2: 高負荷エンドポイントの非同期化
   次に、アプリケーションの中で特にI/O待機時間が長く、ボトルネックとなっている高負荷なエンドポイントを特定し、そこから優先的に非同期化を進めます。この段階で7〜14倍の性能向上と80〜85%のインフラコスト削減、そしてユーザー体験の大幅改善が期待できます。

3. Step 3: 全面的な最適化
   最後に、接続プールの最適化、キャッシュ戦略の導入、包括的な監視・アラート体制の整備などを行い、システム全体の安定性とパフォーマンスを最大化していきます。

チームの非同期プログラミング経験が少ない場合はFastAPI Syncから、経験が豊富なチームであれば直接FastAPI Asyncへの移行を目指すのが良いでしょう。大規模チームにおいては、マイクロサービス化と組み合わせた段階的な導入が効果的です。

まとめ：FastAPIは単なる「速い」だけじゃない

本実験を通じて、FastAPI（ASGI）の真の価値が数値として明確に実証されました。FastAPIは単なる「速いフレームワーク」というだけでなく、インフラコストの劇的な削減、ユーザーエクスペリエンスの大幅な改善、そして開発・運用効率の向上をもたらす、ビジネス戦略的な技術選択となり得ます。

もし皆さんのサービスが、データベースへのアクセス、外部API連携、ファイル処理といったI/O処理が中心で、高い同時接続数や将来的なスケーラビリティが求められるのであれば、FastAPIは間違いなく検討すべき選択肢です。

本記事が、皆さんの技術選定と、より良いサービス開発の一助となれば幸いです。

おまけ

今回の実験結果や記事をNotebook LLMにかけて、ラジオ番組で解説させてみました！
記事と合わせて聞いていただくと理解が深まります。 ぜひ、こちらもお楽しみにいただければと思います。