uptime-monitor

app/config.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import dataclasses
+from typing import List, Optional
+
+import yaml
+
+
+class ConfigError(Exception):
+    """Raised for any problem with the config file's contents."""
+
+
+@dataclasses.dataclass
+class Endpoint:
+    name: str
+    host: str
+    scheme: str  # "http" or "https"
+    port: int
+    dns_server: str  # DNS server used to resolve `host` — the system/router
+    # resolver is never consulted, see app/dns_resolver.py
+    expected_status: int
+    path: str = "/"
+    timeout_seconds: Optional[float] = None
+    verify_tls: bool = True  # set False for endpoints with self-signed certs
+
+
+@dataclasses.dataclass
+class InfluxConfig:
+    url: str
+    token: str
+    org: str
+    bucket: str
+    verify_ssl: bool = False  # default off: influx uses a self-signed cert
+
+
+@dataclasses.dataclass
+class CheckConfig:
+    interval_seconds: int = 600
+    timeout_seconds: float = 10.0
+    immediate_recheck_delay_seconds: float = 5.0
+
+
+@dataclasses.dataclass
+class AppConfig:
+    influx: InfluxConfig
+    check: CheckConfig
+    endpoints: List[Endpoint]
+
+
+def _require(d: dict, key: str, ctx: str):
+    value = d.get(key)
+    if value in (None, ""):
+        raise ConfigError(f"Missing required field '{key}' in {ctx}")
+    return value
+
+
+def load_config(path: str) -> AppConfig:
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            raw = yaml.safe_load(f)
+    except FileNotFoundError:
+        raise ConfigError(f"Config file not found: {path}")
+    except yaml.YAMLError as e:
+        raise ConfigError(f"Invalid YAML in config file '{path}': {e}")
+
+    if not raw:
+        raise ConfigError(f"Config file '{path}' is empty")
+
+    influx_raw = raw.get("influx") or {}
+    influx = InfluxConfig(
+        url=_require(influx_raw, "url", "influx"),
+        token=_require(influx_raw, "token", "influx"),
+        org=_require(influx_raw, "org", "influx"),
+        bucket=_require(influx_raw, "bucket", "influx"),
+        verify_ssl=bool(influx_raw.get("verify_ssl", False)),
+    )
+
+    check_raw = raw.get("check") or {}
+    check = CheckConfig(
+        interval_seconds=int(check_raw.get("interval_seconds", 600)),
+        timeout_seconds=float(check_raw.get("timeout_seconds", 10.0)),
+        immediate_recheck_delay_seconds=float(
+            check_raw.get("immediate_recheck_delay_seconds", 5.0)
+        ),
+    )
+
+    endpoints_raw = raw.get("endpoints") or []
+    if not endpoints_raw:
+        raise ConfigError("No endpoints configured under 'endpoints'")
+
+    endpoints: List[Endpoint] = []
+    seen_names = set()
+    for i, ep in enumerate(endpoints_raw):
+        ctx = f"endpoints[{i}]"
+        name = _require(ep, "name", ctx)
+        if name in seen_names:
+            raise ConfigError(f"Duplicate endpoint name: '{name}'")
+        seen_names.add(name)
+
+        scheme = str(_require(ep, "scheme", ctx)).lower()
+        if scheme not in ("http", "https"):
+            raise ConfigError(
+                f"{ctx} ('{name}'): scheme must be 'http' or 'https', got '{scheme}'"
+            )
+
+        timeout = ep.get("timeout_seconds")
+        endpoints.append(
+            Endpoint(
+                name=name,
+                host=_require(ep, "host", ctx),
+                scheme=scheme,
+                port=int(_require(ep, "port", ctx)),
+                dns_server=_require(ep, "dns_server", ctx),
+                expected_status=int(_require(ep, "expected_status", ctx)),
+                path=ep.get("path", "/"),
+                timeout_seconds=float(timeout) if timeout else None,
+                verify_tls=bool(ep.get("verify_tls", True)),
+            )
+        )
+
+    return AppConfig(influx=influx, check=check, endpoints=endpoints)