# コード例1: 要件定義を構造化して管理する例
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import json
 
 
class RequirementPriority(Enum):
    """MoSCoW 優先度分類"""
    MUST = "Must Have"      # 必須要件
    SHOULD = "Should Have"  # 重要だが必須ではない
    COULD = "Could Have"    # あれば望ましい
    WONT = "Won't Have"     # 今回は対象外
 
 
class RequirementType(Enum):
    FUNCTIONAL = "Functional"          # 機能要件
    NON_FUNCTIONAL = "Non-Functional"  # 非機能要件
    CONSTRAINT = "Constraint"          # 制約事項
 
 
@dataclass
class Requirement:
    """要件を表すデータクラス"""
    id: str                              # 例: "REQ-001"
    title: str                           # 要件タイトル
    description: str                     # 詳細説明
    type: RequirementType                # 要件種別
    priority: RequirementPriority        # 優先度
    acceptance_criteria: list[str]       # 受入基準
    source: str = ""                     # 要件の出所（ステークホルダー名等）
    dependencies: list[str] = field(default_factory=list)  # 依存する要件ID
    status: str = "Draft"               # Draft / Approved / Implemented / Verified
    rationale: Optional[str] = None      # この要件が必要な理由
 
    def is_testable(self) -> bool:
        """受入基準が定義されているかを検証"""
        return len(self.acceptance_criteria) > 0
 
    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "type": self.type.value,
            "priority": self.priority.value,
            "acceptance_criteria": self.acceptance_criteria,
            "dependencies": self.dependencies,
            "status": self.status,
        }
 
 
class RequirementsDocument:
    """要件定義書を管理するクラス"""
 
    def __init__(self, project_name: str):
        self.project_name = project_name
        self.requirements: dict[str, Requirement] = {}
        self._next_id = 1
 
    def add_requirement(self, title: str, description: str,
                        req_type: RequirementType,
                        priority: RequirementPriority,
                        acceptance_criteria: list[str],
                        **kwargs) -> Requirement:
        req_id = f"REQ-{self._next_id:03d}"
        self._next_id += 1
        req = Requirement(
            id=req_id, title=title, description=description,
            type=req_type, priority=priority,
            acceptance_criteria=acceptance_criteria, **kwargs
        )
        self.requirements[req_id] = req
        return req
 
    def get_by_priority(self, priority: RequirementPriority) -> list[Requirement]:
        return [r for r in self.requirements.values() if r.priority == priority]
 
    def validate_all(self) -> list[str]:
        """全要件の整合性を検証"""
        issues = []
        for req in self.requirements.values():
            if not req.is_testable():
                issues.append(f"{req.id}: 受入基準が未定義")
            for dep in req.dependencies:
                if dep not in self.requirements:
                    issues.append(f"{req.id}: 依存先 {dep} が存在しない")
        return issues
 
    def coverage_report(self) -> dict:
        """要件カバレッジレポートを生成"""
        total = len(self.requirements)
        if total == 0:
            return {"total": 0}
        by_status = {}
        for req in self.requirements.values():
            by_status[req.status] = by_status.get(req.status, 0) + 1
        return {
            "total": total,
            "by_status": by_status,
            "must_have_count": len(self.get_by_priority(RequirementPriority.MUST)),
            "verified_ratio": by_status.get("Verified", 0) / total,
        }
 
    def export_json(self) -> str:
        return json.dumps(
            {
                "project": self.project_name,
                "requirements": [r.to_dict() for r in self.requirements.values()],
            },
            indent=2, ensure_ascii=False
        )
 
 
# 使用例
doc = RequirementsDocument("ECサイト構築プロジェクト")
 
