python-backend-expert

Python バックエンドエキスパート

使用する場面

以下の場合にこのスキルを有効にします。

• FastAPI のルートハンドラー（エンドポイント）を作成または変更するとき
• Pydantic v2 のリクエスト／レスポンススキーマを定義または更新するとき
• SQLAlchemy 2.0 の非同期モデル、クエリ、またはリレーションシップを記述するとき
• データアクセスにリポジトリパターンを実装するとき
• サービスレイヤーのビジネスロジックを記述するとき
• Alembic マイグレーションを作成または実行するとき
• Depends() を使った依存性注入チェーンを設定するとき
• ルート／サービス／リポジトリレイヤー全体でエラーを処理するとき

以下の場合、このスキルは使用しないでください。

• バックエンドコードのテストを記述するとき（pytest-patterns を使用）
• FastAPI フレームワークの仕組み（ミドルウェア、WebSocket、OpenAPI のカスタマイズ、CORS、ライフスパン）を扱うとき（fastapi-patterns を使用）
• デプロイまたは CI/CD パイプラインの設定を行うとき（deployment-pipeline を使用）
• API コントラクトの設計またはエンドポイントの計画を行うとき（api-design-patterns を使用）
• アーキテクチャの決定またはレイヤー設計を行うとき（system-architecture を使用）

手順

プロジェクト構成

app/
├── main.py              # FastAPI application factory
├── core/
│   ├── config.py        # pydantic-settings configuration
│   ├── database.py      # Async engine, session factory
│   └── security.py      # Password hashing, JWT utilities
├── models/              # SQLAlchemy ORM models
│   ├── __init__.py
│   ├── base.py          # Declarative base
│   └── user.py
├── schemas/             # Pydantic v2 schemas
│   ├── __init__.py
│   └── user.py
├── repositories/        # Data access layer
│   ├── __init__.py
│   └── user_repo.py
├── services/            # Business logic layer
│   ├── __init__.py
│   └── user_service.py
├── routes/              # FastAPI routers
│   ├── __init__.py
│   └── users.py
├── dependencies/        # Reusable Depends() providers
│   ├── __init__.py
│   └── auth.py
└── exceptions.py        # Domain exception classes

FastAPI エンドポイントのパターン

すべてのエンドポイントは、次の構成に従います。

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

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    session: AsyncSession = Depends(get_async_session),
) -> UserResponse:
    service = UserService(session)
    try:
        user = await service.create_user(data)
        return UserResponse.model_validate(user)
    except ConflictError as e:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> UserResponse:
    service = UserService(session)
    try:
        user = await service.get_user(user_id)
        return UserResponse.model_validate(user)
    except NotFoundError as e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

ルール：

• ルートが扱うのは、ステータスコード、HTTPException、レスポンスの整形などの HTTP 関連事項のみです
• ルートはサービスを呼び出し、リポジトリを直接呼び出すことはありません
• 自動レスポンスシリアライズと OpenAPI ドキュメント生成には、response_model を使用します
• 整数を直接指定せず、status.HTTP_* 定数を使用します
• セッション、認証、サービスの注入には Depends() を使用します

リポジトリパターン

リポジトリにすべてのデータベースアクセスをカプセル化します。

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

from app.models.user import User


class UserRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

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

    async def get_by_email(self, email: str) -> User | None:
        result = await self._session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def list_with_posts(
        self, *, offset: int = 0, limit: int = 20
    ) -> list[User]:
        result = await self._session.execute(
            select(User)
            .options(selectinload(User.posts))
            .offset(offset)
            .limit(limit)
        )
        return list(result.scalars().all())

    async def create(self, user: User) -> User:
        self._session.add(user)
        await self._session.flush()
        await self._session.refresh(user)
        return user

    async def update(self, user: User, **kwargs: object) -> User:
        for key, value in kwargs.items():
            setattr(user, key, value)
        await self._session.flush()
        await self._session.refresh(user)
        return user

    async def delete(self, user: User) -> None:
        await self._session.delete(user)
        await self._session.flush()

ルール：

• モデル（または集約ルート）ごとにリポジトリを1つ用意します
• リポジトリはモデルのインスタンスまたは None を返します。HTTP レスポンスを返すことはありません
• リポジトリにビジネスロジックを含めません
• 生成フィールド（ID、タイムスタンプ）を取得するため、add() の後は必ず flush() と refresh() を実行します
• 非同期コンテキストでリレーションシップを先行読み込みするには、selectinload() を使用します
• リポジトリから HTTPException を送出しないでください

サービスレイヤーのパターン

サービスにはビジネスロジックを含め、リポジトリを連携させます。

from app.exceptions import ConflictError, NotFoundError
from app.models.user import User
from app.repositories.user_repo import UserRepository
from app.schemas.user import UserCreate, UserPatch
from app.core.security import hash_password


class UserService:
    def __init__(self, session: AsyncSession) -> None:
        self.repo = UserRepository(session)

    async def create_user(self, data: UserCreate) -> User:
        # Business rule: email must be unique
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError(f"Email {data.email} already registered")

        # Business logic: hash password before storing
        user = User(
            email=data.email,
            hashed_password=hash_password(data.password),
            display_name=data.display_name,
        )
        return await self.repo.create(user)

    async def get_user(self, user_id: int) -> User:
        user = await self.repo.get_by_id(user_id)
        if user is None:
            raise NotFoundError(f"User {user_id} not found")
        return user

    async def update_user(self, user_id: int, data: UserPatch) -> User:
        user = await self.get_user(user_id)
        update_fields = data.model_dump(exclude_unset=True)
        if "password" in update_fields:
            update_fields["hashed_password"] = hash_password(update_fields.pop("password"))
        return await self.repo.update(user, **update_fields)

ルール：

• サービスはドメイン例外（NotFoundError、ConflictError）を送出します。HTTPException は絶対に送出しません
• ビジネスロジックを記述するのはサービスのみです
• サービスはデータアクセスにリポジトリを呼び出し、生のクエリを実行しません
• サービスはコンストラクターを通じて AsyncSession を受け取り、自身のリポジトリインスタンスを作成します
• サービスはリポジトリを呼び出す前にビジネスルールを検証します

ドメイン例外

ドメイン例外の階層を定義します。

class AppError(Exception):
    """Base application error."""

class NotFoundError(AppError):
    """Resource not found."""

class ConflictError(AppError):
    """Resource conflict (duplicate, version mismatch)."""

class ValidationError(AppError):
    """Business rule violation."""

class PermissionError(AppError):
    """Insufficient permissions."""

FastAPI アプリにグローバル例外ハンドラーを登録します。

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

app = FastAPI()

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError) -> JSONResponse:
    return JSONResponse(status_code=404, content={"detail": str(exc), "code": "NOT_FOUND"})

@app.exception_handler(ConflictError)
async def conflict_handler(request: Request, exc: ConflictError) -> JSONResponse:
    return JSONResponse(status_code=409, content={"detail": str(exc), "code": "CONFLICT"})

これにより、サービスは HTTP を意識せずにドメイン例外を送出でき、ルートに try/except ブロックを記述する必要もなくなります。

Pydantic v2 スキーマの規約

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


class UserCreate(BaseModel):
    """POST request body — writable fields only, no id/timestamps."""
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)
    display_name: str = Field(min_length=1, max_length=100)


class UserPatch(BaseModel):
    """PATCH request body — all fields Optional."""
    email: EmailStr | None = None
    password: str | None = Field(default=None, min_length=8, max_length=128)
    display_name: str | None = Field(default=None, min_length=1, max_length=100)


