# 計算的思考を適用したAPIパフォーマンス改善
 
# 1. 分解: 遅いAPIエンドポイントをプロファイリングで分解
# → DBクエリ: 800ms / ビジネスロジック: 50ms / シリアライズ: 150ms
# → DBクエリがボトルネックと特定
 
# 2. パターン認識: 同様の遅いAPIが他にもある
# → 全てN+1クエリが原因というパターンを発見
 
# 3. 抽象化: 個々のAPIの詳細ではなく、ORM使用パターンに着目
# → prefetch_related / select_related の体系的な適用
 
# 4. アルゴリズム的思考: 改善手順の明確化
def optimize_api():
    """API最適化の標準手順"""
    # Step 1: プロファイリングでボトルネック特定
    profile = measure_endpoint_performance()
 
    # Step 2: N+1問題の自動検出
    n_plus_one = detect_n_plus_one_queries(profile)
 
    # Step 3: JOINまたはプリフェッチの適用
    for query in n_plus_one:
        apply_prefetch(query)
 
    # Step 4: 効果測定
    assert measure_improvement() > 0.5  # 50%以上改善

# 正規表現の限界を理解する例
import re
 
# ✅ 正規言語: 正規表現で表現可能
email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
re.match(email_pattern, 'user@example.com')  # マッチ
 
# ❌ 文脈自由言語: 正規表現では本質的に不可能
# 「対応する括弧のバランスチェック」
# ((()))  → OK
# (()     → NG
# 正規表現では括弧のネストの深さを追跡できない
# → パーサー（プッシュダウンオートマトン）が必要
 
# ❌ 停止問題: アルゴリズムで判定不可能
# 「任意のプログラムが無限ループするか判定する」ことは原理的に不可能
# → 完璧な静的解析ツールは作れない（近似は可能）

# ダイクストラのアルゴリズム — カーナビ、ネットワークルーティングの基盤
import heapq
 
def dijkstra(graph, start):
    """最短経路を求める: O((V+E) log V)"""
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]  # (距離, ノード) の優先度キュー
 
    while pq:
        current_dist, current = heapq.heappop(pq)
        if current_dist > distances[current]:
            continue
 
        for neighbor, weight in graph[current].items():
            distance = current_dist + weight
            if distance < distances[neighbor]:
                distances[neighbor] = distance
                heapq.heappush(pq, (distance, neighbor))
 
    return distances
 
# 使用例: サーバー間の最短レイテンシ経路
network = {
    'tokyo': {'osaka': 5, 'singapore': 30},
    'osaka': {'tokyo': 5, 'singapore': 25},
    'singapore': {'tokyo': 30, 'osaka': 25, 'sydney': 40},
    'sydney': {'singapore': 40}
}
print(dijkstra(network, 'tokyo'))
# {'tokyo': 0, 'osaka': 5, 'singapore': 30, 'sydney': 70}

# 場面別の最適なデータ構造選択
 
# 場面1: ユーザーIDの存在確認（頻繁な検索）
# → ハッシュセット O(1)
active_users = set()
active_users.add(user_id)
if user_id in active_users:  # O(1)
    pass
 
# 場面2: ランキング表示（順序付き、範囲検索）
# → ソート済みリスト or 平衡二分探索木
import sortedcontainers
ranking = sortedcontainers.SortedList(key=lambda x: -x['score'])
ranking.add({'name': 'Alice', 'score': 950})
top_10 = ranking[:10]  # 上位10件
 
# 場面3: Undo/Redo機能
# → スタック
undo_stack = []
redo_stack = []
def execute_action(action):
    undo_stack.append(action)
    redo_stack.clear()
    action.execute()
 
# 場面4: タスクキュー（FIFO処理）
# → キュー（deque）
from collections import deque
task_queue = deque()
task_queue.append(task)      # エンキュー O(1)
next_task = task_queue.popleft()  # デキュー O(1)
 
# 場面5: 文字列のオートコンプリート
# → トライ木（Prefix Tree）
class TrieNode:
    def __init__(self):
        self.children = {}
        self.is_end = False

# 「動けばいい」アプローチ
def find_user(users, target_id):
    for user in users:
        if user['id'] == target_id:
            return user
    return None

# 「なぜ遅いか」を理解した上でのアプローチ
# O(n) の線形探索 → O(1) のハッシュテーブルに変換
def build_user_index(users):
    """前処理: O(n) で辞書を構築"""
    return {user['id']: user for user in users}
 
def find_user(user_index, target_id):
    """検索: O(1) でアクセス"""
    return user_index.get(target_id)
 
# 10万人のユーザーから検索する場合:
# 線形探索: 平均50,000回の比較
# ハッシュ: 1回のハッシュ計算 + 1回のアクセス

