""" 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.sync_settings() 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_settings(self): if "settings" not in self.target_tree: return old = self.api.get_settings() new = self.target_tree["settings"][0] # We need all of the arguments; settings are saved all at once. # We take any settings that are not explicitly set back from the server. for k, v in old.items(): if k not in new.kwargs: new.kwargs[k] = v 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 v is None: v = "" 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) 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)})" class Settings(Item): def __init__(self, api, **kwargs): super().__init__(api, "settings", None, None) self.kwargs = kwargs def save(self): if self.saved: return logger.debug(f"Saving {repr(self)}") self.api.set_settings(**self.kwargs) self.saved = True def __repr__(self): return f"Settings({str(self.kwargs)})"