doc.add_requirement(
    title="ユーザー登録機能",
    description="メールアドレスとパスワードでユーザー登録ができる",
    req_type=RequirementType.FUNCTIONAL,
    priority=RequirementPriority.MUST,
    acceptance_criteria=[
        "メールアドレスの形式バリデーションが行われる",
        "パスワードは8文字以上で英数字記号を含む",
        "登録完了後に確認メールが送信される",
        "既存メールアドレスでの重複登録はエラーとなる",
    ],
    rationale="サービス利用の前提となる基本機能"
)
 
doc.add_requirement(
    title="レスポンスタイム",
    description="全ページの表示が2秒以内に完了する",
    req_type=RequirementType.NON_FUNCTIONAL,
    priority=RequirementPriority.MUST,
    acceptance_criteria=[
        "95パーセンタイルで応答時間が2秒以内",
        "同時接続1000ユーザーの負荷下で達成",
    ]
)
 
# 検証
issues = doc.validate_all()
report = doc.coverage_report()
print(f"要件数: {report['total']}, Must Have: {report['must_have_count']}")
print(f"検証済み比率: {report['verified_ratio']:.0%}")

# コード例2: スクラムボードのシンプルな実装
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timedelta
from typing import Optional
 
 
class StoryStatus(Enum):
    TODO = "To Do"
    IN_PROGRESS = "In Progress"
    IN_REVIEW = "In Review"
    DONE = "Done"
 
 
class StorySize(Enum):
    """ストーリーポイント（フィボナッチ数列）"""
    XS = 1
    S = 2
    M = 3
    L = 5
    XL = 8
    XXL = 13
 
 
@dataclass
class UserStory:
    """ユーザーストーリー"""
    id: str
    title: str
    description: str        # "As a [user], I want [goal], so that [benefit]"
    story_points: StorySize
    acceptance_criteria: list[str]
    status: StoryStatus = StoryStatus.TODO
    assignee: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None
 
    @property
    def points(self) -> int:
        return self.story_points.value
 
    def start(self, assignee: str) -> None:
        self.status = StoryStatus.IN_PROGRESS
        self.assignee = assignee
 
    def submit_for_review(self) -> None:
        if self.status != StoryStatus.IN_PROGRESS:
            raise ValueError("レビュー提出は作業中のストーリーのみ可能")
        self.status = StoryStatus.IN_REVIEW
 
    def complete(self) -> None:
        self.status = StoryStatus.DONE
        self.completed_at = datetime.now()
 
 
@dataclass
class Sprint:
    """スプリント"""
    id: str
    goal: str
    start_date: datetime
    duration_weeks: int = 2
    stories: list[UserStory] = field(default_factory=list)
 
    @property
    def end_date(self) -> datetime:
        return self.start_date + timedelta(weeks=self.duration_weeks)
 
    @property
    def total_points(self) -> int:
        return sum(s.points for s in self.stories)
 
    @property
    def completed_points(self) -> int:
        return sum(s.points for s in self.stories if s.status == StoryStatus.DONE)
 
    @property
    def velocity(self) -> int:
        """このスプリントのベロシティ（完了ポイント数）"""
        return self.completed_points
 
    def add_story(self, story: UserStory) -> None:
        self.stories.append(story)
 
    def burndown_data(self) -> list[dict]:
        """バーンダウンチャートのデータを生成"""
        remaining = self.total_points
        data = [{"day": 0, "remaining": remaining, "ideal": remaining}]
        total_days = self.duration_weeks * 5  # 営業日
        daily_ideal = remaining / total_days
 
        completed_stories = sorted(
            [s for s in self.stories if s.completed_at],
            key=lambda s: s.completed_at
        )
 
        for day in range(1, total_days + 1):
            for story in completed_stories:
                days_elapsed = (story.completed_at - self.start_date).days
                if days_elapsed == day:
                    remaining -= story.points
            data.append({
                "day": day,
                "remaining": remaining,
                "ideal": max(0, self.total_points - daily_ideal * day),
            })
        return data
 
    def summary(self) -> str:
        status_counts = {}
        for story in self.stories:
            key = story.status.value
            status_counts[key] = status_counts.get(key, 0) + 1
        lines = [
            f"Sprint: {self.id} - {self.goal}",
            f"期間: {self.start_date.date()} 〜 {self.end_date.date()}",
            f"合計ポイント: {self.total_points}, 完了: {self.completed_points}",
            f"ベロシティ: {self.velocity}",
            "ステータス別:"
        ]
        for status, count in status_counts.items():
            lines.append(f"  {status}: {count}")
        return "\n".join(lines)
 
 