# 再帰で直感的に実装
def fib(n):
    if n <= 1:
        return n
    return fib(n-1) + fib(n-2)
# fib(50) → 数分〜数時間かかる（O(2^n)）

# 動的計画法を適用（重複部分問題を認識）
def fib(n):
    if n <= 1:
        return n
    prev, curr = 0, 1
    for _ in range(2, n + 1):
        prev, curr = curr, prev + curr
    return curr
# fib(50) → 即座に完了（O(n)）
# さらに行列累乗法で O(log n) も可能

# キャッシュの設計をCSの原理から導く
 
# 1. 局所性の原理（メモリ階層の知識）
# → 時間的局所性: 最近アクセスしたデータは再アクセスされやすい
# → LRU (Least Recently Used) キャッシュが有効
 
from collections import OrderedDict
 
class LRUCache:
    """LRUキャッシュの実装 — OrderedDictを活用"""
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity
 
    def get(self, key):
        if key in self.cache:
            self.cache.move_to_end(key)  # 最近使用に移動 O(1)
            return self.cache[key]
        return None
 
    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # 最古を削除 O(1)
 
# 2. キャッシュ一貫性（分散システムの知識）
# → Write-Through: 書き込み時にキャッシュとDBの両方を更新
# → Write-Back: キャッシュのみ更新し、後でDBに反映
# → Cache-Aside: 読み込み時にキャッシュミスならDBから取得
 
# 3. キャッシュ無効化（計算理論の知識）
# → "There are only two hard things in CS:
#    cache invalidation and naming things." — Phil Karlton
# → TTL、イベント駆動無効化、バージョニングの使い分け

# アムダールの法則: 並列化の限界を事前に計算
 
def amdahl_speedup(parallel_fraction: float, num_processors: int) -> float:
    """アムダールの法則による理論上の最大速度向上を計算
 
    Args:
        parallel_fraction: 並列化可能な割合 (0.0 - 1.0)
        num_processors: プロセッサ数
 
    Returns:
        速度向上倍率
    """
    serial_fraction = 1 - parallel_fraction
    return 1 / (serial_fraction + parallel_fraction / num_processors)
 
# 例: プログラムの80%が並列化可能な場合
print(f"2コア: {amdahl_speedup(0.8, 2):.2f}倍")    # 1.67倍
print(f"4コア: {amdahl_speedup(0.8, 4):.2f}倍")    # 2.50倍
print(f"8コア: {amdahl_speedup(0.8, 8):.2f}倍")    # 3.33倍
print(f"∞コア: {amdahl_speedup(0.8, 10000):.2f}倍")  # 5.00倍（上限！）
# → コアを無限に増やしても5倍が限界（20%の直列部分がボトルネック）
# → 「並列化する前に直列部分を最適化せよ」がCSの教え

# Step 1: プロファイリングでボトルネック特定
import cProfile
 
def slow_endpoint():
    users = get_all_users()          # 0.1s
    orders = get_all_orders()        # 0.2s
    # ↓ ここが99%の時間を消費
    result = []
    for user in users:               # 10,000ユーザー
        user_orders = [o for o in orders if o['user_id'] == user['id']]
        # ↑ orders(100,000件)を毎回全走査 → O(10,000 × 100,000) = O(10^9)
        result.append({**user, 'orders': user_orders})
    return result
 
# Step 2: CS知識を適用
def fast_endpoint():
    users = get_all_users()
    orders = get_all_orders()
    # ハッシュマップでグルーピング: O(n)
    orders_by_user = {}
    for order in orders:
        orders_by_user.setdefault(order['user_id'], []).append(order)
    # 結合: O(m)
    return [{**user, 'orders': orders_by_user.get(user['id'], [])}
            for user in users]
    # 合計: O(n + m) = O(110,000) — 9,000倍高速

-- Before: フルテーブルスキャン 30秒
SELECT * FROM orders
WHERE user_id = 12345
  AND status = 'completed'
  AND created_at > '2025-01-01'
ORDER BY created_at DESC
LIMIT 20;
 
-- CS知識: B+木の構造を理解する
-- B+木は「ソート済み」のデータ構造
-- → 複合インデックスの列順序が重要
-- → 等値条件(=)の列を先に、範囲条件(>)の列を後に
 
-- After: 複合インデックス追加 0.01秒
CREATE INDEX idx_orders_user_status_created
  ON orders(user_id, status, created_at DESC);
 
-- なぜこの順序か:
-- 1. user_id = 12345 → 等値条件で絞り込み
-- 2. status = 'completed' → さらに等値条件で絞り込み
-- 3. created_at DESC → 範囲条件 + ソート（インデックスの順序で取得）
-- → Index Scan Only で完了（テーブルアクセス不要）

