-- テスト用テーブルの準備
CREATE TABLE employees (
    id            SERIAL PRIMARY KEY,
    name          VARCHAR(100) NOT NULL,
    department_id INTEGER NOT NULL,
    salary        INTEGER NOT NULL,
    hired_date    DATE NOT NULL DEFAULT CURRENT_DATE
);
 
CREATE TABLE departments (
    id   SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL
);
 
-- EXPLAIN: 実行計画を推定（実行しない — 安全）
EXPLAIN
SELECT e.name, d.name AS department
FROM employees e
    INNER JOIN departments d ON e.department_id = d.id
WHERE e.salary > 500000;
 
-- 出力例:
-- Hash Join  (cost=1.09..2.24 rows=3 width=64)
--   Hash Cond: (e.department_id = d.id)
--   ->  Seq Scan on employees e  (cost=0.00..1.12 rows=3 width=40)
--         Filter: (salary > 500000)
--   ->  Hash  (cost=1.05..1.05 rows=5 width=36)
--         ->  Seq Scan on departments d  (cost=0.00..1.05 rows=5 width=36)
 
 
-- EXPLAIN ANALYZE: 実際に実行して測定（DMLに注意）
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT e.name, d.name AS department
FROM employees e
    INNER JOIN departments d ON e.department_id = d.id
WHERE e.salary > 500000;
 
-- 出力例:
-- Hash Join  (cost=1.09..2.24 rows=3 width=64)
--            (actual time=0.035..0.042 rows=5 loops=1)
--   Hash Cond: (e.department_id = d.id)
--   Buffers: shared hit=2
--   ->  Seq Scan on employees e  (cost=0.00..1.12 rows=3 width=40)
--                                (actual time=0.012..0.018 rows=5 loops=1)
--         Filter: (salary > 500000)
--         Rows Removed by Filter: 95
--         Buffers: shared hit=1
--   ->  Hash  (cost=1.05..1.05 rows=5 width=36)
--             (actual time=0.008..0.008 rows=5 loops=1)
--         Buckets: 1024  Batches: 1
--         ->  Seq Scan on departments d  (cost=0.00..1.05 rows=5 width=36)
--                                        (actual time=0.003..0.004 rows=5 loops=1)
--         Buffers: shared hit=1
-- Planning Time: 0.15 ms
-- Execution Time: 0.08 ms

-- ===== Sequential Scan (Seq Scan) =====
-- テーブルの全ページを先頭から順に読む
-- 適する: テーブル全体の大部分を取得する場合
EXPLAIN ANALYZE SELECT * FROM employees WHERE status = 'active';
-- Seq Scan on employees  (cost=0.00..1.12 rows=80 width=200)
--   Filter: (status = 'active')
--   Rows Removed by Filter: 20  ← フィルタで除外された行数
 
 
-- ===== Index Scan =====
-- インデックスで行の位置(TID)を特定し、テーブルから行を取得
-- 適する: 少数の行を取得する場合（選択率 < ~5%）
EXPLAIN ANALYZE SELECT * FROM employees WHERE id = 42;
-- Index Scan using employees_pkey on employees
--   (cost=0.15..8.17 rows=1 width=200)
--   Index Cond: (id = 42)
--   Buffers: shared hit=3  ← インデックス2ページ + テーブル1ページ
 
 
-- ===== Index Only Scan =====
-- インデックスだけで回答（テーブルアクセス不要）
-- 適する: カバリングインデックスが存在し、必要カラムが全てインデックス内
EXPLAIN ANALYZE SELECT id, name FROM employees WHERE id BETWEEN 1 AND 100;
-- Index Only Scan using idx_emp_covering on employees
--   Index Cond: (id >= 1 AND id <= 100)
--   Heap Fetches: 0  ← テーブルアクセスなし（VACUUMが最新の場合）
--   Heap Fetches: 15 ← VACUUMが遅れていると可視性チェックが必要
 
 
-- ===== Bitmap Index Scan + Bitmap Heap Scan =====
-- 複数のインデックスをビットマップで合成してから一括取得
-- 適する: 中程度の行数、複数条件のAND/OR
EXPLAIN ANALYZE
SELECT * FROM employees
WHERE department_id = 10 AND salary > 500000;
-- Bitmap Heap Scan on employees
--   Recheck Cond: (department_id = 10)
--   Filter: (salary > 500000)
--   Rows Removed by Filter: 5
--   ->  Bitmap Index Scan on idx_emp_dept
--         Index Cond: (department_id = 10)
-- → まずビットマップでページ位置を収集、
--   次にページ順にソートしてテーブルを読む（ランダムI/Oを削減）
 
 
-- ===== Parallel Seq Scan =====
-- 複数のワーカーで並列にテーブルを読む
-- 適する: 大テーブルの全走査
EXPLAIN ANALYZE SELECT COUNT(*) FROM employees WHERE salary > 300000;
-- Finalize Aggregate
--   ->  Gather  (Workers Planned: 2)
--         ->  Partial Aggregate
--               ->  Parallel Seq Scan on employees
--                     Filter: (salary > 300000)
-- → 2ワーカー + リーダーで並列スキャン

