forked from DGNum/stateless-uptime-kuma
50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
|
"""
|
||
|
|
Classes to generate the item tree from json spec
|
||
|
|
"""
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import sys
|
||
|
|
|
||
|
|
from .uptime_kuma import Monitor, Notification, Tag
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
def die_tag_format_error():
|
||
|
|
logger.error(
|
||
|
|
"Fatal: You must provide tags in monitors in the format [[name, value]]"
|
||
|
|
)
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
|
||
|
|
def from_dict(api, tree):
|
||
|
|
notif = tree.get("notifications", [])
|
||
|
|
indexed_notifications = {n["name"]: Notification(api, **n) for n in notif}
|
||
|
|
tags = tree.get("tags", [])
|
||
|
|
indexed_tags = {t["name"]: Tag(api, **t) for t in tags}
|
||
|
|
monitors = tree.get("monitors", [])
|
||
|
|
indexed_monitors = {}
|
||
|
|
for m in monitors:
|
||
|
|
associated_tags = []
|
||
|
|
for tag in m.get("tags", []):
|
||
|
|
if not isinstance(tag, list):
|
||
|
|
die_tag_format_error()
|
||
|
|
try:
|
||
|
|
associated_tags.append((indexed_tags[tag[0]], tag[1]))
|
||
|
|
except IndexError:
|
||
|
|
die_tag_format_error()
|
||
|
|
m["tags"] = associated_tags
|
||
|
|
associated_notifications = [
|
||
|
|
indexed_notifications[notif] for notif in m.get("notifications", [])
|
||
|
|
]
|
||
|
|
m["notifications"] = associated_notifications
|
||
|
|
if "name" not in m:
|
||
|
|
logger.error("Fatal: All monitors must have a name")
|
||
|
|
sys.exit(1)
|
||
|
|
indexed_monitors[m["name"]] = Monitor(api, **m)
|
||
|
|
return {
|
||
|
|
"monitors": indexed_monitors.values(),
|
||
|
|
"tags": indexed_tags.values(),
|
||
|
|
"notifications": indexed_notifications.values(),
|
||
|
|
}
|