from __future__ import annotations import json import os import tempfile import time from pathlib import Path class HealthSnapshotStore: def __init__(self, path: Path, stale_after_s: int) -> None: self.path = path self.stale_after_s = stale_after_s def write(self, snapshots: dict[str, AccountSnapshot]) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) payload = self._serialize_snapshots(snapshots) with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=self.path.parent, delete=False, ) as temp_file: json.dump(payload, temp_file, sort_keys=True) temp_file.flush() os.fsync(temp_file.fileno()) temp_path = Path(temp_file.name) temp_path.replace(self.path) def read(self) -> dict[str, AccountSnapshot]: snapshots, valid = self._read_snapshots() if not valid: return {} return snapshots def read_fresh(self) -> dict[str, AccountSnapshot] | None: age = self.age_seconds() if age is None or age > self.stale_after_s: return None snapshots, valid = self._read_snapshots() if not valid: return None return snapshots def age_seconds(self) -> float | None: try: modified_at = self.path.stat().st_mtime except OSError: return None return time.time() - modified_at @staticmethod def _serialize_snapshots(snapshots: dict[str, AccountSnapshot]) -> dict[str, dict[str, object]]: return { slug: { "status": snapshot.status.value, "primary_used_pct": snapshot.primary_used_pct, "secondary_used_pct": snapshot.secondary_used_pct, "primary_reset_at": snapshot.primary_reset_at, "secondary_reset_at": snapshot.secondary_reset_at, "checked_at": snapshot.checked_at, "detail": snapshot.detail, "named_limits": [ { "kind": limit.kind, "group": limit.group, "percent": limit.percent, "resets_at": limit.resets_at, "active": limit.active, } for limit in snapshot.named_limits ], "extra_usage": None if snapshot.extra_usage is None else { "used": snapshot.extra_usage.used, "limit": snapshot.extra_usage.limit, "unit": snapshot.extra_usage.unit, "display": snapshot.extra_usage.display, }, "spend": None if snapshot.spend is None else { "amount": snapshot.spend.amount, "limit": snapshot.spend.limit, "currency": snapshot.spend.currency, "display": snapshot.spend.display, }, } for slug, snapshot in snapshots.items() } def _read_snapshots(self) -> tuple[dict[str, AccountSnapshot], bool]: from health_client import ( AccountSnapshot, ExtraUsageSummary, HealthStatus, NamedLimit, SpendSummary, ) try: payload = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {}, False if not isinstance(payload, dict): return {}, False snapshots: dict[str, AccountSnapshot] = {} for slug, value in payload.items(): if not isinstance(slug, str) or not isinstance(value, dict): continue try: status = HealthStatus(value.get("status")) except ValueError: continue primary = value.get("primary_used_pct") secondary = value.get("secondary_used_pct") primary_reset_at = value.get("primary_reset_at") secondary_reset_at = value.get("secondary_reset_at") checked_at = value.get("checked_at") detail = value.get("detail") named_limits: list[NamedLimit] = [] raw_limits = value.get("named_limits") if isinstance(raw_limits, list): for raw_limit in raw_limits: if not isinstance(raw_limit, dict): continue kind = raw_limit.get("kind") active = raw_limit.get("active") if not isinstance(kind, str) or not kind or not isinstance(active, bool): continue group = raw_limit.get("group") named_limits.append( NamedLimit( kind=kind, group=group if isinstance(group, str) else None, percent=self._coerce_percent(raw_limit.get("percent")), resets_at=self._coerce_float(raw_limit.get("resets_at")), active=active, ) ) extra_usage = self._parse_extra_usage(value.get("extra_usage"), ExtraUsageSummary) spend = self._parse_spend(value.get("spend"), SpendSummary) snapshots[slug] = AccountSnapshot( status=status, primary_used_pct=primary if isinstance(primary, int) else None, secondary_used_pct=secondary if isinstance(secondary, int) else None, primary_reset_at=float(primary_reset_at) if isinstance(primary_reset_at, (int, float)) and not isinstance(primary_reset_at, bool) else None, secondary_reset_at=float(secondary_reset_at) if isinstance(secondary_reset_at, (int, float)) and not isinstance(secondary_reset_at, bool) else None, checked_at=self._coerce_float(checked_at), detail=detail if isinstance(detail, str) and detail else None, named_limits=tuple(named_limits), extra_usage=extra_usage, spend=spend, ) return snapshots, True @staticmethod def _coerce_float(value: object) -> float | None: if isinstance(value, bool): return None if isinstance(value, (int, float)): return float(value) return None @staticmethod def _coerce_percent(value: object) -> int | None: if not isinstance(value, int) or isinstance(value, bool): return None return max(0, min(100, value)) @staticmethod def _parse_extra_usage( value: object, summary_type: type[object], ) -> object | None: if not isinstance(value, dict): return None used = HealthSnapshotStore._coerce_float(value.get("used")) limit = HealthSnapshotStore._coerce_float(value.get("limit")) unit = value.get("unit") display = value.get("display") if ( used is None and limit is None and not isinstance(unit, str) and not isinstance(display, str) ): return None return summary_type( used=used, limit=limit, unit=unit if isinstance(unit, str) else None, display=display if isinstance(display, str) else None, ) @staticmethod def _parse_spend( value: object, summary_type: type[object], ) -> object | None: if not isinstance(value, dict): return None amount = HealthSnapshotStore._coerce_float(value.get("amount")) limit = HealthSnapshotStore._coerce_float(value.get("limit")) currency = value.get("currency") display = value.get("display") if ( amount is None and limit is None and not isinstance(currency, str) and not isinstance(display, str) ): return None return summary_type( amount=amount, limit=limit, currency=currency if isinstance(currency, str) else None, display=display if isinstance(display, str) else None, )