-- ===== Nested Loop Join =====
-- 外側テーブルの各行に対して内側テーブルをスキャン
-- 適する: 外側が小、内側にインデックスあり
-- 計算量: O(N × M)（インデックスあれば O(N × log M)）
EXPLAIN ANALYZE
SELECT * FROM orders o
    INNER JOIN customers c ON o.customer_id = c.id
WHERE o.id = 42;
-- Nested Loop  (cost=0.56..16.59 rows=1 width=400)
--   ->  Index Scan using orders_pkey on orders o  (rows=1)
--   ->  Index Scan using customers_pkey on customers c  (rows=1)
-- → 外側1行 × 内側1行 = 1回のIndex Scan
 
 
-- ===== Hash Join =====
-- 小テーブルでハッシュ表を構築、大テーブルで照合
-- 適する: 等値結合、中〜大テーブル
-- 計算量: O(N + M)、メモリ使用: O(min(N, M))
EXPLAIN ANALYZE
SELECT * FROM orders o
    INNER JOIN customers c ON o.customer_id = c.id;
-- Hash Join  (cost=1.09..35.24 rows=1000 width=400)
--   Hash Cond: (o.customer_id = c.id)
--   ->  Seq Scan on orders o        ← 大テーブル（probe側）
--   ->  Hash                         ← ハッシュ表の構築
--         Buckets: 1024  Batches: 1  Memory Usage: 40kB
--         ->  Seq Scan on customers c  ← 小テーブル（build側）
 
 
-- ===== Merge Join =====
-- 両テーブルをソートして並行走査
-- 適する: 大テーブル同士の結合、ソート済みデータ
-- 計算量: O(N log N + M log M)（既ソートなら O(N + M)）
EXPLAIN ANALYZE
SELECT * FROM large_orders o
    INNER JOIN large_customers c ON o.customer_id = c.id
ORDER BY o.customer_id;
-- Merge Join  (cost=...)
--   Merge Cond: (o.customer_id = c.id)
--   ->  Sort  ->  Seq Scan on large_orders
--   ->  Sort  ->  Seq Scan on large_customers

-- テーブルの統計情報を確認
SELECT
    attname AS column_name,
    n_distinct,           -- ユニーク値数（負数は行数に対する割合）
    null_frac,            -- NULL率（0.0〜1.0）
    avg_width,            -- 平均バイト幅
    most_common_vals,     -- 最頻値（上位N個）
    most_common_freqs,    -- 最頻値の出現率
    histogram_bounds      -- ヒストグラムの境界値
FROM pg_stats
WHERE tablename = 'employees' AND attname = 'department_id';
 
-- n_distinct の解釈:
--   正の値: ユニーク値の推定数（例: 10 → 10種類の値）
--   負の値: 行数に対する割合（例: -0.5 → 行数の50%がユニーク）
--   -1: 全行がユニーク
 
-- 統計情報の手動更新
ANALYZE employees;
 
-- 特定カラムの統計精度を上げる（デフォルト100、最大10000）
ALTER TABLE employees ALTER COLUMN department_id SET STATISTICS 1000;
ANALYZE employees;
-- → histogram_bounds のバケット数が増え、カーディナリティ推定が改善
 
-- テーブルの行数推定 vs 実際
SELECT
    relname,
    reltuples::BIGINT AS estimated_rows,    -- 推定行数（ANALYZEで更新）
    pg_stat_get_live_tuples(oid) AS actual_rows  -- 実際のlive行数
FROM pg_class
WHERE relname = 'employees';

-- 拡張統計: 複数カラム間の相関を考慮した統計情報
-- PostgreSQL 10+ で利用可能
 
-- 問題: city='東京' AND prefecture='東京都' の選択率を
--        各カラム独立に計算すると実際より小さく見積もる
-- → 「東京に住む人は東京都」という相関を知らないため
 
-- 依存統計（functional dependencies）
CREATE STATISTICS stat_emp_city_pref (dependencies)
ON city, prefecture FROM addresses;
ANALYZE addresses;
 
-- n-distinct統計（複合カーディナリティ）
CREATE STATISTICS stat_orders_status_date (ndistinct)
ON status, DATE_TRUNC('month', order_date) FROM orders;
ANALYZE orders;
 
-- MCV統計（最頻値の組み合わせ）— PostgreSQL 12+
CREATE STATISTICS stat_orders_mcv (mcv)
ON status, payment_method FROM orders;
ANALYZE orders;
 
-- 拡張統計の効果確認
EXPLAIN SELECT * FROM addresses
WHERE city = '東京' AND prefecture = '東京都';
-- Before: rows=10 (各カラムの選択率を独立に乗算: 0.1 × 0.1 = 0.01)
-- After:  rows=1000 (相関を考慮: 東京→東京都の依存関係)

-- ===== パターン1: OR → UNION ALL =====
-- NG: OR条件でインデックスが効きにくい
SELECT * FROM orders
WHERE customer_id = 42 OR product_id = 100;
-- → Bitmap OR（効率が悪い場合がある）
 
-- OK: UNION ALLに書き換え
SELECT * FROM orders WHERE customer_id = 42
UNION ALL
SELECT * FROM orders WHERE product_id = 100
  AND customer_id != 42;  -- 重複排除
 
 
-- ===== パターン2: IN (サブクエリ) → EXISTS =====
-- NG: IN (サブクエリ) が非効率な場合
SELECT * FROM products
WHERE id IN (SELECT product_id FROM order_items WHERE quantity > 10);
 
-- OK: EXISTS に書き換え（相関サブクエリとして最適化される）
SELECT * FROM products p
WHERE EXISTS (
    SELECT 1 FROM order_items oi
    WHERE oi.product_id = p.id AND oi.quantity > 10
);
 
-- OK: JOIN に書き換え（重複に注意）
SELECT DISTINCT p.*
FROM products p
    INNER JOIN order_items oi ON p.id = oi.product_id
WHERE oi.quantity > 10;
 
 
-- ===== パターン3: NOT IN → NOT EXISTS =====
-- NG: NOT IN はNULLの扱いが危険 + Anti Join最適化が効きにくい
SELECT * FROM products
WHERE id NOT IN (SELECT product_id FROM discontinued_products);
-- → product_id にNULLがあると結果が空になる（SQL標準の3値論理）
 
-- OK: NOT EXISTS（NULLセーフ + Anti Joinに最適化される）
SELECT * FROM products p
WHERE NOT EXISTS (
    SELECT 1 FROM discontinued_products dp
    WHERE dp.product_id = p.id
);
-- → Anti Join: Hash Anti Join or Merge Anti Join で効率的に処理
 
 
-- ===== パターン4: CTE（WITH句）の注意点 =====
-- PostgreSQL 12未満: CTEは最適化バリア（インライン展開されない）
-- PostgreSQL 12+: 1回だけ参照されるCTEは自動的にインライン展開
 
-- NG (PG11以前): CTEがSeq Scanを強制
WITH active_orders AS (
    SELECT * FROM orders WHERE status = 'active'
)
SELECT * FROM active_orders WHERE customer_id = 42;
-- → CTEが全active注文をマテリアライズしてからフィルタ
 
-- OK: サブクエリまたはCTE + MATERIALIZED/NOT MATERIALIZED
-- PG12+ではデフォルトでインライン展開される
WITH active_orders AS NOT MATERIALIZED (
    SELECT * FROM orders WHERE status = 'active'
)
SELECT * FROM active_orders WHERE customer_id = 42;
-- → customer_id=42 の条件がordersテーブルまでプッシュダウン
 
 
-- ===== パターン5: OFFSET の代わりにカーソルベースページング =====
-- NG: OFFSET が大きいと全行をスキャンしてからスキップ
SELECT * FROM products ORDER BY id LIMIT 20 OFFSET 100000;
-- → 100020行を読んで最初の100000行を捨てる
 
-- OK: WHERE id > last_seen_id でカーソルベースページング
SELECT * FROM products
WHERE id > 100000  -- 前回最後のIDを使う
ORDER BY id
LIMIT 20;
-- → Index Scanで20行だけ読む（大幅に高速）

-- ===== pg_stat_statements =====
-- 最も遅いクエリの特定（拡張モジュール）
-- postgresql.conf: shared_preload_libraries = 'pg_stat_statements'
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
 
SELECT
    calls,
    ROUND(total_exec_time::NUMERIC, 2) AS total_ms,
    ROUND(mean_exec_time::NUMERIC, 2) AS avg_ms,
    ROUND(stddev_exec_time::NUMERIC, 2) AS stddev_ms,
    rows,
    ROUND(100.0 * shared_blks_hit / NULLIF(shared_blks_hit + shared_blks_read, 0), 2)
        AS cache_hit_pct,
    LEFT(query, 120) AS query_preview
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 20;
 
-- キャッシュヒット率が低いクエリ（I/Oバウンド）
SELECT
    query,
    calls,
    shared_blks_hit,
    shared_blks_read,
    ROUND(100.0 * shared_blks_hit / NULLIF(shared_blks_hit + shared_blks_read, 0), 2)
        AS cache_hit_pct
FROM pg_stat_statements
WHERE calls > 100
ORDER BY cache_hit_pct ASC
LIMIT 20;
 
 
-- ===== auto_explain =====
-- 自動的にスロークエリの実行計画をログに出力
-- postgresql.conf:
-- shared_preload_libraries = 'auto_explain'
-- auto_explain.log_min_duration = '100ms'
-- auto_explain.log_analyze = on
-- auto_explain.log_buffers = on
-- auto_explain.log_format = 'json'
 
 
-- ===== 実行計画のJSON形式出力（プログラムからの解析用）=====
EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)
SELECT * FROM employees WHERE department_id = 10;
 
-- ===== 実行計画の比較（SET で挙動を変えて確認）=====
-- インデックスを無効化して比較
SET enable_indexscan = off;
SET enable_bitmapscan = off;
EXPLAIN ANALYZE SELECT * FROM employees WHERE id = 42;
-- → Seq Scan が使われる
 
-- 元に戻す
RESET enable_indexscan;
RESET enable_bitmapscan;

-- ===== パラレルクエリの制御 =====
-- パラレルクエリのパラメータ
SHOW max_parallel_workers_per_gather;  -- デフォルト: 2
SHOW parallel_tuple_cost;              -- デフォルト: 0.1
SHOW min_parallel_table_scan_size;     -- デフォルト: 8MB
 
-- パラレル度を上げて大テーブルの集約を高速化
SET max_parallel_workers_per_gather = 4;
EXPLAIN ANALYZE
SELECT department_id, AVG(salary), COUNT(*)
FROM employees
GROUP BY department_id;
-- Finalize GroupAggregate
--   ->  Gather Merge  (Workers Planned: 4)
--         ->  Sort
--               ->  Partial HashAggregate
--                     ->  Parallel Seq Scan on employees
 
-- ===== work_mem チューニング =====
-- work_mem: ソートやハッシュ操作で使用するメモリ量
SHOW work_mem;  -- デフォルト: 4MB
 
-- Hash Join で Batches > 1 の場合 → work_mem不足
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM orders o JOIN customers c ON o.customer_id = c.id;
-- Hash Join
--   ->  Hash
--         Buckets: 65536  Batches: 4  ← Batches > 1 = ディスクに溢れている
--         Memory Usage: 4096kB
 
-- work_mem を増やして改善
SET work_mem = '64MB';  -- セッション単位で設定（グローバル設定は慎重に）
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM orders o JOIN customers c ON o.customer_id = c.id;
-- Hash Join
--   ->  Hash
--         Buckets: 262144  Batches: 1  ← 全てメモリ内で処理
--         Memory Usage: 32768kB
 
-- ソートでのwork_mem影響
EXPLAIN ANALYZE
SELECT * FROM orders ORDER BY created_at DESC;
-- Sort Method: external merge  Disk: 102400kB  ← ディスクソート（遅い）
-- work_mem増加後:
-- Sort Method: quicksort  Memory: 51200kB       ← メモリソート（速い）

-- ===== JIT（Just-In-Time）コンパイル =====
-- PostgreSQL 11+: 複雑なクエリの式評価をネイティブコードにコンパイル
SHOW jit;                    -- on (デフォルト)
SHOW jit_above_cost;         -- 100000 (このコスト以上でJIT有効)
SHOW jit_inline_above_cost;  -- 500000
SHOW jit_optimize_above_cost; -- 500000
 
-- JITが有効な場合の出力
EXPLAIN ANALYZE
SELECT SUM(salary * 1.1 + bonus) FROM employees;
-- JIT:
--   Functions: 4
--   Options: Inlining true, Optimization true, Expressions true
--   Timing: Generation 1.5 ms, Inlining 10.2 ms, Optimization 50.3 ms,
--           Emission 30.1 ms, Total 92.1 ms
-- → JITのオーバーヘッドが大きい場合はjit_above_costを上げる
 
 
-- ===== コスト設定の調整 =====
-- SSD環境でのrandom_page_costの最適化
SHOW random_page_cost;  -- デフォルト: 4.0（HDD想定）
SET random_page_cost = 1.1;  -- SSD環境では1.1〜1.5が推奨
-- → Index Scanがより積極的に選択される
 
-- effective_cache_size の設定
SHOW effective_cache_size;  -- デフォルト: 4GB
-- サーバの物理メモリの50-75%を設定
-- → オプティマイザがIndex Scanを選びやすくなる
SET effective_cache_size = '32GB';

-- NG: 「遅いからインデックスを追加」という安易な対応
CREATE INDEX idx_guess ON orders (status);
-- 効果がない理由: statusの選択率が低い（80%が'completed'）
-- → Seq Scanの方が速いためオプティマイザがインデックスを無視
 
-- OK: EXPLAIN ANALYZEで原因を特定してから対策
EXPLAIN ANALYZE
SELECT * FROM orders WHERE status = 'pending';
-- → Seq Scan、Filter: Rows Removed: 95000
-- → status='pending'は全体の5% → インデックスは有効
 
-- さらに部分インデックスで最適化
CREATE INDEX idx_orders_pending ON orders (created_at)
WHERE status = 'pending';

-- 推定 rows=10 vs 実際 rows=50000 のような乖離
-- → オプティマイザが間違った実行計画を選択する原因
-- → 例: Hash Joinの方が適切なのにNested Loopを選択
 
-- 対策1: ANALYZEを実行して統計を更新
ANALYZE orders;
 
-- 対策2: 統計精度を上げる
ALTER TABLE orders ALTER COLUMN status SET STATISTICS 1000;
ANALYZE orders;
 
-- 対策3: 拡張統計（複数列の相関）
CREATE STATISTICS stat_orders_status_date (dependencies)
ON status, order_date FROM orders;
ANALYZE orders;
 
-- 対策4: 乖離の確認方法
EXPLAIN ANALYZE SELECT * FROM orders WHERE status = 'pending';
-- rows=10 (estimated) vs rows=50000 (actual) → 5000倍の乖離
-- → ANALYZEと拡張統計で改善

-- NG: 不要なカラムまで取得
SELECT * FROM orders WHERE customer_id = 42;
-- → 全カラム取得 → Index Only Scanが使えない
-- → ネットワーク転送量が増大
 
-- OK: 必要なカラムのみ取得
SELECT id, status, total FROM orders WHERE customer_id = 42;
-- → カバリングインデックスがあればIndex Only Scan可能
-- → 転送データ量が削減

EXPLAIN (ANALYZE, BUFFERS)
SELECT o.id, o.total, c.name, c.email
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.created_at > '2024-01-01'
  AND o.status = 'delivered'
ORDER BY o.total DESC
LIMIT 100;
 
-- 出力:
-- Limit  (cost=45000..45002 rows=100 width=120)
--        (actual time=2500.1..2500.3 rows=100 loops=1)
--   ->  Sort  (cost=45000..45500 rows=50000 width=120)
--              (actual time=2500.0..2500.2 rows=100 loops=1)
--         Sort Key: o.total DESC
--         Sort Method: top-N heapsort  Memory: 50kB
--         ->  Hash Join  (cost=800..44000 rows=50000 width=120)
--                        (actual time=15.0..2400.0 rows=48000 loops=1)
--               Hash Cond: (o.customer_id = c.id)
--               ->  Seq Scan on orders o  (cost=0..40000 rows=50000 width=80)
--                                         (actual time=0.1..2300.0 rows=48000 loops=1)
--                     Filter: (created_at > '2024-01-01' AND status = 'delivered')
--                     Rows Removed by Filter: 952000
--                     Buffers: shared hit=5000 read=30000
--               ->  Hash  (cost=500..500 rows=10000 width=40)
--                          (actual time=10.0..10.0 rows=10000 loops=1)
--                     ->  Seq Scan on customers c
--                     Buffers: shared hit=200
-- Planning Time: 0.5 ms
-- Execution Time: 2500.5 ms

-- 遅いクエリ: 各顧客の最新注文を取得
SELECT c.id, c.name,
       (SELECT MAX(o.created_at) FROM orders o WHERE o.customer_id = c.id) AS last_order,
       (SELECT SUM(o.total) FROM orders o WHERE o.customer_id = c.id) AS total_spent,
       (SELECT COUNT(*) FROM orders o WHERE o.customer_id = c.id) AS order_count
FROM customers c
WHERE c.status = 'active';

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

BEGIN;
EXPLAIN ANALYZE UPDATE orders SET status = 'cancelled' WHERE id = 42;
ROLLBACK;  -- 実際の更新は取り消される