stateless-uptime-kuma/stateless_uptime_kuma/uptime_kuma.py

"""
Classes to make the needed operations to reach the specified state.
"""

import logging
import sys

logger = logging.getLogger(__name__)


class Manager:
    def __init__(self, api, target_tree={}, prune_unused=False):
        self.api = api
        self.prune_unused = prune_unused
        self.target_tree = target_tree

    def process(self):
        self.sync_tags()
        self.sync_notifications()
        self.sync_monitors()
        self.sync_status_pages()
        self.save()

    def save(self):
        for v in self.target_tree.values():
            for i in v:
                i.save()  # this method should be safe to be called in whatever order

    def sync_monitors(self):
        old = self.api.get_monitors()
        new = self.target_tree.get("monitors", [])
        self.sync(new, old, "name")

    def sync_notifications(self):
        old = self.api.get_notifications()
        new = self.target_tree.get("notifications", [])
        self.sync(new, old, "name")

    def sync_tags(self):
        old = self.api.get_tags()
        new = self.target_tree.get("tags", [])
        self.sync(new, old, "name")

    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"]
                logger.debug(f"Synced item named {k}")
                if k.old_name is not None:
                    logger.warn(f"Found unused oldName for {k}")
            elif k.old_name in indexed_old:
                k.id = indexed_old[k.old_name]["id"]
                logger.info(f"Found renamed item {k.old_name} -> {k}")
            else:
                k.id = None  # Useless
                logger.debug(f"Creating key {k}")


class Item:
    def __init__(self, api, name, id, old_name=None):
        self.api = api
        self.name = name
        self.id = id
        self.old_name = old_name
        self.saved = False

    def save(self):
        raise NotImplementedError()

    def __setattr__(self, name, value):
        if name != "saved":
            self.saved = False
        object.__setattr__(self, name, value)

    def __str__(self):
        return self.name


class Monitor(Item):
    def __init__(
        self, api, name, id=None, old_name=None, tags=[], notifications=[], **kwargs
    ):
        super().__init__(api, name, id, old_name)
        self.kwargs = kwargs
        self.tags = tags
        self.notifications = notifications
        self.saved = False

    def get_status_codes(self):
        if "accepted_statuscodes" not in self.kwargs:
            return set(range(200, 300))
        accepted_statuscodes = set()
        for codes in self.kwargs["accepted_statuscodes"]:
            if "-" in codes:
                c = codes.split("-")
                if len(c) != 2:
                    logger.error(
                        f"Fatal: status codes for {self} must be in the format 'number' or 'number-number'"
                    )
                    sys.exit(1)
                a, b = int(c[0]), int(c[1])
                accepted_statuscodes.update(range(a, b))
            else:
                accepted_statuscodes.add(int(codes))
        return accepted_statuscodes

    def save(self):
        if self.saved:
            return
        logger.debug(f"Saving {repr(self)}")
        for t, _ in self.tags:
            t.save()
        for n in self.notifications:
            n.save()
        if self.id is None:
            rslt = self.api.add_monitor(
                name=self.name,
                notificationIDList=[i.id for i in self.notifications],
                **self.kwargs,
            )
            self.id = rslt["monitorID"]
            for t, value in self.tags:
                self.api.add_monitor_tag(tag_id=t.id, monitor_id=self.id, value=value)
        else:
            rslt = self.api.edit_monitor(
                self.id,
                name=self.name,
                notificationIDList=[i.id for i in self.notifications],
                **self.kwargs,
            )
            current_tags = set(
                (i["tag_id"], i["value"]) for i in self.api.get_monitor(self.id)["tags"]
            )
            for t, v in self.tags:
                if (t.id, v) not in current_tags:
                    logger.debug(f"Adding tag: '{t}, {v}'")
                    self.api.add_monitor_tag(tag_id=t.id, monitor_id=self.id, value=v if v is not None else "")
        self.saved = True

    def __repr__(self):
        return f"Monitor({str(self)})"


class Tag(Item):
    def __init__(self, api, name, id=None, old_name=None, color="#000000"):
        super().__init__(api, name, id, old_name)
        self.tag = name
        self.color = color

    def save(self):
        if self.saved:
            return
        logger.debug(f"Saving {repr(self)}")
        if self.id is None:
            rslt = self.api.add_tag(
                name=self.tag,
                color=self.color,
            )
            self.id = rslt["id"]
        else:
            self.api.edit_tag(
                id_=self.id,
                name=self.tag,
                color=self.color,
            )
        self.saved = True

    def __repr__(self):
        return f"Tag({str(self)})"


class Notification(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:
            rslt = self.api.add_notification(
                name=self.name,
                **self.kwargs,
            )
            self.id = rslt["id"]
        else:
            self.api.edit_notification(
                id_=self.id,
                name=self.name,
                **self.kwargs,
            )
        self.saved = True

    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)})"