fastapi-coder

FastAPI コーダー

基本原則

原則	適用方法
非同期優先	すべての箇所で async/await を使用し、必要な場合にのみ同期処理を使用します
型安全性	すべてのリクエスト／レスポンスデータに Pydantic モデルを使用します
依存性注入	グローバル状態ではなく、共有ロジックには Depends() を使用します
OpenAPI 駆動	スキーマは自動生成されるため、クリーンな状態を保ちます
関心の分離	ルート → サービス → リポジトリ

プロジェクト構成

app/
├── main.py              # FastAPI app initialization
├── api/
│   ├── __init__.py
│   ├── deps.py          # Shared dependencies
│   └── routes/          # Route handlers by domain
│       ├── users.py
│       └── items.py
├── core/
│   ├── config.py        # Settings via pydantic-settings
│   ├── security.py      # Auth utilities
│   └── exceptions.py    # Custom exceptions
├── models/              # Pydantic schemas
│   ├── user.py
│   └── item.py
├── services/            # Business logic
│   └── user_service.py
├── repositories/        # Data access
│   └── user_repo.py
└── tests/
    ├── conftest.py      # Shared fixtures
    └── test_users.py

基本的なパターン

ルートハンドラー

from fastapi import APIRouter, Depends, HTTPException, status
from app.models.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.api.deps import get_user_service

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

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    """Create a new user."""
    return await service.create(user_in)

Pydantic モデル

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

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}

依存関係

from typing import Annotated
from fastapi import Depends, Header, HTTPException
from app.core.security import verify_token

async def get_current_user(
    authorization: Annotated[str, Header()],
) -> User:
    token = authorization.removeprefix("Bearer ")
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

サービスレイヤー

from app.repositories.user_repo import UserRepository
from app.models.user import UserCreate, UserResponse

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create(self, user_in: UserCreate) -> UserResponse:
        # Business logic here
        existing = await self.repo.get_by_email(user_in.email)
        if existing:
            raise ValueError("Email already registered")
        return await self.repo.create(user_in)

例外処理

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

class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

バックグラウンドタスク

from fastapi import BackgroundTasks

async def send_welcome_email(email: str):
    # Async email sending
    ...

@router.post("/users/")
async def create_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
):
    user = await create_user_in_db(user_in)
    background_tasks.add_task(send_welcome_email, user.email)
    return user

データベース連携

SQLAlchemy 2.0 非同期処理

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

engine = create_async_engine("postgresql+asyncpg://...", echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

リポジトリパターン

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

認証パターン

JWT 認証

from datetime import datetime, timedelta
from jose import jwt, JWTError
from app.core.config import settings

def create_access_token(data: dict) -> str:
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({**data, "exp": expire}, settings.SECRET_KEY, algorithm="HS256")

async def verify_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None

テストパターン

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users/", json={
        "email": "[email protected]",
        "name": "Test User",
        "password": "securepass123",
    })
    assert response.status_code == 201
    assert response.json()["email"] == "[email protected]"

品質チェックリスト

• すべてのルートに response_model と status_code がある
• すべてのリクエスト／レスポンスデータに Pydantic モデルを使用している
• 共有ロジック（認証、DB、サービス）に依存関係を使用している
• サービスレイヤーでビジネスロジックをルートから分離している
• データアクセスにリポジトリパターンを使用している
• 適切なハンドラーを備えたカスタム例外を使用している
• データベース操作が非同期である
• ノンブロッキング操作にバックグラウンドタスクを使用している
• httpx AsyncClient を使った包括的なテストがある
• 全体を通して型ヒントを使用している