上流データの stale 検知と代替ソース自動切替：fallback チェーンで「データが古い」を運用で吸収する

1
def fetch_latest():
2
    resp = requests.get("https://api.example.com/latest")
3
    resp.raise_for_status()
4
    return resp.json()

1
from datetime import datetime, timedelta
2
from dataclasses import dataclass
3

4
@dataclass
5
class FreshnessResult:
6
    is_fresh: bool
7
    age_minutes: float
8
    threshold_minutes: float
9
    data_timestamp: datetime
10

11
def check_freshness(data: dict, source_name: str) -> FreshnessResult:
12
    """データの中身のタイムスタンプから鮮度を判定"""
13
    cfg = freshness_config.get(source_name)
14
    threshold = cfg.max_age_minutes
15

14 collapsed lines
16
    # データに含まれるタイムスタンプを取得
17
    ts_str = data.get("last_updated") or data.get("timestamp")
18
    if not ts_str:
19
        raise ValueError(f"{source_name}: timestamp field missing")
20

21
    data_ts = datetime.fromisoformat(ts_str)
22
    age = (datetime.now(data_ts.tzinfo) - data_ts).total_seconds() / 60
23

24
    return FreshnessResult(
25
        is_fresh=age <= threshold,
26
        age_minutes=age,
27
        threshold_minutes=threshold,
28
        data_timestamp=data_ts,
29
    )

1
def fetch_with_freshness_check(source_name: str) -> dict:
2
    data = _fetch(source_name)
3
    result = check_freshness(data, source_name)
4

5
    if not result.is_fresh:
6
        logger.warning(
7
            f"{source_name} is stale: "
8
            f"age={result.age_minutes:.0f}min > threshold={result.threshold_minutes}min"
9
        )
10
        raise StaleDataError(source_name, result)
11

12
    return data

1
SOURCE_CHAIN = [
2
    "primary_api",      # 最優先
3
    "secondary_api",    # primary が stale なら
4
    "cached_snapshot",  # secondary も stale なら（数時間前のキャッシュ）
5
    "manual_override",  # 全部 stale なら（運用者が手動で置く）
6
]
7

8
def fetch_with_fallback(metric_name: str) -> tuple[dict, str]:
9
    """優先順に試して、最初に fresh だったソースを返す"""
10
    last_error = None
11
    for source in SOURCE_CHAIN:
12
        try:
13
            data = fetch_with_freshness_check(f"{source}.{metric_name}")
14
            return data, source
15
        except (StaleDataError, requests.RequestException) as e:
5 collapsed lines
16
            logger.warning(f"{source} failed: {e}")
17
            last_error = e
18

19
    # 全部失敗
20
    raise AllSourcesStaleError(metric_name, last_error)

1
def fetch_with_fallback_and_notify(metric_name: str):
2
    data, used_source = fetch_with_fallback(metric_name)
3
    if used_source != SOURCE_CHAIN[0]:
4
        send_slack(
5
            f"⚠️ {metric_name} は {used_source} を使用 "
6
            f"(primary stale or down)"
7
        )
8
    return data, used_source

1
SNAPSHOT_DIR = Path("/var/data/snapshots")
2

3
def save_snapshot(metric_name: str, data: dict):
4
    """fresh なデータを取得できたら snapshot として保存"""
5
    path = SNAPSHOT_DIR / f"{metric_name}.json"
6
    path.write_text(json.dumps({
7
        "data": data,
8
        "saved_at": datetime.now().isoformat(),
9
    }))
10

11
def fetch_from_snapshot(metric_name: str) -> dict:
12
    path = SNAPSHOT_DIR / f"{metric_name}.json"
13
    if not path.exists():
14
        raise FileNotFoundError
15
    snapshot = json.loads(path.read_text())
5 collapsed lines
16
    saved_at = datetime.fromisoformat(snapshot["saved_at"])
17
    age_h = (datetime.now() - saved_at).total_seconds() / 3600
18
    if age_h > 24:
19
        raise StaleSnapshotError(f"snapshot is {age_h:.0f}h old")
20
    return snapshot["data"]

1
def normalize(data: dict, source: str) -> dict:
2
    """ソースごとに異なる形式を共通フォーマットに正規化"""
3
    if source.startswith("primary"):
4
        return {
5
            "timestamp": data["last_updated"],
6
            "value": data["value"],
7
        }
8
    elif source.startswith("secondary"):
9
        return {
10
            "timestamp": data["timestamp"],
11
            "value": data["metric_value"] / 1000,  # 単位変換
12
        }
13
    elif source.startswith("cached"):
14
        return data  # snapshot は既に正規化済み
15
    else:
1 collapsed line
16
        raise ValueError(f"unknown source: {source}")

1
def sanity_check(new_data: dict, last_known_data: dict | None) -> bool:
2
    if last_known_data is None:
3
        return True
4
    new_v = new_data["value"]
5
    last_v = last_known_data["value"]
6
    if last_v == 0:
7
        return new_v < 1.0  # 急に大きな値は怪しい
8
    ratio = new_v / last_v
9
    return 0.5 <= ratio <= 2.0  # 半分〜2 倍の範囲

1
data = fetch_with_freshness_check("primary_api")
2
save_snapshot(metric_name, data)  # 取得直後に保存