# 統一エラーレスポンス形式（RFC 7807 Problem Details準拠）
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
 
class ProblemDetail(BaseModel):
    """RFC 7807 Problem Details for HTTP APIs"""
    type: str              # エラータイプのURI
    title: str             # 人間が読めるエラータイトル
    status: int            # HTTPステータスコード
    detail: Optional[str]  # エラーの詳細説明
    instance: Optional[str]  # エラーが発生したリクエストURI
    errors: Optional[list[dict]] = None  # バリデーションエラー詳細
 
app = FastAPI()
 
# エラーハンドラー
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content=ProblemDetail(
            type=f"https://api.example.com/errors/{exc.detail.get('code', 'unknown')}",
            title=exc.detail.get("title", "Error"),
            status=exc.status_code,
            detail=exc.detail.get("detail"),
            instance=str(request.url),
            errors=exc.detail.get("errors"),
        ).model_dump(exclude_none=True),
    )

# バリデーションエラー (422)
{
    "type": "https://api.example.com/errors/validation_error",
    "title": "Validation Error",
    "status": 422,
    "detail": "入力データに問題があります",
    "instance": "/api/v1/users",
    "errors": [
        {"field": "email", "message": "メールアドレスの形式が不正です", "code": "invalid_format"},
        {"field": "age", "message": "年齢は0以上の整数で指定してください", "code": "out_of_range"}
    ]
}
 
# 認証エラー (401)
{
    "type": "https://api.example.com/errors/authentication_required",
    "title": "Authentication Required",
    "status": 401,
    "detail": "このリソースへのアクセスには認証が必要です"
}
 
# レート制限 (429)
{
    "type": "https://api.example.com/errors/rate_limit_exceeded",
    "title": "Rate Limit Exceeded",
    "status": 429,
    "detail": "リクエスト上限に達しました。60秒後に再試行してください",
    "retry_after": 60
}
 
# ビジネスルール違反 (409)
{
    "type": "https://api.example.com/errors/insufficient_stock",
    "title": "Insufficient Stock",
    "status": 409,
    "detail": "商品「MacBook Pro」の在庫が不足しています（要求: 5, 在庫: 2）"
}

# カーソルベースページネーション (FastAPI)
from fastapi import FastAPI, Query
import base64, json
 
app = FastAPI()
 
@app.get("/api/v1/users")
async def list_users(
    cursor: str = Query(None, description="ページネーションカーソル"),
    limit: int = Query(20, ge=1, le=100, description="取得件数"),
):
    # カーソルのデコード
    if cursor:
        decoded = json.loads(base64.b64decode(cursor))
        last_id = decoded['id']
        query = "SELECT * FROM users WHERE id > %s ORDER BY id LIMIT %s"
        users = db.execute(query, (last_id, limit + 1))
    else:
        query = "SELECT * FROM users ORDER BY id LIMIT %s"
        users = db.execute(query, (limit + 1,))
 
    users = list(users)
    has_next = len(users) > limit
    if has_next:
        users = users[:limit]
 
    # 次のカーソル生成
    next_cursor = None
    if has_next and users:
        next_cursor = base64.b64encode(
            json.dumps({"id": users[-1].id}).encode()
        ).decode()
 
    return {
        "data": [user.to_dict() for user in users],
        "pagination": {
            "next_cursor": next_cursor,
            "has_next": has_next,
            "limit": limit,
        }
    }

# 複合条件でのカーソルベースページネーション
# 例: created_at DESC, id DESC でソート
 
@app.get("/api/v1/orders")
async def list_orders(
    cursor: str = Query(None),
    limit: int = Query(20, ge=1, le=100),
    status: str = Query(None, description="ステータスフィルタ"),
):
    if cursor:
        decoded = json.loads(base64.b64decode(cursor))
        # 複合カーソル: 同じ created_at の場合に id で一意に特定
        query = """
            SELECT * FROM orders
            WHERE (created_at, id) < (%s, %s)
            {status_filter}
            ORDER BY created_at DESC, id DESC
            LIMIT %s
        """
        params = [decoded["created_at"], decoded["id"]]
    else:
        query = """
            SELECT * FROM orders
            {status_filter}
            ORDER BY created_at DESC, id DESC
            LIMIT %s
        """
        params = []
 
    # ステータスフィルタの動的追加
    if status:
        status_filter = "AND status = %s"
        params.append(status)
    else:
        status_filter = ""
 
    query = query.replace("{status_filter}", status_filter)
    params.append(limit + 1)
    orders = db.execute(query, params)
 
    orders = list(orders)
    has_next = len(orders) > limit
    if has_next:
        orders = orders[:limit]
 
    next_cursor = None
    if has_next and orders:
        last = orders[-1]
        next_cursor = base64.b64encode(json.dumps({
            "created_at": last.created_at.isoformat(),
            "id": last.id,
        }).encode()).decode()
 
    return {
        "data": [o.to_dict() for o in orders],
        "pagination": {"next_cursor": next_cursor, "has_next": has_next},
    }

# URL パスベース（最も一般的）
# GET /api/v1/users
# GET /api/v2/users
 
from fastapi import APIRouter
 
v1_router = APIRouter(prefix="/api/v1")
v2_router = APIRouter(prefix="/api/v2")
 
@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: int):
    """v1: name フィールドを返す"""
    user = await get_user(user_id)
    return {"id": user.id, "name": user.name, "email": user.email}
 
@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: int):
    """v2: first_name / last_name に分離"""
    user = await get_user(user_id)
    return {
        "id": user.id,
        "first_name": user.first_name,
        "last_name": user.last_name,
        "email": user.email,
    }

# FastAPI + JWT 認証
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta
 
app = FastAPI()
security = HTTPBearer()
 
SECRET_KEY = "your-secret-key"  # 実際は環境変数から取得
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
def create_access_token(user_id: str, roles: list[str]) -> str:
    """アクセストークン生成"""
    payload = {
        "sub": user_id,
        "roles": roles,
        "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
        "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
 
def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    """トークン検証ミドルウェア"""
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail={
            "code": "AUTH_TOKEN_EXPIRED",
            "title": "Token Expired",
            "detail": "認証トークンの有効期限が切れています",
        })
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail={
            "code": "AUTH_INVALID_TOKEN",
            "title": "Invalid Token",
            "detail": "無効な認証トークンです",
        })
 
def require_role(required_role: str):
    """ロールベース認可"""
    def role_checker(token: dict = Depends(verify_token)):
        if required_role not in token.get("roles", []):
            raise HTTPException(status_code=403, detail={
                "code": "AUTH_INSUFFICIENT_PERMISSIONS",
                "title": "Forbidden",
                "detail": f"この操作には '{required_role}' ロールが必要です",
            })
        return token
    return role_checker
 
# 使用例
@app.get("/api/v1/users")
async def list_users(token: dict = Depends(verify_token)):
    """認証必須エンドポイント"""
    return {"users": [...]}
 
@app.delete("/api/v1/users/{user_id}")
async def delete_user(user_id: str, token: dict = Depends(require_role("admin"))):
    """admin ロール必須"""
    pass

# FastAPI による自動ドキュメント生成
from fastapi import FastAPI, Query, Path, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum
 
app = FastAPI(
    title="注文管理API",
    version="1.0.0",
    description="ECサイトの注文管理を行うRESTful API",
)
 
class OrderStatus(str, Enum):
    pending = "pending"
    confirmed = "confirmed"
    shipped = "shipped"
    delivered = "delivered"
    cancelled = "cancelled"
 
class OrderItemCreate(BaseModel):
    """注文アイテム"""
    product_id: str = Field(..., description="商品ID", example="prod-456")
    quantity: int = Field(..., ge=1, le=100, description="数量", example=2)
 
class OrderCreate(BaseModel):
    """注文作成リクエスト"""
    user_id: str = Field(..., description="ユーザーID", example="user-123")
    items: list[OrderItemCreate] = Field(
        ..., description="注文アイテム", min_length=1
    )
    note: Optional[str] = Field(None, max_length=500, description="備考")
 
    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "user_id": "user-123",
                    "items": [
                        {"product_id": "prod-456", "quantity": 2},
                        {"product_id": "prod-789", "quantity": 1},
                    ],
                    "note": "配達は午前中にお願いします",
                }
            ]
        }
    }
 
class OrderResponse(BaseModel):
    """注文レスポンス"""
    id: str = Field(..., description="注文ID")
    status: OrderStatus = Field(..., description="注文ステータス")
    total_amount: int = Field(..., description="合計金額（円）")
    created_at: str = Field(..., description="作成日時 (ISO 8601)")
 
@app.post(
    "/api/v1/orders",
    response_model=OrderResponse,
    status_code=201,
    summary="注文作成",
    tags=["Orders"],
    responses={
        409: {"description": "在庫不足"},
        422: {"description": "バリデーションエラー"},
    },
)
async def create_order(order: OrderCreate):
    """新しい注文を作成する。
 
    - 注文アイテムは1件以上必須
    - 在庫がない場合は 409 Conflict を返す
    - 作成成功時は 201 Created + Location ヘッダーを返す
    """
    result = await order_service.create(order)
    return result

# OpenAPI スキーマに基づく契約テスト
import pytest
from fastapi.testclient import TestClient
from jsonschema import validate
 
client = TestClient(app)
 
class TestOrderAPI:
    """注文API の契約テスト"""
 
    def test_create_order_returns_201(self):
        """正常系: 注文作成は 201 を返す"""
        response = client.post("/api/v1/orders", json={
            "user_id": "user-123",
            "items": [{"product_id": "prod-456", "quantity": 2}],
        })
        assert response.status_code == 201
        data = response.json()
        assert "id" in data
        assert "status" in data
        assert data["status"] == "pending"
 
    def test_create_order_with_empty_items_returns_422(self):
        """異常系: 空のアイテムは 422 を返す"""
        response = client.post("/api/v1/orders", json={
            "user_id": "user-123",
            "items": [],
        })
        assert response.status_code == 422
 
    def test_create_order_without_auth_returns_401(self):
        """異常系: 認証なしは 401 を返す"""
        response = client.post("/api/v1/orders", json={
            "user_id": "user-123",
            "items": [{"product_id": "prod-456", "quantity": 2}],
        }, headers={})  # Authorization ヘッダーなし
        assert response.status_code == 401
 
    def test_list_orders_pagination(self):
        """ページネーション: next_cursor が返される"""
        response = client.get("/api/v1/orders?limit=2")
        assert response.status_code == 200
        data = response.json()
        assert "data" in data
        assert "pagination" in data
        assert "has_next" in data["pagination"]
 
    def test_response_matches_schema(self):
        """レスポンスが OpenAPI スキーマに準拠している"""
        expected_schema = {
            "type": "object",
            "required": ["id", "status", "total_amount", "created_at"],
            "properties": {
                "id": {"type": "string"},
                "status": {"type": "string", "enum": ["pending", "confirmed", "shipped", "delivered", "cancelled"]},
                "total_amount": {"type": "integer"},
                "created_at": {"type": "string"},
            },
        }
        response = client.post("/api/v1/orders", json={
            "user_id": "user-123",
            "items": [{"product_id": "prod-456", "quantity": 1}],
        })
        validate(response.json(), expected_schema)

# レート制限の実装例 (FastAPI + Redis)
import redis
from fastapi import Request, HTTPException
import time
 
redis_client = redis.Redis(host="localhost", port=6379, db=0)
 
class RateLimiter:
    """トークンバケツ方式のレート制限"""
 
    def __init__(self, rate: int, per: int):
        """
        rate: 許可するリクエスト数
        per: 時間窓（秒）
        """
        self.rate = rate
        self.per = per
 
    async def check(self, key: str) -> tuple[bool, dict]:
        """レート制限チェック"""
        now = time.time()
        pipe = redis_client.pipeline()
 
        # スライディングウィンドウ
        window_start = now - self.per
        pipe.zremrangebyscore(key, 0, window_start)  # 古いエントリを削除
        pipe.zadd(key, {str(now): now})               # 現在のリクエストを追加
        pipe.zcard(key)                                # ウィンドウ内のリクエスト数
        pipe.expire(key, self.per)                     # TTL設定
        _, _, count, _ = pipe.execute()
 
        remaining = max(0, self.rate - count)
        headers = {
            "X-RateLimit-Limit": str(self.rate),
            "X-RateLimit-Remaining": str(remaining),
            "X-RateLimit-Reset": str(int(now + self.per)),
        }
 
        if count > self.rate:
            headers["Retry-After"] = str(self.per)
            return False, headers
        return True, headers
 
# ミドルウェアとして適用
rate_limiter = RateLimiter(rate=100, per=60)  # 60秒に100リクエスト
 
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # クライアント識別（APIキー or IP）
    client_id = request.headers.get("X-API-Key") or request.client.host
    key = f"rate_limit:{client_id}"
 
    allowed, headers = await rate_limiter.check(key)
 
    if not allowed:
        return JSONResponse(
            status_code=429,
            content={
                "type": "https://api.example.com/errors/rate_limit_exceeded",
                "title": "Rate Limit Exceeded",
                "status": 429,
                "detail": f"{rate_limiter.per}秒間に{rate_limiter.rate}リクエストまで",
            },
            headers=headers,
        )
 
    response = await call_next(request)
    for k, v in headers.items():
        response.headers[k] = v
    return response

// GraphQL スキーマ定義
const typeDefs = `
  type User {
    id: ID!
    name: String!
    email: String!
    orders(first: Int, after: String): OrderConnection!
  }
 
  type Order {
    id: ID!
    status: OrderStatus!
    totalAmount: Int!
    items: [OrderItem!]!
    createdAt: String!
  }
 
  type OrderItem {
    product: Product!
    quantity: Int!
    price: Int!
  }
 
  type Product {
    id: ID!
    name: String!
    price: Int!
  }
 
  type OrderConnection {
    edges: [OrderEdge!]!
    pageInfo: PageInfo!
  }
 
  type OrderEdge {
    cursor: String!
    node: Order!
  }
 
  type PageInfo {
    hasNextPage: Boolean!
    endCursor: String
  }
 
  enum OrderStatus {
    PENDING
    CONFIRMED
    SHIPPED
    DELIVERED
    CANCELLED
  }
 
  type Query {
    user(id: ID!): User
    orders(userId: ID!, first: Int, after: String): OrderConnection!
  }
 
  type Mutation {
    createOrder(input: CreateOrderInput!): Order!
    cancelOrder(id: ID!): Order!
  }
 
  input CreateOrderInput {
    userId: ID!
    items: [OrderItemInput!]!
  }
 
  input OrderItemInput {
    productId: ID!
    quantity: Int!
  }
`;
 
// リゾルバ（N+1問題をDataLoaderで解決）
import DataLoader from "dataloader";
 
const productLoader = new DataLoader<string, Product>(async (ids) => {
  const products = await db.products.findMany({ where: { id: { in: ids as string[] } } });
  const productMap = new Map(products.map(p => [p.id, p]));
  return ids.map(id => productMap.get(id)!);
});
 
const resolvers = {
  OrderItem: {
    product: (item: OrderItem) => productLoader.load(item.productId),
  },
};

// order_service.proto
syntax = "proto3";
 
package order.v1;
 
service OrderService {
  // 単項RPC: 注文作成
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);
 
  // 単項RPC: 注文取得
  rpc GetOrder(GetOrderRequest) returns (Order);
 
  // サーバーストリーミング: 注文ステータスの監視
  rpc WatchOrderStatus(WatchOrderStatusRequest) returns (stream OrderStatusUpdate);
 
  // クライアントストリーミング: バルク注文作成
  rpc BulkCreateOrders(stream CreateOrderRequest) returns (BulkCreateOrdersResponse);
}
 
message CreateOrderRequest {
  string user_id = 1;
  repeated OrderItem items = 2;
  string note = 3;
}
 
message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
}
 
message Order {
  string id = 1;
  string user_id = 2;
  OrderStatus status = 3;
  int32 total_amount = 4;
  repeated OrderItem items = 5;
  google.protobuf.Timestamp created_at = 6;
}
 
enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_CONFIRMED = 2;
  ORDER_STATUS_SHIPPED = 3;
  ORDER_STATUS_DELIVERED = 4;
  ORDER_STATUS_CANCELLED = 5;
}

# HATEOAS: レスポンスにナビゲーション用のリンクを含める
{
    "data": {
        "id": "ord-123",
        "status": "pending",
        "total_amount": 3500,
        "_links": {
            "self": {"href": "/api/v1/orders/ord-123"},
            "cancel": {"href": "/api/v1/orders/ord-123/cancel", "method": "POST"},
            "items": {"href": "/api/v1/orders/ord-123/items"},
            "user": {"href": "/api/v1/users/usr-456"},
        }
    }
}
 
# ステータスによって利用可能なアクションが変わる
# status: "shipped" の場合 → cancel リンクは含まれない
# status: "delivered" の場合 → return リンクが追加される

// BAD: エンドポイントごとに形式が異なる
// GET /users     → [{"id": 1, "name": "Alice"}]
// GET /orders    → {"results": [{"id": 1}], "count": 10}
// GET /products  → {"data": {"items": [...]}}
 
// GOOD: 統一された Envelope 形式
// GET /users
{
  "data": [{"id": 1, "name": "Alice"}],
  "pagination": {"next_cursor": "...", "has_next": true}
}
// GET /orders
{
  "data": [{"id": 1, "status": "placed"}],
  "pagination": {"next_cursor": "...", "has_next": false}
}

# BAD: DB のカラム名やテーブル構造がそのままレスポンスに漏れる
{
    "user_tbl_id": 123,           # テーブル名が露出
    "usr_pwd_hash": "abc123...",  # パスワードハッシュが露出
    "created_ts": 1710489600,     # 内部タイムスタンプ形式
    "is_del_flg": 0,              # 内部フラグ
}
 
# GOOD: API 用の DTO に変換してレスポンス
{
    "id": "usr-123",
    "name": "Alice",
    "email": "alice@example.com",
    "created_at": "2025-03-15T10:00:00Z",
}

// 各シナリオごとの JSON レスポンス（status, type, title, detail, errors）

// 1. 在庫不足 (409 Conflict)
{
  "type": "https://api.bookstore.com/errors/insufficient_stock",
  "title": "Insufficient Stock",
  "status": 409,
  "detail": "書籍「Effective Java」の在庫が不足しています（要求: 3, 在庫: 1）",
  "instance": "/api/v1/orders"
}
 
// 2. 重複レビュー (409 Conflict)
{
  "type": "https://api.bookstore.com/errors/duplicate_review",
  "title": "Duplicate Review",
  "status": 409,
  "detail": "この書籍に対するレビューは既に投稿済みです",
  "instance": "/api/v1/books/book-123/reviews"
}
 
// 3. トークン期限切れ (401 Unauthorized)
{
  "type": "https://api.bookstore.com/errors/token_expired",
  "title": "Token Expired",
  "status": 401,
  "detail": "認証トークンの有効期限が切れています。再ログインしてください"
}
 
// 4. バリデーションエラー (422 Unprocessable Entity)
{
  "type": "https://api.bookstore.com/errors/validation_error",
  "title": "Validation Error",
  "status": 422,
  "detail": "入力データに2件の問題があります",
  "instance": "/api/v1/books",
  "errors": [
    {"field": "title", "message": "タイトルは1〜200文字で入力してください", "code": "string_too_short"},
    {"field": "price", "message": "価格は0以上の整数で入力してください", "code": "value_error"}
  ]
}

# 1. 内部は v2 のデータモデルに統一
class UserModel:
    id: int
    first_name: str
    last_name: str
    email: str
 
# 2. v1 互換レイヤー（アダプター）
@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: int, response: Response):
    user = await user_service.get(user_id)
 
    # 廃止通知ヘッダー
    response.headers["Sunset"] = "Sat, 01 Sep 2026 00:00:00 GMT"
    response.headers["Deprecation"] = "true"
    response.headers["Link"] = (
        '<https://api.example.com/docs/v1-to-v2-migration>; rel="deprecation"'
    )
 
    # v1 形式に変換
    return {
        "id": user.id,
        "name": f"{user.first_name} {user.last_name}",  # 後方互換
        "email": user.email,
    }
 
# 3. v2 エンドポイント
@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: int):
    user = await user_service.get(user_id)
    return {
        "id": user.id,
        "first_name": user.first_name,
        "last_name": user.last_name,
        "email": user.email,
    }
 
# 4. 廃止スケジュール
"""
Month 1: v2 リリース + v1 Deprecation ヘッダー追加
Month 2: クライアントにメール通知 + ドキュメントに移行ガイド掲載
Month 3: v1 の使用状況をモニタリング（アクセスログ分析）
Month 4: 使用率の高いクライアントに個別通知
Month 5: v1 レスポンスに warning フィールド追加
Month 6: v1 を 410 Gone にして廃止
"""
 
# 5. v1 使用状況のモニタリング
@app.middleware("http")
async def track_api_version(request: Request, call_next):
    if request.url.path.startswith("/api/v1/"):
        client_id = request.headers.get("X-API-Key", "unknown")
        logger.warning(f"Deprecated v1 API access: {client_id} -> {request.url.path}")
        # メトリクスに記録
        metrics.increment("api.v1.deprecated_access", tags={"client": client_id})
    return await call_next(request)

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]

// 推奨: null を明示
{"name": "Alice", "nickname": null, "avatar_url": null}
 
// 非推奨: フィールド省略（nickname があるのかないのか不明）
{"name": "Alice"}