python-fastapi-ops

FastAPI のパターン

FastAPI による最新の非同期 API 開発です。

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

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan - startup and shutdown."""
    # Startup
    app.state.db = await create_db_pool()
    yield
    # Shutdown
    await app.state.db.close()

app = FastAPI(
    title="My API",
    version="1.0.0",
    lifespan=lifespan,
)

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

リクエスト／レスポンスモデル

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    """Request model with validation."""
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)

class UserResponse(BaseModel):
    """Response model."""
    id: int
    name: str
    email: EmailStr
    created_at: datetime

    model_config = {"from_attributes": True}  # Enable ORM mode

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    db_user = await create_user_in_db(user)
    return db_user

パスパラメーターとクエリパラメーター

from fastapi import Query, Path
from typing import Annotated

@app.get("/users/{user_id}")
async def get_user(
    user_id: Annotated[int, Path(..., ge=1, description="User ID")],
):
    return await fetch_user(user_id)

@app.get("/users")
async def list_users(
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
    search: str | None = None,
):
    return await fetch_users(skip=skip, limit=limit, search=search)

依存性注入

from fastapi import Depends
from typing import Annotated

async def get_db():
    """Database session dependency."""
    async with async_session() as session:
        yield session

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    """Authenticate and return current user."""
    user = await authenticate_token(db, token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# Annotated types for reuse
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]

@app.get("/me")
async def get_me(user: CurrentUser):
    return user

例外処理

from fastapi import HTTPException
from fastapi.responses import JSONResponse

# Built-in HTTP exceptions
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await fetch_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

# Custom exception handler
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item {exc.item_id} not found"},
    )

ルーターの構成

from fastapi import APIRouter

# users.py
router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
async def list_users():
    return []

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id}

# main.py
from app.routers import users, items

app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")

クイックリファレンス

機能	使用方法
パスパラメーター	@app.get("/items/{id}")
クエリパラメーター	def f(q: str = None)
ボディ	def f(item: ItemCreate)
依存関係	Depends(get_db)
認証	Depends(get_current_user)
レスポンスモデル	response_model=ItemResponse
ステータスコード	status_code=201

その他のリソース

• ./references/dependency-injection.md - 高度な依存性注入パターン、スコープ、キャッシュ
• ./references/middleware-patterns.md - ミドルウェアチェーン、CORS、エラー処理
• ./references/validation-serialization.md - Pydantic v2 のパターン、カスタムバリデーター
• ./references/background-tasks.md - バックグラウンドタスク、非同期ワーカー、スケジューリング

スクリプト

• ./scripts/scaffold-api.sh - API エンドポイントのボイラープレートを生成

アセット

• ./assets/fastapi-template.py - 本番環境向け FastAPI アプリのスケルトン

---

関連項目

前提条件：

• python-typing-ops - Pydantic モデルと型ヒント
• python-async-ops - 非同期エンドポイントのパターン

関連スキル：

• python-database-ops - SQLAlchemy との統合
• python-observability-ops - ロギング、メトリクス、トレーシング用ミドルウェア
• python-pytest-ops - TestClient による API テスト