app/
├── api/
│   ├── dependencies/
│   │   ├── auth.py        # get_current_user, require_permission
│   │   └── database.py    # get_db セッション依存
│   └── routers/
│       ├── auth.py        # /auth/* エンドポイント
│       ├── users.py       # /users/* エンドポイント
│       └── items.py       # /items/* エンドポイント
├── models/                # SQLAlchemy ORM モデル
│   ├── user.py
│   └── item.py
├── schemas/               # Pydantic スキーマ（Request/Response）
│   ├── user.py
│   └── item.py
├── services/              # ビジネスロジック層
│   ├── auth_service.py
│   └── item_service.py
├── crud/                  # DB CRUD 操作（生のクエリのみ）
│   ├── user_crud.py
│   └── item_crud.py
├── core/
│   ├── security.py        # JWT 処理・パスワードハッシュ
│   └── config.py          # pydantic-settings による設定管理
├── database.py            # 非同期エンジン・セッションファクトリ
└── main.py
conftest.py
gunicorn.conf.py
Dockerfile
docker-compose.yml

# app/api/dependencies/auth.py

from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.core.security import decode_access_token
from app.models.user import User
from app.crud.user_crud import get_user_by_id
from app.api.dependencies.database import get_db

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

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    payload = decode_access_token(token)
    if payload is None or payload.get("type") != "access":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    user = await get_user_by_id(db, int(payload["sub"]))
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return user

def require_permission(required: str):
    """権限チェックを返す高階関数"""
    def checker(user: Annotated[User, Depends(get_current_user)]) -> User:
        if required not in user.permissions:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
        return user
    return checker

# ルーター全体に認証を適用する場合
router = APIRouter(
    prefix="/items",
    dependencies=[Depends(get_current_user)],  # 全エンドポイントに適用
)

# 戻り値が不要な場合（ここが重要）
@router.delete("/{item_id}", dependencies=[Depends(require_permission("admin"))])
async def delete_item(item_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
    ...

# 戻り値を使う場合（引数として受け取る）
@router.get("/me")
async def get_my_items(current_user: Annotated[User, Depends(get_current_user)]):
    ...

# app/api/dependencies/database.py
from typing import Annotated, AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import AsyncSessionLocal

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

# 型エイリアスとして定義
DbSession = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

DATABASE_URL = "postgresql+asyncpg://user:pass@db:5432/mydb"

engine = create_async_engine(
    DATABASE_URL,
    echo=False,          # 本番では False
    pool_pre_ping=True,  # 切断済みコネクションの自動検知
    pool_size=10,
    max_overflow=20,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # ← 最重要設定
)

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called;
can't call await_() here.

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session
        # with ブロック終了時に自動的にセッションをクローズ
        # 例外発生時は自動ロールバック

# app/models/user.py
from datetime import datetime
from sqlalchemy import String, Boolean
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

# crud/user_crud.py
async def create_user(db: AsyncSession, user_data: dict) -> User:
    user = User(**user_data)
    db.add(user)
    await db.flush()  # IDを取得するためflushのみ（コミットしない）
    return user

# services/auth_service.py
async def register_user(db: AsyncSession, schema: UserCreate) -> User:
    hashed = hash_password(schema.password)
    user = await create_user(db, {"email": schema.email, "hashed_password": hashed})
    await send_welcome_email(user.email)  # 他の処理も同一トランザクション内
    await db.commit()  # サービス層でまとめてコミット
    return user

from sqlalchemy.orm import selectinload

result = await db.execute(
    select(User).options(selectinload(User.items))
)
users = result.scalars().all()

# アクセストークンのペイロード例
{
    "sub": "123",           # ユーザーID
    "type": "access",       # トークン種別（必須）
    "jti": "uuid-v4",       # JWT ID（リフレッシュトークンで再利用検知に使用）
    "scope": "read write",  # 権限スコープ
    "iss": "myapp",         # 発行者
    "aud": "myapp-api",     # 対象オーディエンス
    "exp": 1700000000,      # 有効期限
}

# app/core/security.py
from datetime import datetime, timedelta, timezone
from typing import Any
import jwt
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(subject: str, extra: dict[str, Any] | None = None) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {
        "sub": subject,
        "type": "access",
        "exp": expire,
        "iss": settings.JWT_ISSUER,
        "aud": settings.JWT_AUDIENCE,
        **(extra or {}),
    }
    return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def decode_access_token(token: str) -> dict | None:
    try:
        return jwt.decode(
            token,
            settings.SECRET_KEY,
            algorithms=[settings.ALGORITHM],
            audience=settings.JWT_AUDIENCE,
        )
    except jwt.InvalidTokenError:
        return None

# services/auth_service.py
async def refresh_tokens(db: AsyncSession, refresh_token: str) -> TokenPair:
    payload = decode_refresh_token(refresh_token)
    if payload is None:
        raise HTTPException(status_code=401, detail="Invalid token")

    jti = payload["jti"]
    stored = await get_refresh_token_by_jti(db, jti)

    if stored is None or stored.is_revoked:
        # 再利用検知 → 全セッション失効
        await revoke_all_user_tokens(db, int(payload["sub"]))
        raise HTTPException(status_code=401, detail="Token reuse detected")

    await revoke_refresh_token(db, jti)  # 旧トークンを失効
    new_access = create_access_token(payload["sub"])
    new_refresh = await create_refresh_token(db, payload["sub"])  # 新トークン発行
    await db.commit()
    return TokenPair(access_token=new_access, refresh_token=new_refresh)

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, field_validator, model_validator

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    password_confirm: str

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("パスワードは8文字以上必要です")
        if not any(c.isupper() for c in v):
            raise ValueError("大文字を1文字以上含めてください")
        return v

    @model_validator(mode="after")  # モデル全体を受け取る
    def passwords_match(self) -> "UserCreate":
        if self.password != self.password_confirm:
            raise ValueError("パスワードが一致しません")
        return self

class UserCreate(BaseModel):        # POST /users のリクエスト
    email: EmailStr
    password: str
    password_confirm: str

class UserUpdate(BaseModel):        # PATCH /users/{id} のリクエスト
    email: EmailStr | None = None
    is_active: bool | None = None

class UserResponse(BaseModel):      # 全エンドポイントのレスポンス
    id: int
    email: str
    is_active: bool
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)  # ORMオブジェクトから生成可能に

# app/core/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    DATABASE_URL: str
    SECRET_KEY: str
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 15
    REFRESH_TOKEN_EXPIRE_DAYS: int = 30
    JWT_ISSUER: str = "myapp"
    JWT_AUDIENCE: str = "myapp-api"

    model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)

@lru_cache
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

# conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.main import app
from app.database import Base
from app.api.dependencies.database import get_db

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
async def engine():
    eng = create_async_engine(TEST_DATABASE_URL)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await eng.dispose()

@pytest.fixture
async def db_session(engine):
    async_session = async_sessionmaker(engine, expire_on_commit=False)
    async with async_session() as session:
        yield session
        await session.rollback()  # テストごとにロールバックして独立性を保つ

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

app.dependency_overrides[get_db] = override_get_db

@pytest.fixture
async def auth_client(client, db_session):
    """認証済みクライアントのフィクスチャ"""
    # テストユーザーを作成してトークンを取得
    response = await client.post("/auth/login", data={
        "username": "test@example.com",
        "password": "Testpassword1",
    })
    token = response.json()["access_token"]
    client.headers["Authorization"] = f"Bearer {token}"
    return client

async def test_get_items_requires_auth(client):
    response = await client.get("/items/")
    assert response.status_code == 401

async def test_get_items_authenticated(auth_client):
    response = await auth_client.get("/items/")
    assert response.status_code == 200

# gunicorn.conf.py
import multiprocessing

# 非同期ワーカーの場合、CPUバウンドではないため (2×CPU+1) は過剰
# 1〜4程度から始めて負荷試験で調整する
workers = min(4, multiprocessing.cpu_count() * 2 + 1)
worker_class = "uvicorn.workers.UvicornWorker"
bind = "0.0.0.0:8000"
timeout = 60
graceful_timeout = 30
keepalive = 5
accesslog = "-"
errorlog = "-"
loglevel = "info"

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 起動時の処理（DB接続確認など）
    yield
    # シャットダウン時の処理
    await engine.dispose()  # コネクションプールを解放

app = FastAPI(lifespan=lifespan)

# Dockerfile
FROM python:3.12-slim AS builder

WORKDIR /app
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project

FROM python:3.12-slim AS runtime

WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH"

# 非rootユーザーで実行（セキュリティ）
RUN useradd --create-home appuser
USER appuser

COPY --chown=appuser:appuser app/ ./app/
COPY gunicorn.conf.py ./

EXPOSE 8000
CMD ["gunicorn", "-c", "gunicorn.conf.py", "app.main:app"]

# app/api/routers/health.py
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.dependencies.database import get_db

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

@router.get("/live")
async def liveness():
    """プロセスが生きているかのみ確認（Kubernetes liveness probe）"""
    return {"status": "ok"}

@router.get("/ready")
async def readiness(db: AsyncSession = Depends(get_db)):
    """依存サービス（DB等）への疎通確認（Kubernetes readiness probe）"""
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ok", "db": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"DB unreachable: {e}")

@router.get("/startup")
async def startup():
    """起動完了確認（Kubernetes startup probe）"""
    return {"status": "ok"}

# docker-compose.yml
services:
  app:
    build: .
    ports:
      - "8000:8000"
    env_file: .env
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
      interval: 5s
      timeout: 5s
      retries: 5
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data: