FastAPI入門 — なぜ次世代Python Webフレームワークなのか
Flaskが多機能なスイスアーミーナイフで, Djangoがフル装備のSUVだとすれば, FastAPIは電気スーパーカーです - 加速が速く, 省エネで, 自動生成ダッシュボードまで付いています。
1. 学ぶ内容
- ASGIとWSGIの根本的な違い:なぜ非同期が未来なのか
- FastAPI vs. Flask vs. Django REST Framework:性能と開発体験の比較
- 型ヒントが自動バリデーションとドキュメント生成をどう驱动するか
- StarletteとPydanticの基盤アーキテクチャの深掘り
- PriceTrackerプロジェクト概要:Aliceが何を構築しようとしているか
2. Aliceの実際のストーリー
(1) 悩み:同期フレームワークでは数百万リクエストに対応できない
Aliceはバックエンドエンジニアで, PriceTracker - eコマースプラットフォーム向けのSaaS価格追跡API - を構築しています。数百万の商品価格のリアルタイムクエリを処理する必要があります。最初はFlaskでプロトタイプを構築しましたが, 同時リクエストが1,000 QPSを超えると, 同期WSGIモデルのせいで各リクエストが順番待ちとなり, P99レイテンシが3,000 msに急増しました。Bob (フロントエンドエンジニア)はページの読み込みが遅いと不満を言い, Charlie (DevOps)は水平スケーリングのコストが高すぎると言いました。
(2) FastAPIによる解決策
FastAPIはASGI非同期プロトコルに基づいており, 単一プロセスで数千の同時接続を処理できます。型ヒントがOpenAPIドキュメントを自動生成し, リクエストデータをバリデーションするため, Aliceはバリデーションコードやドキュメントを手作業で書く必要がありません。
from fastapi import FastAPI
app = FastAPI()
@app.get("/products/{product_id}")
async def get_product(product_id: int):
# 型ヒントが自動的にバリデーションとドキュメント生成を行う
return {"product_id": product_id, "name": "Widget"}
(3) 成果
FastAPIへの移行後, PriceTrackerの単一ノードQPSは500から4,000以上に向上し, P99レイテンシは50 msに低下, Bobのフロントエンドページは5倍速く読み込まれるようになり, Charlieのサーバーコストは60%削減されました。
3. ASGIとWSGI:非同期が未来
(1) WSGIの同期ボトルネック
WSGI (Web Server Gateway Interface)はPython Webアプリケーションの従来の標準です。各リクエストがスレッドを占有し, I/O操作 (データベースクエリやネットワークリクエストなど)が発生するとスレッドがブロックされて待機します。
flowchart LR
Client1[クライアント1] -->|リクエスト| WSGI[WSGIサーバー]
Client2[クライアント2] -->|リクエスト| WSGI
Client3[クライアント3] -->|リクエスト| WSGI
WSGI -->|スレッド1| DB1[(データベース)]
WSGI -->|スレッド2| DB1
WSGI -->|スレッド3 - ブロック| DB1
| 次元 | WSGI | ASGI |
|---|---|---|
| 接続モデル | 1リクエスト1スレッド | 非同期コルーチン, 単一スレッドで複数接続 |
| 同時接続制限 | スレッドプールに制限 (通常10〜100) | ほぼ無制限 (コルーチンは軽量) |
| I/O待機 | スレッドをブロック | 非ブロック, 別のコルーチンに切り替え |
| WebSocket | 非対応 | ネイティブ対応 |
| 代表的なサーバー | Gunicorn + Flask | Uvicorn + FastAPI |
(2) ASGIの非同期の利点
ASGI (Asynchronous Server Gateway Interface)はWSGIの非同期拡張であり, async/await構文をサポートし, 単一プロセスで数千の同時接続を処理できます。
import asyncio
import time
# WSGIスタイル - スレッドをブロック
def sync_handler():
time.sleep(1) # スレッドが1秒間ブロックされる
return "done"
# ASGIスタイル - 非ブロック
async def async_handler():
await asyncio.sleep(1) # イベントループが他のタスクに切り替え
return "done"
(1) ▶ サンプル:同期と非同期の同時実行の比較
import asyncio
import time
async def fetch_price(product_id: int) -> dict:
# データベースI/Oレイテンシをシミュレーション
await asyncio.sleep(0.1)
return {"product_id": product_id, "price": 9.99}
async def main():
start = time.perf_counter()
# 100件の同時リクエスト - 非同期は約0.1秒で完了
results = await asyncio.gather(*[fetch_price(i) for i in range(100)])
elapsed = time.perf_counter() - start
print(f"Async: {len(results)} items in {elapsed:.2f}s")
asyncio.run(main())
出力:
実行成功
出力:
Async: 100 items in 0.10s
4. FastAPI vs. Flask vs. Django DRF
(1) フレームワークエコシステム内の位置づけ
flowchart LR
FastAPI[FastAPI] --> Starlette[Starlette ASGI]
Starlette --> Uvicorn[Uvicornサーバー]
Uvicorn --> ASGI_Protocol[ASGIプロトコル]
FastAPI --> Pydantic[Pydantic V2]
Flask2[Flask] --> Werkzeug[Werkzeug WSGI]
Werkzeug --> Gunicorn[Gunicorn]
Django2[Django DRF] --> Django_Core[Djangoコア]
| 次元 | FastAPI | Flask | Django DRF |
|---|---|---|---|
| 性能 (TechEmpower RPS) | ~40,000 | ~1,200 | ~800 |
| 非同期対応 | ネイティブasync/await | 拡張が必要 | 限定的な対応 |
| 自動ドキュメント | OpenAPI自動生成 | Flask-RESTXが必要 | drf-spectacularが必要 |
| 型バリデーション | Pydantic自動バリデーション | 手動バリデーション | 手動シリアライザ定義 |
| 学習曲線 | 低い (型ヒントがドキュメントに) | 低い | 高い |
| プロジェクト規模 | 小〜中規模APIサービス | 小規模サービス | 大規模フルスタックプロジェクト |
(2) なぜPriceTrackerにFastAPIが適しているか
| PriceTrackerの要件 | FastAPIの利点 | Flaskの欠点 |
|---|---|---|
| 数百万QPSのクエリ | 非同期コルーチンによる高同時接続 | 同期ブロックによる低同時接続 |
| WebSocketによるリアルタイム価格配信 | ネイティブ対応 | 非対応 |
| Bob向けの自動APIドキュメント | OpenAPI自動生成 | 追加設定が必要 |
| リクエストデータバリデーション | Pydantic自動 | カスタムバリデーションデコレータ |
| JWT認証 | 内蔵OAuth2ツール | サードパーティライブラリが必要 |
(1) ▶ サンプル:同じAPIを3つの方法で書く比較
# === FastAPI: 型ヒント = 自動バリデーション + ドキュメント ===
from fastapi import FastAPI
from pydantic import BaseModel
class Product(BaseModel):
name: str
price: float
app = FastAPI()
@app.post("/products")
async def create_product(product: Product):
return product # 自動バリデーション, 自動ドキュメント生成
出力:
# 関数定義成功
# === Flask: 手動バリデーション, 自動ドキュメントなし ===
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/products", methods=["POST"])
def create_product():
data = request.get_json()
if not data or "name" not in data or "price" not in data:
return jsonify({"error": "Invalid data"}), 400
if not isinstance(data["price"], (int, float)):
return jsonify({"error": "Price must be number"}), 400
return jsonify(data)
5. 型ヒントがすべてを驱动する
(1) 型ヒントの三重の価値
FastAPIはPython型ヒントを使って3つのことを同時に実現します:データバリデーション, シリアライズ/デシリアライズ, OpenAPIドキュメント生成です。
| 型ヒント | 従来のフレームワーク | FastAPI |
|---|---|---|
| データバリデーション | 手書きのif/else | Pydantic自動 |
| JSONシリアライズ | 手動json.dumps |
model_dump()自動 |
| APIドキュメント | 手書きのSwagger YAML | OpenAPI自動生成 |
| IDE自動補完 | なし | 完全な型推論 |
(1) ▶ サンプル:型ヒントの自動バリデーション
from fastapi import FastAPI, Query
from typing import Optional
app = FastAPI()
@app.get("/prices")
async def search_prices(
min_price: float = Query(0.0, ge=0, description="米ドルでの最低価格"),
max_price: float = Query(999999.0, le=999999, description="米ドルでの最高価格"),
category: Optional[str] = Query(None, max_length=50),
):
return {"min_price": min_price, "max_price": max_price, "category": category}
出力:
# 関数定義成功
(2) ▶ サンプル:自動生成されるOpenAPIドキュメント
# 上記のエンドポイントを定義した後, 以下にアクセス:
# http://localhost:8000/docs -> Swagger UI
# http://localhost:8000/redoc -> ReDoc
# http://localhost:8000/openapi.json -> 生のOpenAPIスキーマ
出力 (
/openapi.jsonの抜粋):
{
"paths": {
"/prices": {
"get": {
"summary": "Search Prices",
"parameters": [
{"name": "min_price", "in": "query", "schema": {"type": "number", "minimum": 0.0}}
]
}
}
}
}
(2) Pydanticの役割
Pydantic V2はFastAPIのデータバリデーションエンジンです。コアはRustで書き直されており, V1より5〜50倍高速です。
| 機能 | Pydantic V1 | Pydantic V2 |
|---|---|---|
| コアエンジン | Python | Rust (pydantic-core) |
| バリデーション速度 | ベンチマーク | 5〜50倍高速 |
| バリデータ | @validator |
@field_validator/@model_validator |
| 設定 | class Config |
model_config = ConfigDict(...) |
| シリアライズ | .dict() |
.model_dump() |
6. 基盤アーキテクチャ:Starlette + Pydantic
(1) FastAPIの3層アーキテクチャ
FastAPI自体は薄いラッパーであり, 核心的な機能はStarlette (ASGIフレームワーク)とPydantic (データバリデーション)から来ています。
| レベル | コンポーネント | 責務 |
|---|---|---|
| アプリケーション層 | FastAPI | ルート登録, 依存性注入, OpenAPI生成 |
| ASGI層 | Starlette | ミドルウェア, リクエスト/レスポンス処理, WebSocket |
| バリデーション層 | Pydantic | データバリデーション, シリアライズ, 型変換 |
| サービス層 | Uvicorn | ASGIサーバー, イベントループ管理 |
(1) ▶ サンプル:FastAPIでStarlette機能を直接使用する
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
app = FastAPI()
# StarletteミドルウェアはFastAPIとシームレスに連携
app.add_middleware(CORSMiddleware, allow_origins=["*"])
# Starletteレスポンスクラスも使用可能
@app.get("/health")
async def health_check():
return JSONResponse({"status": "healthy"})
出力:
# 関数定義成功
(2) ▶ サンプル:FastAPIでPydanticモデルを使用する
from fastapi import FastAPI
from pydantic import BaseModel, Field
class PriceCreate(BaseModel):
product_id: int = Field(gt=0)
price: float = Field(gt=0, description="米ドルでの価格")
currency: str = Field(default="USD", max_length=3)
app = FastAPI()
@app.post("/prices")
async def create_price(data: PriceCreate):
# dataはPydanticによって既にバリデーション・解析済み
validated = data.model_dump()
return {"status": "created", "data": validated}
出力:
# 関数定義成功
7. PriceTrackerプロジェクト概要
(1) Aliceは何を構築しようとしているか?
PriceTrackerはSaaS価格追跡APIです。核心的な機能は以下の通りです:
| 機能モジュール | APIエンドポイント | 説明 |
|---|---|---|
| 商品管理 | /products CRUD |
数百万商品レコード |
| 価格追跡 | /prices CRUD |
リアルタイム価格クエリと通知 |
| ユーザー認証 | /auth/login, /auth/register |
JWTダルトークン認証 |
| サブスクリプションプラン | Free/Pro/Enterprise | SaaSマルチテナント権限 |
| 一括インポート | /import/csv |
1,000行CSVファイルの非同期インポート |
| リアルタイムプッシュ通知 | WebSocket /ws/prices |
即時価格変動アラート |
flowchart LR
Bob[Bob - フロントエンド] -->|HTTP/WebSocket| API[PriceTracker API]
Charlie[Charlie - DevOps] -->|監視| API
API -->|クエリ| DB[(PostgreSQL)]
API -->|キャッシュ| Redis[(Redis)]
API -->|タスク| Celery[Celeryワーカー]
Celery -->|スクレイピング| External[外部サイト]
(1) ▶ サンプル:PriceTracker—最小動作版
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
app = FastAPI(title="PriceTracker API", version="0.1.0")
class ProductCreate(BaseModel):
name: str = Field(max_length=200)
category: str = Field(max_length=100)
base_price: float = Field(gt=0, description="米ドルでの基本価格")
class ProductResponse(BaseModel):
id: int
name: str
category: str
base_price: float
PRODUCTS_DB: dict[int, dict] = {}
_counter = 0
@app.post("/products", response_model=ProductResponse)
async def create_product(product: ProductCreate):
global _counter
_counter += 1
record = {"id": _counter, **product.model_dump()}
PRODUCTS_DB[_counter] = record
return record
@app.get("/products/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int):
if product_id not in PRODUCTS_DB:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Product not found")
return PRODUCTS_DB[product_id]
出力:
# 関数定義成功
8. 総合サンプル
FastAPIの核心的な強みは, 型ヒント驱动の自動バリデーションとドキュメント生成にあります。以下は, パスパラメータ, Pydanticモデル, レスポンスモデルを統合した価格クエリエンドポイントのデモです。
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI(title="PriceTracker Demo")
class PriceResponse(BaseModel):
product_id: int = Field(gt=0)
product_name: str
price: float = Field(gt=0)
currency: str = "USD"
PRICES_DB: dict[int, dict] = {
1: {"product_id": 1, "product_name": "Widget", "price": 9.99},
2: {"product_id": 2, "product_name": "Gadget", "price": 24.50},
}
@app.get("/products/{product_id}", response_model=PriceResponse)
async def get_price(product_id: int):
if product_id not in PRICES_DB:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Product not found")
return PRICES_DB[product_id]
出力:
GET /products/1 → {"product_id":1,"product_name":"Widget","price":9.99,"currency":"USD"}
GET /products/99 → 404 Not Found
❓ よくある質問
async/awaitを使わなければなりませんか?asyncのほうがパフォーマンスが良いです。📖 まとめ
- FastAPIはASGI非同期プロトコルに基づいており, 単一プロセスで数千の同時リクエストを処理でき, WSGIフレームワークを大幅に上回るパフォーマンスを発揮します。
- 型ヒントはデータバリデーション, シリアライズ, OpenAPIドキュメント生成を同時に驱动し, ボイラープレートコードを70%削減します。
- FastAPIはStarlette (ASGIフレームワーク)とPydantic (データバリデーション)の軽量ラッパーであり, 各層は独立して使用可能です。
- Pydantic V2はコアをRustで書き直し, V1より5〜50倍高速であり, FastAPIのパフォーマンスの鍵です。
- PriceTrackerプロジェクトは25レッスンにわたり, ゼロからの構築から本番デプロイまで, 数百万ユーザーを処理するSaaSシナリオを網羅します。
📝 練習問題
- 基本問題 (難易度 ⭐):FastAPIとUvicornをインストールし,
{"message": "Hello PriceTracker"}を返すGETエンドポイントを作成し,uvicornで起動してください。ヒント:pip install fastapi uvicorn - 応用問題 (難易度 ⭐⭐):エンドポイントにパスパラメータ
nameを追加し,{"message": "Hello, {name}"}を返すようにし, ブラウザで/docsにアクセスして自動生成されたドキュメントを確認してください。ヒント:@app.get("/hello/{name}") - チャレンジ問題 (難易度 ⭐⭐⭐):
product_name: strとprice: float(0より大きい必要がある)を含むPydanticモデルPriceInputを作成し, POSTエンドポイントでデータを受信してバリデーション結果を返してください。ヒント:BaseModelを継承しField(gt=0)を使用



