べき等性 (Idempotency)：「同じ処理を 2 回実行しても結果が変わらない」を保証する

-- 非べき等: 重複実行で行が増えるINSERT INTO users (id, name, email)VALUES (1, '佐藤', 'sato@example.com');
-- べき等: 重複時は何もしない (or 更新)INSERT INTO users (id, name, email)VALUES (1, '佐藤', 'sato@example.com')ON CONFLICT (id) DO NOTHING;
-- べき等: 存在すれば更新 (UPSERT)INSERT INTO users (id, name, email)VALUES (1, '佐藤', 'sato@example.com')ON CONFLICT (id) DO UPDATESET name = EXCLUDED.name,    email = EXCLUDED.email;

-- 標準 SQL の MERGE 文MERGE INTO target_table TUSING source_table S  ON T.id = S.idWHEN MATCHED THEN  UPDATE SET T.value = S.value, T.updated_at = CURRENT_TIMESTAMP()WHEN NOT MATCHED THEN  INSERT (id, value, created_at)  VALUES (S.id, S.value, CURRENT_TIMESTAMP());
-- これを 100 回実行しても結果は同じ

{{    config(        materialized='incremental',        unique_key='order_id',        on_schema_change='sync_all_columns'    )}}
SELECT    order_id,    user_id,    amount,    created_atFROM {{ source('raw', 'orders') }}{% if is_incremental() %}  WHERE created_at >= (SELECT MAX(created_at) FROM {{ this }})        - INTERVAL 1 DAY  -- オーバーラップで取りこぼし防止{% endif %}-- unique_key='order_id' を指定すると、dbt が MERGE 文を生成-- 重複実行しても行は重複しない

def process_order(order_id, db):    # 既に処理済みかチェック    cursor = db.execute(        "SELECT 1 FROM processed_orders WHERE order_id = %s",        (order_id,)    )    if cursor.fetchone():        print(f"order {order_id} は処理済み、skip")        return
    # トランザクション内で「処理 + 処理済みマーク」を一括    with db.transaction():        # 実際の処理 (例: 在庫減らす、メール送る)        db.execute(            "UPDATE inventory SET stock = stock - 1 WHERE id = %s",            (order_id,)        )        # 処理済みマーク        db.execute(            "INSERT INTO processed_orders (order_id, processed_at) VALUES (%s, NOW())",            (order_id,)        )

from flask import Flask, request, jsonifyimport hashlib, json
app = Flask(__name__)idempotency_cache = {}  # 本番は Redis 等
@app.post('/charge')def charge():    key = request.headers.get('Idempotency-Key')    if not key:        return jsonify({'error': 'Idempotency-Key required'}), 400
    # 同じキーで過去に処理した結果をキャッシュから返す    if key in idempotency_cache:        return jsonify(idempotency_cache[key]), 200
    # 実際の課金処理    body = request.json    result = process_charge(body['amount'], body['customer_id'])
    # 結果をキャッシュ (24h 等)    idempotency_cache[key] = result    return jsonify(result), 201
# クライアント側import requests, uuidkey = str(uuid.uuid4())  # クライアントが生成r = requests.post('/charge',    headers={'Idempotency-Key': key},    json={'amount': 1000, 'customer_id': 'c1'})# リトライ時も同じ key を送る → 2 回目以降はキャッシュ返答