forked from DGNum/uptime-kuma-api
Clean up code and implement best practices (#27)
* clean up code and implement best practices: - `type(a) == list` replace with `isinstance(a, list)` - `adict['key']` replaced with `adict.get('key')` - annotation `-> list` replace by more accurate `-> list[dict]` * improve compatibility with previous python versions with Typing * little fix
This commit is contained in:
parent
e42f6461c0
commit
19bd8aecfa
1 changed files with 46 additions and 37 deletions
|
@ -1,3 +1,5 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import datetime
|
import datetime
|
||||||
import json
|
import json
|
||||||
import random
|
import random
|
||||||
|
@ -11,21 +13,28 @@ import requests
|
||||||
import socketio
|
import socketio
|
||||||
from packaging.version import parse as parse_version
|
from packaging.version import parse as parse_version
|
||||||
|
|
||||||
from . import AuthMethod
|
from . import (AuthMethod,
|
||||||
from . import DockerType
|
DockerType,
|
||||||
from . import Event
|
Event,
|
||||||
from . import IncidentStyle
|
IncidentStyle,
|
||||||
from . import MaintenanceStrategy
|
MaintenanceStrategy,
|
||||||
from . import MonitorType
|
MonitorType,
|
||||||
from . import NotificationType, notification_provider_options, notification_provider_conditions
|
NotificationType,
|
||||||
from . import ProxyProtocol
|
ProxyProtocol,
|
||||||
from . import UptimeKumaException
|
UptimeKumaException,
|
||||||
from .docstrings import append_docstring, monitor_docstring, notification_docstring, proxy_docstring, \
|
notification_provider_conditions,
|
||||||
docker_host_docstring, maintenance_docstring, tag_docstring
|
notification_provider_options)
|
||||||
|
|
||||||
|
from .docstrings import (append_docstring,
|
||||||
|
docker_host_docstring,
|
||||||
|
maintenance_docstring,
|
||||||
|
monitor_docstring,
|
||||||
|
notification_docstring,
|
||||||
|
proxy_docstring,
|
||||||
|
tag_docstring)
|
||||||
|
|
||||||
def int_to_bool(data, keys) -> None:
|
def int_to_bool(data, keys) -> None:
|
||||||
if type(data) == list:
|
if isinstance(data, list):
|
||||||
for d in data:
|
for d in data:
|
||||||
int_to_bool(d, keys)
|
int_to_bool(d, keys)
|
||||||
else:
|
else:
|
||||||
|
@ -40,7 +49,7 @@ def gen_secret(length: int) -> str:
|
||||||
|
|
||||||
|
|
||||||
def _convert_monitor_return(monitor) -> None:
|
def _convert_monitor_return(monitor) -> None:
|
||||||
if type(monitor["notificationIDList"]) == dict:
|
if isinstance(monitor["notificationIDList"], dict):
|
||||||
monitor["notificationIDList"] = [int(i) for i in monitor["notificationIDList"].keys()]
|
monitor["notificationIDList"] = [int(i) for i in monitor["notificationIDList"].keys()]
|
||||||
|
|
||||||
|
|
||||||
|
@ -138,7 +147,7 @@ def _build_status_page_data(
|
||||||
|
|
||||||
icon: str = "/icon.svg",
|
icon: str = "/icon.svg",
|
||||||
publicGroupList: list = None
|
publicGroupList: list = None
|
||||||
) -> (str, dict, str, list):
|
) -> tuple(str, dict, str, list):
|
||||||
if theme not in ["light", "dark"]:
|
if theme not in ["light", "dark"]:
|
||||||
raise ValueError
|
raise ValueError
|
||||||
if not domainNameList:
|
if not domainNameList:
|
||||||
|
@ -473,9 +482,9 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
def _call(self, event, data=None) -> Any:
|
def _call(self, event, data=None) -> Any:
|
||||||
r = self.sio.call(event, data)
|
r = self.sio.call(event, data)
|
||||||
if type(r) == dict and "ok" in r:
|
if isinstance(r, dict) and "ok" in r:
|
||||||
if not r["ok"]:
|
if not r["ok"]:
|
||||||
raise UptimeKumaException(r["msg"])
|
raise UptimeKumaException(r.get("msg"))
|
||||||
r.pop("ok")
|
r.pop("ok")
|
||||||
return r
|
return r
|
||||||
|
|
||||||
|
@ -594,7 +603,7 @@ class UptimeKumaApi(object):
|
||||||
@property
|
@property
|
||||||
def version(self) -> str:
|
def version(self) -> str:
|
||||||
info = self.info()
|
info = self.info()
|
||||||
return info["version"]
|
return info.get("version")
|
||||||
|
|
||||||
def _build_monitor_data(
|
def _build_monitor_data(
|
||||||
self,
|
self,
|
||||||
|
@ -879,7 +888,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# monitor
|
# monitor
|
||||||
|
|
||||||
def get_monitors(self) -> list:
|
def get_monitors(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all monitors.
|
Get all monitors.
|
||||||
|
|
||||||
|
@ -1092,7 +1101,7 @@ class UptimeKumaApi(object):
|
||||||
with self.wait_for_event(Event.MONITOR_LIST):
|
with self.wait_for_event(Event.MONITOR_LIST):
|
||||||
return self._call('deleteMonitor', id_)
|
return self._call('deleteMonitor', id_)
|
||||||
|
|
||||||
def get_monitor_beats(self, id_: int, hours: int) -> list:
|
def get_monitor_beats(self, id_: int, hours: int) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get monitor beats for a specific monitor in a time range.
|
Get monitor beats for a specific monitor in a time range.
|
||||||
|
|
||||||
|
@ -1135,7 +1144,7 @@ class UptimeKumaApi(object):
|
||||||
int_to_bool(r, ["important", "status"])
|
int_to_bool(r, ["important", "status"])
|
||||||
return r
|
return r
|
||||||
|
|
||||||
def get_game_list(self) -> list:
|
def get_game_list(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get a list of games that are supported by the GameDig monitor type.
|
Get a list of games that are supported by the GameDig monitor type.
|
||||||
|
|
||||||
|
@ -1176,7 +1185,7 @@ class UptimeKumaApi(object):
|
||||||
# Exists in 1.20.0 - 1.21.0
|
# Exists in 1.20.0 - 1.21.0
|
||||||
if not r:
|
if not r:
|
||||||
r = self._call('getGameList')
|
r = self._call('getGameList')
|
||||||
return r["gameList"]
|
return r.get("gameList")
|
||||||
|
|
||||||
@append_docstring(monitor_docstring("add"))
|
@append_docstring(monitor_docstring("add"))
|
||||||
def add_monitor(self, **kwargs) -> dict:
|
def add_monitor(self, **kwargs) -> dict:
|
||||||
|
@ -1294,7 +1303,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# notification
|
# notification
|
||||||
|
|
||||||
def get_notifications(self) -> list:
|
def get_notifications(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all notifications.
|
Get all notifications.
|
||||||
|
|
||||||
|
@ -1489,7 +1498,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# proxy
|
# proxy
|
||||||
|
|
||||||
def get_proxies(self) -> list:
|
def get_proxies(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all proxies.
|
Get all proxies.
|
||||||
|
|
||||||
|
@ -1547,7 +1556,7 @@ class UptimeKumaApi(object):
|
||||||
"""
|
"""
|
||||||
proxies = self.get_proxies()
|
proxies = self.get_proxies()
|
||||||
for proxy in proxies:
|
for proxy in proxies:
|
||||||
if proxy["id"] == id_:
|
if proxy.get("id") == id_:
|
||||||
return proxy
|
return proxy
|
||||||
raise UptimeKumaException("proxy does not exist")
|
raise UptimeKumaException("proxy does not exist")
|
||||||
|
|
||||||
|
@ -1633,7 +1642,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# status page
|
# status page
|
||||||
|
|
||||||
def get_status_pages(self) -> list:
|
def get_status_pages(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all status pages.
|
Get all status pages.
|
||||||
|
|
||||||
|
@ -1914,7 +1923,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# heartbeat
|
# heartbeat
|
||||||
|
|
||||||
def get_heartbeats(self) -> list:
|
def get_heartbeats(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get heartbeats.
|
Get heartbeats.
|
||||||
|
|
||||||
|
@ -1961,7 +1970,7 @@ class UptimeKumaApi(object):
|
||||||
int_to_bool(i["data"], ["important", "status"])
|
int_to_bool(i["data"], ["important", "status"])
|
||||||
return r
|
return r
|
||||||
|
|
||||||
def get_important_heartbeats(self) -> list:
|
def get_important_heartbeats(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get important heartbeats.
|
Get important heartbeats.
|
||||||
|
|
||||||
|
@ -1994,7 +2003,7 @@ class UptimeKumaApi(object):
|
||||||
int_to_bool(i["data"], ["important", "status"])
|
int_to_bool(i["data"], ["important", "status"])
|
||||||
return r
|
return r
|
||||||
|
|
||||||
def get_heartbeat(self) -> list:
|
def get_heartbeat(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get heartbeat.
|
Get heartbeat.
|
||||||
|
|
||||||
|
@ -2021,7 +2030,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# avg ping
|
# avg ping
|
||||||
|
|
||||||
def avg_ping(self) -> list:
|
def avg_ping(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get average ping.
|
Get average ping.
|
||||||
|
|
||||||
|
@ -2042,7 +2051,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# cert info
|
# cert info
|
||||||
|
|
||||||
def cert_info(self) -> list:
|
def cert_info(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get certificate info.
|
Get certificate info.
|
||||||
|
|
||||||
|
@ -2063,7 +2072,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# uptime
|
# uptime
|
||||||
|
|
||||||
def uptime(self) -> list:
|
def uptime(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get monitor uptime.
|
Get monitor uptime.
|
||||||
|
|
||||||
|
@ -2162,7 +2171,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# tags
|
# tags
|
||||||
|
|
||||||
def get_tags(self) -> list:
|
def get_tags(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all tags.
|
Get all tags.
|
||||||
|
|
||||||
|
@ -2717,7 +2726,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# docker host
|
# docker host
|
||||||
|
|
||||||
def get_docker_hosts(self) -> list:
|
def get_docker_hosts(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all docker hosts.
|
Get all docker hosts.
|
||||||
|
|
||||||
|
@ -2862,7 +2871,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# maintenance
|
# maintenance
|
||||||
|
|
||||||
def get_maintenances(self) -> list:
|
def get_maintenances(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all maintenances.
|
Get all maintenances.
|
||||||
|
|
||||||
|
@ -3251,7 +3260,7 @@ class UptimeKumaApi(object):
|
||||||
"""
|
"""
|
||||||
return self._call('resumeMaintenance', id_)
|
return self._call('resumeMaintenance', id_)
|
||||||
|
|
||||||
def get_monitor_maintenance(self, id_: int) -> list:
|
def get_monitor_maintenance(self, id_: int) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Gets all monitors of a maintenance.
|
Gets all monitors of a maintenance.
|
||||||
|
|
||||||
|
@ -3309,7 +3318,7 @@ class UptimeKumaApi(object):
|
||||||
"""
|
"""
|
||||||
return self._call('addMonitorMaintenance', (id_, monitors))
|
return self._call('addMonitorMaintenance', (id_, monitors))
|
||||||
|
|
||||||
def get_status_page_maintenance(self, id_: int) -> list:
|
def get_status_page_maintenance(self, id_: int) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Gets all status pages of a maintenance.
|
Gets all status pages of a maintenance.
|
||||||
|
|
||||||
|
@ -3365,7 +3374,7 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# api key
|
# api key
|
||||||
|
|
||||||
def get_api_keys(self) -> list:
|
def get_api_keys(self) -> list[dict]:
|
||||||
"""
|
"""
|
||||||
Get all api keys.
|
Get all api keys.
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue