103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""Alert notification domain models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class AlertHistoryLoadError(RuntimeError):
|
|
def __init__(self, history_path: Path, message: str) -> None:
|
|
super().__init__(message)
|
|
self.history_path = history_path
|
|
|
|
|
|
@dataclass
|
|
class AlertEvent:
|
|
severity: str
|
|
message: str
|
|
ltv_ratio: float
|
|
warning_threshold: float
|
|
critical_threshold: float
|
|
spot_price: float
|
|
updated_at: str
|
|
email_alerts_enabled: bool
|
|
|
|
def __post_init__(self) -> None:
|
|
for field_name in ("severity", "message", "updated_at"):
|
|
value = getattr(self, field_name)
|
|
if not isinstance(value, str):
|
|
raise TypeError(f"{field_name} must be a string")
|
|
for field_name in ("ltv_ratio", "warning_threshold", "critical_threshold", "spot_price"):
|
|
value = getattr(self, field_name)
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise TypeError(f"{field_name} must be numeric")
|
|
numeric_value = float(value)
|
|
if not math.isfinite(numeric_value):
|
|
raise ValueError(f"{field_name} must be finite")
|
|
setattr(self, field_name, numeric_value)
|
|
if not isinstance(self.email_alerts_enabled, bool):
|
|
raise TypeError("email_alerts_enabled must be a bool")
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "AlertEvent":
|
|
return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
|
|
|
|
|
|
@dataclass
|
|
class AlertStatus:
|
|
severity: str
|
|
message: str
|
|
ltv_ratio: float
|
|
warning_threshold: float
|
|
critical_threshold: float
|
|
email_alerts_enabled: bool
|
|
history: list[AlertEvent]
|
|
history_unavailable: bool = False
|
|
history_notice: str | None = None
|
|
|
|
|
|
class AlertHistoryRepository:
|
|
"""File-backed alert history store."""
|
|
|
|
HISTORY_PATH = Path("data/alert_history.json")
|
|
|
|
def __init__(self, history_path: Path | None = None) -> None:
|
|
self.history_path = history_path or self.HISTORY_PATH
|
|
self.history_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
def load(self) -> list[AlertEvent]:
|
|
if not self.history_path.exists():
|
|
return []
|
|
try:
|
|
with self.history_path.open() as f:
|
|
data = json.load(f)
|
|
except json.JSONDecodeError as exc:
|
|
raise AlertHistoryLoadError(self.history_path, f"Alert history is not valid JSON: {exc}") from exc
|
|
except OSError as exc:
|
|
raise AlertHistoryLoadError(self.history_path, f"Alert history could not be read: {exc}") from exc
|
|
if not isinstance(data, list):
|
|
raise AlertHistoryLoadError(self.history_path, "Alert history payload must be a list")
|
|
events: list[AlertEvent] = []
|
|
for index, item in enumerate(data):
|
|
if not isinstance(item, dict):
|
|
raise AlertHistoryLoadError(self.history_path, f"Alert history entry {index} must be an object")
|
|
try:
|
|
events.append(AlertEvent.from_dict(item))
|
|
except (TypeError, ValueError) as exc:
|
|
raise AlertHistoryLoadError(
|
|
self.history_path,
|
|
f"Alert history entry {index} is invalid: {exc}",
|
|
) from exc
|
|
return events
|
|
|
|
def save(self, events: list[AlertEvent]) -> None:
|
|
with self.history_path.open("w") as f:
|
|
json.dump([event.to_dict() for event in events], f, indent=2)
|