From 1c73542d297bf997ee2fee5d1b9210ce4baa8694 Mon Sep 17 00:00:00 2001 From: sinavir Date: Sun, 21 Apr 2024 11:55:34 +0200 Subject: [PATCH] feat: add status page declarative setup --- nixos/module.nix | 4 +++ stateless_uptime_kuma/tree_gen.py | 19 ++++++++++- stateless_uptime_kuma/uptime_kuma.py | 49 +++++++++++++++++++++++++--- 3 files changed, 66 insertions(+), 6 deletions(-) diff --git a/nixos/module.nix b/nixos/module.nix index ee01bea..740e7c6 100644 --- a/nixos/module.nix +++ b/nixos/module.nix @@ -40,6 +40,10 @@ in type = with lib.types; attrsOf probesFormat.type; default = { }; }; + status_pages = lib.mkOption { + type = with lib.types; attrsOf probesFormat.type; + default = { }; + }; }; }; config.statelessUptimeKuma = { diff --git a/stateless_uptime_kuma/tree_gen.py b/stateless_uptime_kuma/tree_gen.py index 97fc2fb..b32ec98 100644 --- a/stateless_uptime_kuma/tree_gen.py +++ b/stateless_uptime_kuma/tree_gen.py @@ -5,7 +5,7 @@ Classes to generate the item tree from json spec import logging import sys -from .uptime_kuma import Monitor, Notification, Tag +from .uptime_kuma import Monitor, Notification, StatusPage, Tag # type: ignore logger = logging.getLogger(__name__) @@ -44,8 +44,25 @@ def from_dict(api, tree, autocreate_tags=True): ] monitor_kwargs["notifications"] = associated_notifications indexed_monitors[monitor_name] = Monitor(api, monitor_name, **monitor_kwargs) + status_pages = tree.get("status_pages", []) + indexed_status_pages = {} + for slug, kwargs in status_pages.items(): + for group in kwargs.get("publicGroupList", []): + if "monitorList" in group: + monitorList = [] + for monitor in group["monitorList"]: + if monitor not in indexed_monitors: + logger.error( + "Fatal: status page is referencing a monitor that doesn't exist" + ) + sys.exit(1) + monitorList.append(indexed_monitors[monitor]) + group["monitorList"] = monitorList + indexed_status_pages[slug] = StatusPage(api, slug, **kwargs) + return { "monitors": indexed_monitors.values(), "tags": indexed_tags.values(), "notifications": indexed_notifications.values(), + "status_pages": indexed_status_pages.values(), } diff --git a/stateless_uptime_kuma/uptime_kuma.py b/stateless_uptime_kuma/uptime_kuma.py index c9f03b0..d156fd0 100644 --- a/stateless_uptime_kuma/uptime_kuma.py +++ b/stateless_uptime_kuma/uptime_kuma.py @@ -18,6 +18,7 @@ class Manager: self.sync_tags() self.sync_notifications() self.sync_monitors() + self.sync_status_pages() self.save() def save(self): @@ -28,20 +29,25 @@ class Manager: def sync_monitors(self): old = self.api.get_monitors() new = self.target_tree.get("monitors", []) - self.sync(new, old) + self.sync(new, old, "name") def sync_notifications(self): old = self.api.get_notifications() new = self.target_tree.get("notifications", []) - self.sync(new, old) + self.sync(new, old, "name") def sync_tags(self): old = self.api.get_tags() new = self.target_tree.get("tags", []) - self.sync(new, old) + self.sync(new, old, "name") - def sync(self, new, old): - indexed_old = {elem["name"]: elem for elem in old} + def sync_status_pages(self): + old = self.api.get_status_pages() + new = self.target_tree.get("status_pages", []) + self.sync(new, old, "slug") + + def sync(self, new, old, pk): + indexed_old = {elem[pk]: elem for elem in old} for k in new: if k.name in indexed_old: k.id = indexed_old[k.name]["id"] @@ -194,3 +200,36 @@ class Notification(Item): def __repr__(self): return f"Notification({str(self)})" + + +class StatusPage(Item): + def __init__(self, api, name, id=None, old_name=None, **kwargs): + super().__init__(api, name, id, old_name) + self.kwargs = kwargs + + def save(self): + if self.saved: + return + logger.debug(f"Saving {repr(self)}") + if self.id is None: + self.api.add_status_page( + slug=self.name, + title=self.kwargs["title"], + ) + rslt = self.api.get_status_page(self.name) + self.id = rslt["id"] + kwargs = self.kwargs + for group in kwargs.get("publicGroupList", []): + if "monitorList" in group: + monitorList = [] + for monitor in group["monitorList"]: + monitorList.append({"id": monitor.id}) + group["monitorList"] = monitorList + self.api.save_status_page( + slug=self.name, + **kwargs, + ) + self.saved = True + + def __repr__(self): + return f"StatusPage({str(self)})"