class ScrumBoard:
    """スクラムボード（複数スプリント管理）"""
 
    def __init__(self, team_name: str):
        self.team_name = team_name
        self.sprints: list[Sprint] = []
        self.product_backlog: list[UserStory] = []
 
    def add_to_backlog(self, story: UserStory) -> None:
        self.product_backlog.append(story)
 
    def create_sprint(self, sprint_id: str, goal: str,
                      start_date: datetime) -> Sprint:
        sprint = Sprint(id=sprint_id, goal=goal, start_date=start_date)
        self.sprints.append(sprint)
        return sprint
 
    def plan_sprint(self, sprint: Sprint, story_ids: list[str]) -> None:
        """スプリントプランニング: バックログからストーリーを選択"""
        for sid in story_ids:
            for i, story in enumerate(self.product_backlog):
                if story.id == sid:
                    sprint.add_story(story)
                    self.product_backlog.pop(i)
                    break
 
    def average_velocity(self, last_n: int = 3) -> float:
        """直近 n スプリントの平均ベロシティ"""
        recent = self.sprints[-last_n:]
        if not recent:
            return 0.0
        return sum(s.velocity for s in recent) / len(recent)
 
 
# 使用例
board = ScrumBoard("開発チームA")
 
# バックログにストーリーを追加
stories = [
    UserStory("US-001", "ログイン機能",
              "As a user, I want to log in, so that I can access my account",
              StorySize.M,
              ["メールとパスワードでログインできる", "エラー時にメッセージが表示される"]),
    UserStory("US-002", "商品検索",
              "As a user, I want to search products, so that I can find what I need",
              StorySize.L,
              ["キーワード検索ができる", "カテゴリでフィルタリングできる"]),
    UserStory("US-003", "カート機能",
              "As a user, I want to add items to cart, so that I can purchase multiple items",
              StorySize.XL,
              ["商品をカートに追加できる", "数量を変更できる", "カートから削除できる"]),
]
for s in stories:
    board.add_to_backlog(s)
 
# スプリント作成と計画
sprint1 = board.create_sprint("Sprint-1", "基本認証機能の実装", datetime.now())
board.plan_sprint(sprint1, ["US-001", "US-002"])
 
# 作業の進行
sprint1.stories[0].start("田中")
sprint1.stories[0].submit_for_review()
sprint1.stories[0].complete()
 
print(sprint1.summary())

# コード例3: TDD の Red-Green-Refactor サイクル
 
# === Step 1: Red（失敗するテストを書く） ===
import unittest
 
 
class TestShoppingCart(unittest.TestCase):
    """ショッピングカートのテスト"""
 
    def test_empty_cart_has_zero_total(self):
        cart = ShoppingCart()
        self.assertEqual(cart.total(), 0)
 
    def test_add_single_item(self):
        cart = ShoppingCart()
        cart.add_item("りんご", price=150, quantity=1)
        self.assertEqual(cart.total(), 150)
 
    def test_add_multiple_items(self):
        cart = ShoppingCart()
        cart.add_item("りんご", price=150, quantity=2)
        cart.add_item("バナナ", price=100, quantity=3)
        self.assertEqual(cart.total(), 600)  # 150*2 + 100*3
 
    def test_apply_percentage_discount(self):
        cart = ShoppingCart()
        cart.add_item("りんご", price=1000, quantity=1)
        cart.apply_discount(percent=10)
        self.assertEqual(cart.total(), 900)
 
    def test_remove_item(self):
        cart = ShoppingCart()
        cart.add_item("りんご", price=150, quantity=1)
        cart.remove_item("りんご")
        self.assertEqual(cart.total(), 0)
 
    def test_item_count(self):
        cart = ShoppingCart()
        cart.add_item("りんご", price=150, quantity=2)
        cart.add_item("バナナ", price=100, quantity=3)
        self.assertEqual(cart.item_count(), 5)
 
 
# === Step 2: Green（テストを通す最小限の実装） ===
 
@dataclass
class CartItem:
    name: str
    price: int
    quantity: int
 
    @property
    def subtotal(self) -> int:
        return self.price * self.quantity
 
 
class ShoppingCart:
    def __init__(self):
        self._items: dict[str, CartItem] = {}
        self._discount_percent: int = 0
 
    def add_item(self, name: str, price: int, quantity: int) -> None:
        if name in self._items:
            self._items[name].quantity += quantity
        else:
            self._items[name] = CartItem(name=name, price=price, quantity=quantity)
 
    def remove_item(self, name: str) -> None:
        self._items.pop(name, None)
 
    def apply_discount(self, percent: int) -> None:
        if not 0 <= percent <= 100:
            raise ValueError("割引率は0〜100の範囲で指定")
        self._discount_percent = percent
 
    def item_count(self) -> int:
        return sum(item.quantity for item in self._items.values())
 
    def subtotal(self) -> int:
        return sum(item.subtotal for item in self._items.values())
 
    def total(self) -> int:
        sub = self.subtotal()
        discount = sub * self._discount_percent // 100
        return sub - discount
 
 
# === Step 3: Refactor（コードを改善、テストは通るまま） ===
# 上記の実装は既にリファクタリング済み:
# - CartItem を独立したデータクラスに分離
# - subtotal 計算をプロパティ化
# - 割引ロジックを明確に分離

# コード例4: GitHub Actions による CI/CD パイプライン設定
name: CI/CD Pipeline
 
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
 
env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}
 
jobs:
  # ===== CI: ビルドとテスト =====
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install dependencies
        run: |
          pip install ruff mypy
          pip install -r requirements.txt
      - name: Lint with ruff
        run: ruff check .
      - name: Type check with mypy
        run: mypy src/
 
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12"]
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: test
          POSTGRES_DB: testdb
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-dev.txt
      - name: Run unit tests
        run: pytest tests/unit/ -v --cov=src --cov-report=xml
      - name: Run integration tests
        run: pytest tests/integration/ -v
        env:
          DATABASE_URL: postgresql://postgres:test@localhost:5432/testdb
      - name: Upload coverage
        uses: codecov/codecov-action@v4
        with:
          file: coverage.xml
 
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: fs
          scan-ref: .
          severity: CRITICAL,HIGH
 
  # ===== CD: ビルドとデプロイ =====
  build-and-push:
    needs: [lint, test, security-scan]
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v4
      - name: Log in to Container Registry
        uses: docker/login-action@v3
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - name: Build and push Docker image
        uses: docker/build-push-action@v5
        with:
          context: .
          push: true
          tags: |
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
 
  deploy-staging:
    needs: build-and-push
    runs-on: ubuntu-latest
    environment: staging
    steps:
      - name: Deploy to staging
        run: |
          echo "Deploying ${{ github.sha }} to staging environment"
          # kubectl set image deployment/app \
          #   app=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
 
  deploy-production:
    needs: deploy-staging
    runs-on: ubuntu-latest
    environment:
      name: production
      url: https://app.example.com
    steps:
      - name: Deploy to production
        run: |
          echo "Deploying ${{ github.sha }} to production environment"
          # 本番デプロイのコマンド

# ADR-001: API 通信に GraphQL を採用する
 
## ステータス
承認済み（2024-01-15）
 
## コンテキスト
フロントエンドが必要とするデータは画面ごとに異なり、
REST API では過剰取得（Over-fetching）や不足取得（Under-fetching）が
頻繁に発生している。モバイルアプリも計画されており、
帯域幅の効率的な利用が求められる。
 
## 決定
API 通信プロトコルとして GraphQL を採用する。
サーバー実装には Apollo Server を使用する。
 
## 理由
- クライアントが必要なデータのみを取得できる
- 型システムによるスキーマ定義で API の契約が明確になる
- 複数のデータソースを統合するリゾルバーパターンが使える
- モバイル・Web で同一の API を効率的に利用できる
 
## 結果
- チームは GraphQL の学習コストを負担する必要がある
- N+1 問題への対策（DataLoader）が必要
- キャッシュ戦略が REST と異なるため再設計が必要
- パフォーマンスモニタリングツールの選定が必要

# コード例5: プロジェクト管理ツールの実装例
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import math
 
 
class TaskStatus(Enum):
    BACKLOG = "Backlog"
    TODO = "To Do"
    IN_PROGRESS = "In Progress"
    DONE = "Done"
    BLOCKED = "Blocked"
 
 
@dataclass
class ThreePointEstimate:
    """三点見積もり"""
    optimistic: float     # 楽観値（日）
    most_likely: float    # 最頻値（日）
    pessimistic: float    # 悲観値（日）
 
    @property
    def expected(self) -> float:
        """PERT 期待値"""
        return (self.optimistic + 4 * self.most_likely + self.pessimistic) / 6
 
    @property
    def standard_deviation(self) -> float:
        """標準偏差"""
        return (self.pessimistic - self.optimistic) / 6
 
    @property
    def variance(self) -> float:
        """分散"""
        return self.standard_deviation ** 2
 
    def confidence_interval(self, confidence: float = 0.95) -> tuple[float, float]:
        """信頼区間を計算（正規分布近似）"""
        # 95% 信頼区間 → z = 1.96
        z_values = {0.90: 1.645, 0.95: 1.96, 0.99: 2.576}
        z = z_values.get(confidence, 1.96)
        margin = z * self.standard_deviation
        return (self.expected - margin, self.expected + margin)
 
 
@dataclass
class Task:
    """プロジェクトタスク"""
    id: str
    name: str
    estimate: Optional[ThreePointEstimate] = None
    status: TaskStatus = TaskStatus.BACKLOG
    assignee: Optional[str] = None
    dependencies: list[str] = field(default_factory=list)
    actual_days: Optional[float] = None
 
    @property
    def estimated_days(self) -> float:
        if self.estimate:
            return self.estimate.expected
        return 0.0
 
 
class ProjectPlanner:
    """プロジェクト計画ツール"""
 
    def __init__(self, project_name: str):
        self.project_name = project_name
        self.tasks: dict[str, Task] = {}
 
    def add_task(self, task: Task) -> None:
        self.tasks[task.id] = task
 
    def total_estimate(self) -> ThreePointEstimate:
        """全タスクの合計見積もり"""
        total_o = sum(t.estimate.optimistic for t in self.tasks.values() if t.estimate)
        total_m = sum(t.estimate.most_likely for t in self.tasks.values() if t.estimate)
        total_p = sum(t.estimate.pessimistic for t in self.tasks.values() if t.estimate)
        return ThreePointEstimate(total_o, total_m, total_p)
 
    def monte_carlo_simulation(self, iterations: int = 10000) -> dict:
        """モンテカルロシミュレーションによる完了日予測"""
        import random
 
        results = []
        tasks_with_estimates = [t for t in self.tasks.values() if t.estimate]
 
        for _ in range(iterations):
            total = 0.0
            for task in tasks_with_estimates:
                est = task.estimate
                # PERT 分布を三角分布で近似
                sample = random.triangular(
                    est.optimistic, est.pessimistic, est.most_likely
                )
                total += sample
            results.append(total)
 
        results.sort()
        return {
            "mean": sum(results) / len(results),
            "median": results[len(results) // 2],
            "p50": results[int(len(results) * 0.50)],
            "p75": results[int(len(results) * 0.75)],
            "p85": results[int(len(results) * 0.85)],
            "p95": results[int(len(results) * 0.95)],
            "min": results[0],
            "max": results[-1],
        }
 
    def velocity_forecast(self, velocity_history: list[int],
                          remaining_points: int) -> dict:
        """ベロシティに基づくリリース予測"""
        if not velocity_history:
            return {"error": "ベロシティ履歴が必要"}
 
        avg_velocity = sum(velocity_history) / len(velocity_history)
        std_dev = (
            sum((v - avg_velocity) ** 2 for v in velocity_history)
            / len(velocity_history)
        ) ** 0.5
 
        sprints_needed = math.ceil(remaining_points / avg_velocity) if avg_velocity > 0 else float('inf')
        optimistic = math.ceil(remaining_points / (avg_velocity + std_dev)) if (avg_velocity + std_dev) > 0 else float('inf')
        pessimistic = math.ceil(remaining_points / max(avg_velocity - std_dev, 1))
 
        return {
            "average_velocity": round(avg_velocity, 1),
            "remaining_points": remaining_points,
            "sprints_needed": {
                "optimistic": optimistic,
                "expected": sprints_needed,
                "pessimistic": pessimistic,
            },
        }
 
    def progress_report(self) -> dict:
        """進捗レポート"""
        total = len(self.tasks)
        done = sum(1 for t in self.tasks.values() if t.status == TaskStatus.DONE)
        in_progress = sum(1 for t in self.tasks.values() if t.status == TaskStatus.IN_PROGRESS)
        blocked = sum(1 for t in self.tasks.values() if t.status == TaskStatus.BLOCKED)
 
        estimated_total = sum(t.estimated_days for t in self.tasks.values())
        actual_total = sum(
            t.actual_days for t in self.tasks.values()
            if t.actual_days is not None
        )
        done_estimated = sum(
            t.estimated_days for t in self.tasks.values()
            if t.status == TaskStatus.DONE
        )
 
        return {
            "total_tasks": total,
            "done": done,
            "in_progress": in_progress,
            "blocked": blocked,
            "completion_rate": f"{done / total * 100:.1f}%" if total > 0 else "N/A",
            "estimated_total_days": round(estimated_total, 1),
            "actual_total_days": round(actual_total, 1),
            "estimation_accuracy": (
                f"{actual_total / done_estimated * 100:.1f}%"
                if done_estimated > 0 else "N/A"
            ),
        }
 
 
# 使用例
planner = ProjectPlanner("ECサイトリニューアル")
 
planner.add_task(Task(
    id="T-001", name="認証システム",
    estimate=ThreePointEstimate(3, 5, 10),
    status=TaskStatus.DONE, actual_days=6
))
planner.add_task(Task(
    id="T-002", name="商品検索API",
    estimate=ThreePointEstimate(5, 8, 15),
    status=TaskStatus.IN_PROGRESS, assignee="鈴木"
))
planner.add_task(Task(
    id="T-003", name="決済連携",
    estimate=ThreePointEstimate(8, 12, 20),
    status=TaskStatus.TODO, dependencies=["T-001"]
))
 
# 合計見積もり
total = planner.total_estimate()
print(f"合計期待値: {total.expected:.1f} 日")
lower, upper = total.confidence_interval(0.95)
print(f"95%信頼区間: {lower:.1f} 〜 {upper:.1f} 日")
 
# モンテカルロシミュレーション
mc = planner.monte_carlo_simulation()
print(f"P85 完了予測: {mc['p85']:.1f} 日")
 
# 進捗レポート
report = planner.progress_report()
print(f"進捗: {report['completion_rate']}")

# コード例6: フィーチャーフラグの実装パターン
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import hashlib
import time
 
 
class FlagType(Enum):
    """フィーチャーフラグの種類"""
    RELEASE = "release"          # リリースフラグ（新機能の段階的公開）
    EXPERIMENT = "experiment"    # 実験フラグ（A/Bテスト）
    OPS = "ops"                  # 運用フラグ（機能の緊急停止）
    PERMISSION = "permission"    # 権限フラグ（特定ユーザーへの公開）
 
 
@dataclass
class FeatureFlag:
    """フィーチャーフラグ"""
    name: str
    flag_type: FlagType
    enabled: bool = False
    description: str = ""
    rollout_percentage: int = 0        # 段階的公開の割合 (0-100)
    allowed_users: list[str] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)
    expires_at: Optional[float] = None  # 有効期限
 
    def is_expired(self) -> bool:
        if self.expires_at is None:
            return False
        return time.time() > self.expires_at
 
 
class FeatureFlagManager:
    """フィーチャーフラグ管理"""
 
    def __init__(self):
        self._flags: dict[str, FeatureFlag] = {}
 
    def register(self, flag: FeatureFlag) -> None:
        self._flags[flag.name] = flag
 
    def is_enabled(self, flag_name: str, user_id: Optional[str] = None) -> bool:
        """指定したフラグが有効かを判定"""
        flag = self._flags.get(flag_name)
        if flag is None:
            return False
 
        # 期限切れチェック
        if flag.is_expired():
            return False
 
        # グローバル無効
        if not flag.enabled:
            return False
 
        # ユーザー固有の判定
        if user_id:
            # 許可リストに含まれるユーザーは常に有効
            if user_id in flag.allowed_users:
                return True
 
            # 段階的ロールアウト（ユーザーIDのハッシュで判定）
            if flag.rollout_percentage < 100:
                hash_val = int(hashlib.md5(
                    f"{flag_name}:{user_id}".encode()
                ).hexdigest(), 16)
                return (hash_val % 100) < flag.rollout_percentage
 
        return True
 
    def get_all_flags(self) -> dict[str, dict]:
        """全フラグの状態を返す"""
        return {
            name: {
                "enabled": flag.enabled,
                "type": flag.flag_type.value,
                "rollout": flag.rollout_percentage,
                "expired": flag.is_expired(),
            }
            for name, flag in self._flags.items()
        }
 
    def cleanup_expired(self) -> list[str]:
        """期限切れフラグを除去"""
        expired = [name for name, f in self._flags.items() if f.is_expired()]
        for name in expired:
            del self._flags[name]
        return expired
 
 
# 使用例
manager = FeatureFlagManager()
 
# 新機能の段階的公開
manager.register(FeatureFlag(
    name="new_search_ui",
    flag_type=FlagType.RELEASE,
    enabled=True,
    description="新しい検索UIを段階的に公開",
    rollout_percentage=20,  # まず20%のユーザーに公開
))
 
# 緊急停止用フラグ
manager.register(FeatureFlag(
    name="payment_processing",
    flag_type=FlagType.OPS,
    enabled=True,
    description="決済処理の有効/無効切り替え",
    rollout_percentage=100,
))
 
# アプリケーションコードでの使用
def search_handler(user_id: str, query: str):
    if manager.is_enabled("new_search_ui", user_id):
        return new_search(query)  # 新しい検索ロジック
    else:
        return legacy_search(query)  # 既存の検索ロジック
 
def new_search(query: str) -> dict:
    return {"engine": "new", "query": query}
 
def legacy_search(query: str) -> dict:
    return {"engine": "legacy", "query": query}
 
# ユーザーごとに異なる結果
print(manager.is_enabled("new_search_ui", "user_001"))  # ハッシュ次第でTrue/False
print(manager.is_enabled("new_search_ui", "user_002"))