class UserResponse(BaseModel):
    """Response body — all fields including id and timestamps."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    display_name: str
    is_active: bool
    created_at: datetime
    updated_at: datetime

Pydantic v2 の主なパターン：

• class Config: orm_mode = True の代わりに ConfigDict(from_attributes=True) を使用します
• from_orm() の代わりに model_validate() を使用します
• .dict() の代わりに model_dump() を使用します
• PATCH で「送信されていない」値と「null に設定された」値を区別するには、model_dump(exclude_unset=True) を使用します
• バリデーション制約には Field() を使用します
• Optional[str] ではなく、str | None 構文を使用します（Python 3.12 以降）

非同期セッション管理

from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.core.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        async with session.begin():
            yield session

ルール：

• expire_on_commit=False にすると、コミット後のデタッチされたインスタンスに関するエラーを防止できます
• session.begin() コンテキストマネージャーは、成功時に自動コミットし、例外発生時にロールバックします
• Depends(get_async_session) を通じて、リクエストごとに1つのセッションを使用します
• 同時実行タスク間でセッションを共有しないでください
• バックグラウンドタスクでは新しいセッションを作成し、リクエストのセッションを再利用しないでください

SQLAlchemy 2.0 モデルのパターン

from datetime import datetime
from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    display_name: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    # Relationships — ALWAYS use selectin or joined for async
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author", lazy="selectin"
    )

ルール：

• Mapped[type] アノテーション（SQLAlchemy 2.0 スタイル）を使用します
• Column() の代わりに mapped_column() を使用します
• 非同期処理との互換性のため、リレーションシップに lazy="selectin" を設定します
• データベースで生成されるデフォルト値には server_default を使用します
• 必ず created_at と updated_at のタイムスタンプを含めます

Alembic マイグレーションのワークフロー

# Generate migration from model changes
alembic revision --autogenerate -m "add_users_table"

# Review the generated migration file before applying

# Apply migration
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Show current revision
alembic current

# Show migration history
alembic history

マイグレーションの命名規則：

# alembic/env.py
naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

ルール：

• 自動生成されたマイグレーションは、適用する前に必ず確認します
• すべてのマイグレーションに、正常に動作する downgrade() 関数が必要です
• 論理的なスキーマ変更ごとに、マイグレーションを1つ作成します
• アップグレードとダウングレードの両方をテストします
• 説明的なマイグレーションメッセージを使用します："add_users_table"、"add_email_index_to_users"

依存性注入パターン

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_async_session
from app.services.user_service import UserService


async def get_user_service(
    session: AsyncSession = Depends(get_async_session),
) -> UserService:
    return UserService(session)


# Chain dependencies for auth
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    user_id = decode_token(token)
    service = UserService(session)
    return await service.get_user(user_id)


async def require_admin(
    user: User = Depends(get_current_user),
) -> User:
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin required")
    return user

例

リクエストの完全なフロー

POST /users へのリクエストは、すべてのレイヤーを経由します：

1. ルートが UserCreate を受け取ります（Pydantic がリクエストボディを検証します）
2. ルートが Depends() を使って UserService.create_user(data) を呼び出します
3. サービスが UserRepository.get_by_email() を使ってビジネスルール（メールアドレスの一意性）を確認します
4. サービスがパスワードをハッシュ化し、User モデルのインスタンスを作成します
5. サービスが UserRepository.create(user) を呼び出して永続化します
6. リポジトリがセッションに追加し、フラッシュしてからリフレッシュし、生成されたフィールドを取得します
7. ルートが model_validate() を使って ORM モデルを UserResponse に変換します

メールアドレスが重複している場合、サービスは ConflictError を発生させ、グローバル例外ハンドラーが 409 Conflict を返します。ルートで try/except を使う必要はありません。

エッジケース

• デタッチされたインスタンスのエラー： session.add() の後は必ず flush() と refresh() を呼び出します。セッションファクトリーで expire_on_commit=False を設定します。

• バックグラウンドタスクでの非同期セッション： リクエストのセッションを再利用してはいけません。新しいセッションを作成します：

  async def background_job():
      async with async_session_factory() as session:
          async with session.begin():
              # do work
  

• N+1 クエリ： アクセスするリレーションシップを含むリポジトリクエリでは、selectinload() を使用します。モデルのリレーションシップのデフォルトとして lazy="selectin" を設定します。

• 一括操作： 1件ずつ追加する代わりに、一括挿入には session.execute(insert(User).values(list_of_dicts)) を使用します。

• 複数のサービスにまたがるトランザクション： すべてのサービスに同じセッションを渡します。セッションの begin() コンテキストマネージャーがトランザクションの境界を処理します。

• Pydantic v2 の計算フィールド： レスポンススキーマの派生値には @computed_field を使用します。references/pydantic-v2-migration.md を参照してください。

高度なクエリ最適化パターンについては、references/sqlalchemy-patterns.md を参照してください。