# CSで使う数学の具体例
 
import math
 
# 対数（アルゴリズム解析の基本）
# 二分探索で100万件を探索する回数
n = 1_000_000
steps = math.ceil(math.log2(n))  # 20回
print(f"二分探索のステップ数: {steps}")
 
# 組み合わせ論（パスワード強度の計算）
# 英数字(62文字)の8文字パスワードの組み合わせ数
chars = 62
length = 8
combinations = chars ** length  # 218兆
time_to_crack = combinations / 10_000_000_000  # 1秒1億回の試行
print(f"ブルートフォース所要時間: {time_to_crack / 3600:.0f}時間")
 
# 確率（ハッシュ衝突の計算 — 誕生日のパラドックス）
# n人中に同じ誕生日のペアがいる確率
def birthday_collision_probability(n, d=365):
    """n人がd種類のうち衝突する確率"""
    prob_no_collision = 1.0
    for i in range(n):
        prob_no_collision *= (d - i) / d
    return 1 - prob_no_collision
 
print(f"23人で衝突確率: {birthday_collision_probability(23):.1%}")  # 50.7%
# → ハッシュ空間が2^128でも、2^64個で50%の衝突確率

# デバッグ用ユーティリティ
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]

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]

# セキュアコーディング例
import hashlib
import secrets
import hmac
from typing import Optional
 
class SecurityUtils:
    """セキュリティユーティリティ"""
 
    @staticmethod
    def generate_token(length: int = 32) -> str:
        """暗号学的に安全なトークン生成"""
        return secrets.token_urlsafe(length)
 
    @staticmethod
    def hash_password(password: str, salt: Optional[str] = None) -> tuple:
        """パスワードのハッシュ化"""
        if salt is None:
            salt = secrets.token_hex(16)
        hashed = hashlib.pbkdf2_hmac(
            'sha256',
            password.encode('utf-8'),
            salt.encode('utf-8'),
            iterations=100000
        )
        return hashed.hex(), salt
 
    @staticmethod
    def verify_password(password: str, hashed: str, salt: str) -> bool:
        """パスワードの検証"""
        new_hash, _ = SecurityUtils.hash_password(password, salt)
        return hmac.compare_digest(new_hash, hashed)
 
    @staticmethod
    def sanitize_input(value: str) -> str:
        """入力値のサニタイズ"""
        dangerous_chars = ['<', '>', '"', "'", '&', '\\']
        result = value
        for char in dangerous_chars:
            result = result.replace(char, '')
        return result.strip()
 
# 使用例
token = SecurityUtils.generate_token()
hashed, salt = SecurityUtils.hash_password("my_password")
is_valid = SecurityUtils.verify_password("my_password", hashed, salt)

# マイグレーションスクリプトのテンプレート
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Callable
 
logger = logging.getLogger(__name__)
 
class MigrationRunner:
    """段階的マイグレーション実行エンジン"""
 
    def __init__(self, migration_dir: str):
        self.migration_dir = Path(migration_dir)
        self.migrations: List[Dict] = []
        self.completed: List[str] = []
 
    def register(self, version: str, description: str,
                 up: Callable, down: Callable):
        """マイグレーションの登録"""
        self.migrations.append({
            'version': version,
            'description': description,
            'up': up,
            'down': down,
            'registered_at': datetime.now().isoformat()
        })
 
    def run_up(self, target_version: str = None):
        """マイグレーションの実行（アップグレード）"""
        for migration in self.migrations:
            if migration['version'] in self.completed:
                continue
            logger.info(f"実行中: {migration['version']} - "
                       f"{migration['description']}")
            try:
                migration['up']()
                self.completed.append(migration['version'])
                logger.info(f"完了: {migration['version']}")
            except Exception as e:
                logger.error(f"失敗: {migration['version']}: {e}")
                raise
            if target_version and migration['version'] == target_version:
                break
 
    def run_down(self, target_version: str):
        """マイグレーションのロールバック"""
        for migration in reversed(self.migrations):
            if migration['version'] not in self.completed:
                continue
            if migration['version'] == target_version:
                break
            logger.info(f"ロールバック: {migration['version']}")
            migration['down']()
            self.completed.remove(migration['version'])
 
    def status(self) -> Dict:
        """マイグレーション状態の確認"""
        return {
            'total': len(self.migrations),
            'completed': len(self.completed),
            'pending': len(self.migrations) - len(self.completed),
            'versions': {
                m['version']: 'completed'
                if m['version'] in self.completed else 'pending'
                for m in self.migrations
            }
        }