FastAPI 入門と基本的な使い方

1. はじめに

FastAPI は、Python 3.6 以降の型ヒント機能を活用した、高速で現代的な Web フレームワークです。API の構築に特化しており、自動的なデータ検証、シリアライズ、ドキュメント生成機能を提供します。

主な特徴：

高速性：Starlette と Pydantic をベースにし、Node.js や Go に近いパフォーマンスを実現
自動ドキュメント：Swagger UI および ReDoc を用いたインタラクティブな API ドキュメントの生成
型ヒント：Python の型ヒントを使用してデータの検証を行う
非同期対応：async/await をネイティブにサポート
使いやすさ：学習コストが低く、開発効率が高い

2. インストールとセットアップ

2.1 環境要件

Python 3.7+
最新版の pip

2.2 FastAPI と Uvicorn のインストール

pip install fastapi uvicorn[standard]

uvicorn は、ASGI サーバーとして FastAPI アプリケーションを実行するために使用されます。

2.3 仮想環境の作成

python -m venv venv
# Linux/Mac
source venv/bin/activate
# Windows
venv\Scripts\activate

3. 最初の FastAPI アプリケーションを作成する

3.1 基本的なアプリケーション

main.py ファイルを作成します：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

3.2 アプリケーションの実行

uvicorn main:app --reload

引数の説明：

main：Python ファイル名（拡張子なし）
app：FastAPI インスタンスの変数名
--reload：開発時の再読み込み機能を有効にする

3.3 アプリケーションへのアクセス

http://127.0.0.1:8000 でレスポンスを確認
http://127.0.0.1:8000/docs で Swagger UI ドキュメントを表示
http://127.0.0.1:8000/redoc で ReDoc ドキュメントを表示

4. リクエスト処理

4.1 パスパラメータ

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

FastAPI はパスパラメータを宣言された型（例：int）に自動的に変換します。

4.2 クエリパラメータ

@app.get("/items/")
async def read_items(q: str = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

例：/items/?q=test&skip=10&limit=20

4.3 リクエストボディ

Pydantic モデルを使用してリクエストボディを定義します：

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.post("/items/")
async def create_item(item: Item):
    return item

FastAPI は以下の処理を自動で行います：

JSON 形式のリクエストボディを読み込む
型に合わせて変換する
データを検証する
ドキュメントを生成する

5. レスポンスモデル

5.1 レスポンスモデルの定義

class UserOut(BaseModel):
    username: str
    email: str
    full_name: str = None

@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user

response_model により、モデルに一致しないフィールドは自動的に除外されます。

5.2 ステータスコード

from fastapi import status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

6. エラー処理

6.1 HTTPException

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 0:
        raise HTTPException(
            status_code=400,
            detail="Item ID must be positive"
        )
    return {"item_id": item_id}

6.2 カスタム例外ハンドラ

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"message": str(exc)}
    )

7. 依存関係の注入

FastAPI では強力な依存関係注入システムが提供されています：

from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

依存関係はデータベース接続や認証などに使用できます。

8. データベースとの統合

8.1 SQLAlchemy の設定

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL, echo=True
)
AsyncSessionLocal = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as db:
        yield db

8.2 ルートでの使用

@app.post("/users/")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

9. ミドルウェア

9.1 CORS ミドルウェア

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

9.2 カスタムミドルウェア

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

10. テスト

10.1 テストクライアント

from fastapi.testclient import TestClient

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

10.2 データベーステスト

@pytest.fixture
async def db_session():
    async with AsyncSessionLocal() as session:
        yield session
        await session.rollback()

11. 配置

11.1 本番環境での実行

uvicorn main:app --host 0.0.0.0 --port 80 --workers 4

11.2 Docker でのデプロイ

FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

ビルドと実行：

docker build -t fastapi-app .
docker run -d -p 8000:8000 fastapi-app

12. 最適なプラクティス

非同期処理の活用：async/await を積極的に使用してパフォーマンスを向上させる
モジュール分割：APIRouter を使って大規模アプリケーションを整理する
環境設定管理：Pydantic の BaseSettings を使用して設定を管理する
セキュリティ：OAuth2 や JWT による認証・認可を実装する
監視：ロギングとパフォーマンス監視機能を追加する

13. 一般的な問題

依存関係不足：`email-validator` エラーが発生する場合は、`pydantic[email]` をインストール
ポート競合：ポートを変更するには `--port 8001` を使用
データベース接続：本番環境ではコネクションプールを適切に設定
パフォーマンス向上：`uvloop` を使用してパフォーマンスを最適化

14. 学習リソース

公式ドキュメント
FastAPI GitHub
Pydantic ドキュメント

タグ: fastapi Python Webフレームワーク非同期処理 API開発

10月26日 01:56 投稿

異端開発室