backend-dev

バックエンド開発

FastAPI エンドポイント、Pydantic スキーマ、データベース操作、LLM 統合、または Python のサービスロジックを作成または変更する場合に使用します。

コードを書く前に

1. アーキテクチャについて docs/agent/architecture/backend-guide.md を読みます
2. API 契約について docs/agent/apis/front-end-apis.md を読みます
3. LLM のパターンについて docs/agent/llm-integration.md を読みます
4. まず関連するディレクトリにある既存のコードを確認します

必ず守るルール

1. すべての関数に型ヒントを付ける必要があります ― 例外はありません
2. 可変のデフォルト値には copy.deepcopy() を使用します
3. サーバー側で詳細なエラーをログに記録し、クライアントには汎用的なメッセージを返します
4. 共有リソースの初期化には asyncio.Lock() を使用します
5. API キーは api_key= を使って litellm に直接渡し、os.environ は使用しません

プロジェクト構成

apps/backend/app/
├── main.py          # FastAPI app, CORS, lifespan, routers
├── config.py        # Pydantic BaseSettings from env
├── database.py      # TinyDB wrapper (JSON file storage)
├── llm.py           # LiteLLM wrapper (multi-provider)
├── routers/         # API endpoint handlers
├── services/        # Business logic layer
├── schemas/         # Pydantic request/response models
└── prompts/         # LLM prompt templates (Jinja2)

パターン

新しいエンドポイント

from fastapi import APIRouter, HTTPException
from app.schemas.my_schema import MyRequest, MyResponse
import logging

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1", tags=["my-feature"])

@router.post("/my-endpoint", response_model=MyResponse)
async def create_thing(request: MyRequest) -> MyResponse:
    try:
        result = await process_data(request)
        return result
    except Exception as e:
        logger.error(f"Failed to create thing: {e}")
        raise HTTPException(status_code=500, detail="Operation failed. Please try again.")

新しいスキーマ

from pydantic import BaseModel, Field

class MyRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=200)
    description: str | None = None

class MyResponse(BaseModel):
    id: str
    name: str
    status: str = "created"

データベース操作

import copy
from app.database import get_db

DEFAULT_DATA = {"sections": [], "metadata": {}}

async def get_or_create(doc_id: str) -> dict:
    db = get_db()
    existing = db.get(doc_id)
    if existing:
        return existing
    data = copy.deepcopy(DEFAULT_DATA)  # ALWAYS deepcopy mutable defaults
    db.insert(data)
    return data

LLM 呼び出し

from app.llm import get_completion
from app.config import settings

async def improve_text(text: str) -> str:
    prompt = f"Improve this resume text:\n\n{text}"
    result = await get_completion(
        prompt=prompt,
        model=settings.LLM_MODEL,
        api_key=settings.LLM_API_KEY,  # Pass directly, not via env
        json_mode=True,
    )
    return result

エラー処理のパターン

except Exception as e:
    logger.error(f"Operation failed: {e}")  # Detailed for server logs
    raise HTTPException(
        status_code=500,
        detail="Operation failed. Please try again."  # Generic for client
    )

チェックリスト

• すべての関数に型ヒントが付いている
• 可変のデフォルト値に copy.deepcopy() を使用している
• エラー処理で詳細をログに記録し、汎用的なメッセージを返している
• 新しいエンドポイントが main.py のルーター登録に追加されている
• すべてのリクエスト／レスポンス本文にスキーマが定義されている
• API キーを api_key= パラメーター経由で渡している