fastapi-endpoint

FastAPI エンドポイントビルダー

使用するタイミング

次のような場合にこのスキルを使用します:

• 既存の FastAPI プロジェクトに新しい API エンドポイントを追加するとき
• 適切なバリデーションとエラーハンドリングを備えた CRUD 操作を構築するとき
• 依存性注入を使って認証が必要なエンドポイントを設定するとき
• SQLAlchemy 2.0 で非同期データベースクエリを作成するとき
• API ルートの完全なテストカバレッジを生成するとき

フェーズ 1: 探索 (プランモード)

プランモードに入ります。コードを書く前に、既存のプロジェクトを調べて次の点を把握します。

プロジェクト構造

• FastAPI アプリのエントリポイント (main.py、app.py、または app/__init__.py) を見つける
• ルーターの構成パターン (単一ファイルか routers/ ディレクトリか) を確認する
• 既存の models/、schemas/、crud/、services/ ディレクトリの有無を確認する
• インストール済みの依存関係を pyproject.toml または requirements.txt で確認する

既存のパターン

• 既存のエンドポイントはどのように構成されているか? (関数ベースかクラスベースか)
• 使用している ORM は何か? (SQLAlchemy 2.0 非同期、Tortoise、生の SQL、なし)
• データベースセッションはどのように管理されているか? (Depends(get_db)、ミドルウェア、その他)
• どの認証パターンが存在するか? (OAuth2PasswordBearer、API キーヘッダー、カスタム)
• 既存の Pydantic ベースモデルや共有スキーマはあるか?
• 標準のレスポンス形式は何か? (モデルをそのまま返すか、{"data": ..., "meta": ...} でラップするか)

テストのパターン

• テストはどこに置かれているか? (tests/、test_*.py、*_test.py)
• どのテストクライアントを使用しているか? (httpx AsyncClient、TestClient、pytest-asyncio)
• データベースと認証用のテストフィクスチャはあるか?

フェーズ 2: インタビュー (AskUserQuestion)

AskUserQuestion を使って要件を明確にします。質問は一度にまとめて出すのではなく、ラウンドに分けて尋ねてください。

ラウンド 1: コアとなるエンドポイント

Question: "What resource does this endpoint manage?"
Header: "Resource"
Options:
  - "New resource (I'll describe the fields)" — Creating a new data model from scratch
  - "Existing model (extend it)" — Adding endpoints for a model that already exists in the codebase
  - "Relationship endpoint (nested)" — e.g., /users/{id}/orders — endpoint on a related resource

Question: "Which HTTP methods do you need?"
Header: "Methods"
multiSelect: true
Options:
  - "Full CRUD (GET list, GET detail, POST, PUT/PATCH, DELETE)" — All standard operations
  - "Read-only (GET list + GET detail)" — No mutations
  - "Custom action (POST /resource/{id}/action)" — Business logic endpoint, not standard CRUD

ラウンド 2: データモデル (新しいリソースの場合)

Question: "What fields does the resource have? (describe briefly)"
Header: "Fields"
Options:
  - "Simple (< 6 fields, basic types)" — Strings, ints, booleans, dates
  - "Medium (6-15 fields, some relations)" — Includes foreign keys or enums
  - "Complex (nested objects, polymorphic)" — JSON fields, discriminated unions, computed fields

ラウンド 3: 認証とアクセス制御

Question: "How should this endpoint be authenticated?"
Header: "Auth"
Options:
  - "JWT Bearer token (Recommended)" — OAuth2PasswordBearer with JWT decode
  - "API Key header" — X-API-Key header validation
  - "No auth (public)" — Open endpoint, no authentication required
  - "Use existing auth" — Reuse the auth dependency already in the project

Question: "Do you need role-based access control?"
Header: "RBAC"
Options:
  - "No — any authenticated user" — Single permission level
  - "Yes — role check (admin, user, etc.)" — Require specific roles per endpoint
  - "Yes — ownership check" — Users can only access their own resources

ラウンド 4: ページネーション、フィルタリング、キャッシュ

Question: "What pagination style for list endpoints?"
Header: "Pagination"
Options:
  - "Cursor-based (Recommended)" — Best for real-time data, no offset drift
  - "Offset/limit" — Simple, good for admin panels with page numbers
  - "No pagination" — Small datasets, return all results

Question: "Do you need response caching?"
Header: "Caching"
Options:
  - "No caching" — Fresh data on every request
  - "Cache-Control headers" — Client-side caching via HTTP headers
  - "Redis/in-memory cache" — Server-side caching with TTL

フェーズ 3: 計画 (ExitPlanMode)

次の内容を含む、具体的な実装計画を作成します。

1. 作成・変更するファイル — フェーズ 1 で把握したプロジェクト構造に基づく正確なパス
2. Pydantic スキーマ — フィールド型を含む Create、Update、Response、List スキーマ
3. SQLAlchemy モデル — テーブル名、カラム、リレーションシップ、インデックス
4. CRUD/サービス層 — 各操作の非同期関数
5. ルーター — エンドポイントのシグネチャ、ステータスコード、レスポンスモデル
6. 依存関係 — 認証、ページネーション、フィルタリングの依存関係
7. テスト — 正常系、バリデーションエラー、認証失敗、見つからない場合のテストケース

ExitPlanMode を使ってユーザーの承認を得てください。

フェーズ 4: 実行

承認後、次の順序で実装します。

ステップ 1: Pydantic スキーマ

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from uuid import UUID

class ResourceBase(BaseModel):
    """Shared fields between create and response."""
    name: str
    # ... fields from interview

class ResourceCreate(ResourceBase):
    """Fields required to create the resource."""
    pass

class ResourceUpdate(BaseModel):
    """All fields optional for partial updates."""
    name: str | None = None

class ResourceResponse(ResourceBase):
    """Full resource with DB-generated fields."""
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    created_at: datetime
    updated_at: datetime

class ResourceListResponse(BaseModel):
    """Paginated list response."""
    data: list[ResourceResponse]
    next_cursor: str | None = None
    has_more: bool

ステップ 2: SQLAlchemy モデル

from sqlalchemy import Column, String, DateTime, func
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
import uuid
from app.database import Base

class Resource(Base):
    __tablename__ = "resources"

    id = Column(PG_UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String, nullable=False, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

ステップ 3: CRUD/サービス層

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

async def get_resource(db: AsyncSession, resource_id: UUID) -> Resource | None:
    result = await db.execute(select(Resource).where(Resource.id == resource_id))
    return result.scalar_one_or_none()

async def list_resources(
    db: AsyncSession,
    cursor: str | None = None,
    limit: int = 20,
) -> tuple[list[Resource], str | None]:
    query = select(Resource).order_by(Resource.created_at.desc()).limit(limit + 1)
    if cursor:
        query = query.where(Resource.created_at < decode_cursor(cursor))
    result = await db.execute(query)
    items = list(result.scalars().all())
    next_cursor = encode_cursor(items[-1].created_at) if len(items) > limit else None
    return items[:limit], next_cursor

async def create_resource(db: AsyncSession, data: ResourceCreate) -> Resource:
    resource = Resource(**data.model_dump())
    db.add(resource)
    await db.commit()
    await db.refresh(resource)
    return resource

async def update_resource(
    db: AsyncSession, resource_id: UUID, data: ResourceUpdate
) -> Resource | None:
    resource = await get_resource(db, resource_id)
    if not resource:
        return None
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(resource, field, value)
    await db.commit()
    await db.refresh(resource)
    return resource

async def delete_resource(db: AsyncSession, resource_id: UUID) -> bool:
    resource = await get_resource(db, resource_id)
    if not resource:
        return False
    await db.delete(resource)
    await db.commit()
    return True

ステップ 4: 依存関係を含むルーター

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from uuid import UUID

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

@router.get("", response_model=ResourceListResponse)
async def list_resources_endpoint(
    cursor: str | None = Query(None),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),  # if auth required
):
    items, next_cursor = await list_resources(db, cursor=cursor, limit=limit)
    return ResourceListResponse(
        data=items,
        next_cursor=next_cursor,
        has_more=next_cursor is not None,
    )

@router.get("/{resource_id}", response_model=ResourceResponse)
async def get_resource_endpoint(
    resource_id: UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    resource = await get_resource(db, resource_id)
    if not resource:
        raise HTTPException(status_code=404, detail="Resource not found")
    return resource

@router.post("", response_model=ResourceResponse, status_code=status.HTTP_201_CREATED)
async def create_resource_endpoint(
    data: ResourceCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    return await create_resource(db, data)

@router.patch("/{resource_id}", response_model=ResourceResponse)
async def update_resource_endpoint(
    resource_id: UUID,
    data: ResourceUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    resource = await update_resource(db, resource_id, data)
    if not resource:
        raise HTTPException(status_code=404, detail="Resource not found")
    return resource

@router.delete("/{resource_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_resource_endpoint(
    resource_id: UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    deleted = await delete_resource(db, resource_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Resource not found")

ステップ 5: テスト

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_resource(client: AsyncClient, auth_headers: dict):
    response = await client.post(
        "/resources",
        json={"name": "Test Resource"},
        headers=auth_headers,
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Test Resource"
    assert "id" in data

@pytest.mark.asyncio
async def test_get_resource_not_found(client: AsyncClient, auth_headers: dict):
    response = await client.get(
        "/resources/00000000-0000-0000-0000-000000000000",
        headers=auth_headers,
    )
    assert response.status_code == 404

@pytest.mark.asyncio
async def test_list_resources_pagination(client: AsyncClient, auth_headers: dict):
    # Create multiple resources first
    for i in range(5):
        await client.post(
            "/resources",
            json={"name": f"Resource {i}"},
            headers=auth_headers,
        )
    response = await client.get("/resources?limit=2", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert len(data["data"]) == 2
    assert data["has_more"] is True
    assert data["next_cursor"] is not None

@pytest.mark.asyncio
async def test_create_resource_unauthorized(client: AsyncClient):
    response = await client.post("/resources", json={"name": "Test"})
    assert response.status_code in (401, 403)

@pytest.mark.asyncio
async def test_update_resource_partial(client: AsyncClient, auth_headers: dict):
    # Create
    create_resp = await client.post(
        "/resources",
        json={"name": "Original"},
        headers=auth_headers,
    )
    resource_id = create_resp.json()["id"]
    # Partial update
    response = await client.patch(
        f"/resources/{resource_id}",
        json={"name": "Updated"},
        headers=auth_headers,
    )
    assert response.status_code == 200
    assert response.json()["name"] == "Updated"

@pytest.mark.asyncio
async def test_delete_resource(client: AsyncClient, auth_headers: dict):
    create_resp = await client.post(
        "/resources",
        json={"name": "To Delete"},
        headers=auth_headers,
    )
    resource_id = create_resp.json()["id"]
    response = await client.delete(
        f"/resources/{resource_id}", headers=auth_headers
    )
    assert response.status_code == 204
    # Verify deleted
    get_resp = await client.get(
        f"/resources/{resource_id}", headers=auth_headers
    )
    assert get_resp.status_code == 404

従うべき主要なパターン

認証のための依存性注入

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_jwt(token)
    user = await db.get(User, payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

def require_role(*roles: str):
    """Factory for role-based access control."""
    async def checker(current_user: User = Depends(get_current_user)):
        if current_user.role not in roles:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return checker

カーソルベースのページネーションヘルパー

import base64
from datetime import datetime

def encode_cursor(dt: datetime) -> str:
    return base64.urlsafe_b64encode(dt.isoformat().encode()).decode()

def decode_cursor(cursor: str) -> datetime:
    return datetime.fromisoformat(base64.urlsafe_b64decode(cursor).decode())

エラーレスポンス

常に FastAPI の HTTPException を一貫した detail メッセージとともに使用してください。バリデーションエラーについては、Pydantic v2 が RequestValidationError (422) を介して自動的に処理します。

# 404 — not found
raise HTTPException(status_code=404, detail="Resource not found")

# 409 — conflict (duplicate)
raise HTTPException(status_code=409, detail="Resource with this name already exists")

# 403 — forbidden
raise HTTPException(status_code=403, detail="Not allowed to modify this resource")

完了前のチェックリスト

• すべてのエンドポイントが適切なステータスコードを返す (POST は 201、DELETE は 204)
• Pydantic スキーマで ORM モードのために model_config = ConfigDict(from_attributes=True) を使用している
• リストエンドポイントに、設定可能な上限値付きのページネーションがある
• 公開されていないすべてのエンドポイントに認証依存関係が適用されている
• テストが正常系、見つからない場合、未認可、バリデーションエラーをカバーしている
• ルーターがメインの FastAPI アプリに登録されている
• データベースモデルで、フィルタリングや並び替えに使うカラムに適切なインデックスが設定されている