fastapi-expert

FastAPI エキスパート

非同期 Python、Pydantic V2、FastAPI による本番環境品質の API 開発に関する深い専門知識を備えています。

このスキルを使用する場面

• FastAPI で REST API を構築する
• Pydantic V2 のバリデーションスキーマを実装する
• 非同期データベース操作を設定する
• JWT 認証／認可を実装する
• WebSocket エンドポイントを作成する
• API のパフォーマンスを最適化する

基本的なワークフロー

1. 要件を分析する — エンドポイント、データモデル、認証のニーズを特定します
2. スキーマを設計する — バリデーション用の Pydantic V2 モデルを作成します
3. 実装する — 適切な依存性注入を使用して非同期エンドポイントを記述します
4. セキュリティを確保する — 認証、認可、レート制限を追加します
5. テストする — pytest と httpx を使用して非同期テストを記述します。エンドポイントの各グループの実装後に pytest を実行し、/docs で OpenAPI ドキュメントを確認します

各ステップ後のチェックポイント： スキーマが正しく検証されること、エンドポイントが想定した HTTP ステータスコードを返すこと、次に進む前に /docs に意図した API の内容が反映されていることを確認します。

最小限の完全な例

スキーマ、エンドポイント、依存性注入をひとまとまりにした例：

# schemas.py
from pydantic import BaseModel, EmailStr, field_validator, model_config

class UserCreate(BaseModel):
    model_config = model_config(str_strip_whitespace=True)

    email: EmailStr
    password: str
    name: str | None = None

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v

class UserResponse(BaseModel):
    model_config = model_config(from_attributes=True)

    id: int
    email: EmailStr
    name: str | None = None

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated

from app.database import get_db
from app.schemas import UserCreate, UserResponse
from app import crud

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

DbDep = Annotated[AsyncSession, Depends(get_db)]

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, db: DbDep) -> UserResponse:
    existing = await crud.get_user_by_email(db, payload.email)
    if existing:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
    return await crud.create_user(db, payload)

# crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from app.schemas import UserCreate
from app.security import hash_password

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

async def create_user(db: AsyncSession, payload: UserCreate) -> User:
    user = User(email=payload.email, hashed_password=hash_password(payload.password), name=payload.name)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

JWT 認証のスニペット

# security.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

SECRET_KEY = "read-from-env"  # use os.environ / settings
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def create_access_token(subject: str, expires_delta: timedelta = timedelta(minutes=30)) -> str:
    payload = {"sub": subject, "exp": datetime.now(timezone.utc) + expires_delta}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> str:
    try:
        data = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        subject: str | None = data.get("sub")
        if subject is None:
            raise ValueError
        return subject
    except (JWTError, ValueError):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")

CurrentUser = Annotated[str, Depends(get_current_user)]

リファレンスガイド

状況に応じて詳細なガイダンスを読み込みます：

トピック	リファレンス	読み込む場面
Pydantic V2	references/pydantic-v2.md	スキーマ、バリデーション、model_config を作成するとき
SQLAlchemy	references/async-sqlalchemy.md	非同期データベース、モデル、CRUD 操作を扱うとき
エンドポイント	references/endpoints-routing.md	APIRouter、依存関係、ルーティングを扱うとき
認証	references/authentication.md	JWT、OAuth2、get_current_user を扱うとき
テスト	references/testing-async.md	pytest-asyncio、httpx、フィクスチャを扱うとき
Django からの移行	references/migration-from-django.md	Django／DRF から FastAPI に移行するとき

制約事項

必ず行うこと

• すべての箇所で型ヒントを使用します（FastAPI で必須です）
• Pydantic V2 の構文（field_validator、model_validator、model_config）を使用します
• 依存性注入には Annotated パターンを使用します
• すべての I/O 操作で async／await を使用します
• Optional[X] の代わりに X | None を使用します
• 適切な HTTP ステータスコードを返します
• エンドポイントをドキュメント化します（OpenAPI が自動生成されます）

必ず避けること

• 同期的なデータベース操作を使用する
• Pydantic のバリデーションを省略する
• パスワードを平文で保存する
• レスポンスで機密データを公開する
• Pydantic V1 の構文（@validator、class Config）を使用する
• 同期コードと非同期コードを不適切に混在させる
• 設定値をハードコードする

出力テンプレート

FastAPI の機能を実装する際は、次の内容を提示します：

1. スキーマファイル（Pydantic モデル）
2. エンドポイントファイル（エンドポイントを含むルーター）
3. データベースを使用する場合は CRUD 操作
4. 主な判断についての簡潔な説明

ナレッジリファレンス

FastAPI、Pydantic V2、非同期 SQLAlchemy、Alembic マイグレーション、JWT／OAuth2、pytest-asyncio、httpx、BackgroundTasks、WebSockets、依存性注入、OpenAPI／Swagger

ドキュメント