Posted inプログラミング

PythonとFastAPIでREST APIを構築する：ゼロから本番環境まで

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

FastAPIを学ぶきっかけとなった出来事

火曜日の午前2時のことだった。負荷がかかった状態で、Flask APIがリクエストの約15%に500エラーを返していた。ログを確認すると問題は明らかだった。リクエストのバリデーションなし、型変換なし、クライアントが実際に何を送ってきているのかまったく見えない状態。4時間の緊急対応の末、私はきちんと作り直すことを自分に誓った。

その週末、コアエンドポイントをFastAPIへ移行した。流行っているからではない——自分が直面していた問題をそのまま解決してくれるからだ。自動バリデーション、シリアライズ、そしてクライアントが午前2時に電話してこなくても使える組み込みドキュメント。

これは、あのログを眺めながら私が必要としていたチュートリアルだ。実際に動くAPIを構築しながら、本番環境で本当に重要なパターンを押さえ、サンプルコードにしか出てこない部分は省いていく。

最初のルートを書く前に知っておくべき基本概念

FastAPIとは何か

FastAPIは、リクエスト処理にStarlette、データバリデーションにPydanticを使ったASGI Webフレームワークだ。FlaskやDjango REST Frameworkとの重要な違いは、バリデーションがルート関数の実行前に行われること。リクエストボディがスキーマに一致しなければ、FastAPIは自動的に422 Unprocessable Entityを返す——不正なデータはあなたのコードに届かない。

私が移行したあるAPIでは、最初の1週間でまるごと1クラス分のバグが検出された。Flaskをこっそり通り抜けてダウンストリームの状態を壊していた、欠落フィールドや不正形式のフィールドが、FastAPIのバリデーション層を突破できなくなったのだ。

Pydanticモデルはすべての基盤

FastAPIのすべてのデータ構造は、PydanticのBaseModelから始まる。型ヒント付きでフィールドを定義すれば、Pydanticが型変換とバリデーションを自動処理してくれる。クライアントが"age": "25"を文字列で送ってきても、Pydanticがintにキャストする。"age": "banana"を送ってきたら？コードが実行される前に弾かれる。

デフォルトで非同期対応

FastAPIは同期・非同期どちらのルートハンドラーもサポートしている。データベース呼び出しやサードパーティAPIへのHTTPリクエストなど、I/Oバウンドな処理にはasync defを使う。CPUバウンドな処理や単純な同期ロジックには、通常のdefで問題ない（FastAPIが自動的にスレッドプールで実行する）。

実践：タスク管理APIを構築する

ステップ1：依存パッケージのインストール

クリーンな仮想環境から始める：

python -m venv venv
source venv/bin/activate  # Windowsの場合: venv\Scripts\activate
pip install fastapi uvicorn pydantic

UvicornはASGIサーバーだ。ローカルでも本番環境でもアプリの実行に必要になる。

ステップ2：アプリのディレクトリ構成を作成する

mkdir taskapi && cd taskapi
touch main.py models.py

ステップ3：データモデルを定義する

models.pyを開いて、リクエスト・レスポンスのスキーマを定義する：

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    priority: int = Field(default=1, ge=1, le=5)  # 1=低, 5=緊急

class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    priority: int
    created_at: datetime
    completed: bool = False

    class Config:
        from_attributes = True  # ORMモデルへの変換を許可

Field()の制約は自動的に適用される。priorityに6を指定すると、ハンドラーが実行される前に弾かれる。これだけで、防御的なコーディングに費やす時間が大幅に削減できた。

ステップ4：APIルートを構築する

main.pyでルートを組み立てる：

from fastapi import FastAPI, HTTPException, status
from datetime import datetime, timezone
from models import TaskCreate, TaskResponse
from typing import List

app = FastAPI(
    title="タスクAPI",
    description="本番対応のタスク管理REST API",
    version="1.0.0"
)

# このチュートリアル用のインメモリストア（実際のDBに置き換えること）
db: dict = {}
counter = 0

@app.post("/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskCreate):
    global counter
    counter += 1
    record = {
        "id": counter,
        "title": task.title,
        "description": task.description,
        "priority": task.priority,
        "created_at": datetime.now(timezone.utc),
        "completed": False
    }
    db[counter] = record
    return record

@app.get("/tasks", response_model=List[TaskResponse])
async def list_tasks():
    return list(db.values())

@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int):
    if task_id not in db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"タスク {task_id} が見つかりません"
        )
    return db[task_id]

@app.patch("/tasks/{task_id}/complete", response_model=TaskResponse)
async def complete_task(task_id: int):
    if task_id not in db:
        raise HTTPException(status_code=404, detail="タスクが見つかりません")
    db[task_id]["completed"] = True
    return db[task_id]

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int):
    if task_id not in db:
        raise HTTPException(status_code=404, detail="タスクが見つかりません")
    del db[task_id]

ステップ5：起動してテストする

uvicorn main:app --reload --port 8000

ブラウザでhttp://localhost:8000/docsを開いてみよう。FastAPIが自動的にSwagger UIを生成してくれる——追加の設定は不要だ。チーム（あるいは午前2時に電話してきたあのクライアント）がブラウザから直接すべてのエンドポイントをテストできる。

curlでテストする場合：

# タスクを作成する
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "APIのバグを修正する", "priority": 5}'

# タスク一覧を取得する
curl http://localhost:8000/tasks

# タスクを完了にする
curl -X PATCH http://localhost:8000/tasks/1/complete

ステップ6：本番対応のミドルウェアを追加する

デプロイ前に、CORSヘッダーと基本的なリクエストロガーを追加しておこう：

from fastapi.middleware.cors import CORSMiddleware
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],  # 本番環境では必ず制限すること
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = round((time.time() - start) * 1000, 2)
    logger.info(f"{request.method} {request.url.path} → {response.status_code} ({duration}ms)")
    return response

このミドルウェアが、あの午前2時に私が欠いていた視野をもたらしてくれる。メソッド、パス、ステータスコード、処理時間——この4つのフィールドだけで、アプリケーションロジックに触れることなく本番デバッグの80%をカバーできる。

ステップ7：Gunicornで本番環境として起動する

pip install gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

-w 4フラグで4つのワーカープロセスを起動する。よく使われる目安はCPUコア数 × 2 + 1だ。各ワーカーが独立してリクエストを処理するため、1つのリクエストがハングしても他のリクエストはブロックされない。

次のステップ

実際の本番デプロイに向けて、この基盤の上に重ねていくべきものを示す：

  • dictストアをSQLAlchemy + PostgreSQLまたはSQLModel（FastAPIの作者がまさにこのユースケースのために作ったライブラリ）に置き換える
  • python-joseを使ったJWTトークンとFastAPIの依存性注入で認証を追加する
  • pytest + httpxでテストを書く——FastAPIのTestClientがこれをシンプルにしてくれる
  • slowapi（Flask-LimiterのFastAPI互換ポート）でレート制限を追加する

今手元にあるのは、クライアントとサーバー間の型付きコントラクトだ。不正なデータは境界で弾かれ、ドキュメントは自動生成され、すべてのリクエストがログに残る。これがコアだ。あとはデータベースをつなぎ、認証を組み込めば、実際のユーザーの前に公開できるものが完成する。

Share:
Tags: