DGNum/uptime-kuma-api
6
0
Fork 1
Code Issues 17 Pull requests 5 Projects Releases 24 Packages Wiki Activity Actions

Merge branch 'master' into bugfix/memory-leak

Browse source
This commit is contained in:
lucasheld 2023-05-19 14:06:39 +02:00
commit af26eec93f
9 changed files with 117 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
README.md
View file

@ -8,7 +8,12 @@ This package was developed to configure Uptime Kuma with Ansible. The Ansible co
Python version 3.7+ is required.
Supported Uptime Kuma versions: 1.17.0 - 1.21.2
Supported Uptime Kuma versions:
| uptime-kuma-api | Uptime Kuma |
|-----------------|-----------------|
| 1.0.0 | 1.21.3 |
| 0.1.0 - 0.13.0 | 1.17.0 - 1.21.2 |
Installation
---
@ -24,7 +29,7 @@ Documentation
---
The API Reference is available on [Read the Docs](https://uptime-kuma-api.readthedocs.io).
Examples
Example
---
Once you have installed the python package, you can use it to communicate with an Uptime Kuma instance.
@ -49,3 +54,17 @@ At the end, the connection to the API must be disconnected so that the program d
```python
>>> api.disconnect()
```
With a context manager, the disconnect method is called automatically:
```python
from uptime_kuma_api import UptimeKumaApi
with UptimeKumaApi('INSERT_URL') as api:
api.login('INSERT_USERNAME', 'INSERT_PASSWORD')
api.add_monitor(
type=MonitorType.HTTP,
name="Google",
url="https://google.com"
)
```

2
docs/api.rst
View file

@ -39,3 +39,5 @@ Exceptions
----------
.. autoexception:: UptimeKumaException
.. autoexception:: Timeout

2
run_tests.sh
View file

@ -5,7 +5,7 @@ if [ $version ]
then
versions=("$version")
else
versions=(1.21.2 1.21.1 1.20.2 1.19.6 1.18.5 1.17.1)
versions=(1.21.3)
fi
for version in ${versions[*]}

3
setup.py
View file

@ -29,7 +29,7 @@ setup(
author_email="lucasheld@hotmail.de",
license=info["__license__"],
packages=["uptime_kuma_api"],
python_requires=">=3.6, <4",
python_requires=">=3.7, <4",
install_requires=[
"python-socketio[client]>=5.0.0",
"packaging"
@ -43,7 +43,6 @@ setup(
"Operating System :: OS Independent",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",

4
tests/test_maintenance.py
View file

@ -30,7 +30,7 @@ class TestMaintenance(UptimeKumaTestCase):
if parse_version(self.api.version) >= parse_version("1.21.2"):
expected_maintenance.update({
"timezone": "Europe/Berlin"
"timezoneOption": "Europe/Berlin"
})
# add maintenance
@ -242,7 +242,7 @@ class TestMaintenance(UptimeKumaTestCase):
"daysOfMonth": [],
"cron": "50 5 * * *",
"durationMinutes": 120,
"timezone": "Europe/Berlin"
"timezoneOption": "Europe/Berlin"
}
self.do_test_maintenance_strategy(expected_maintenance)

3
uptime_kuma_api/__init__.py
View file

@ -1,11 +1,12 @@
from .__version__ import __title__, __version__, __author__, __copyright__
from .auth_method import AuthMethod
from .monitor_status import MonitorStatus
from .monitor_type import MonitorType
from .notification_providers import NotificationType, notification_provider_options, notification_provider_conditions
from .proxy_protocol import ProxyProtocol
from .incident_style import IncidentStyle
from .docker_type import DockerType
from .maintenance_strategy import MaintenanceStrategy
from .exceptions import UptimeKumaException
from .exceptions import UptimeKumaException, Timeout
from .event import Event
from .api import UptimeKumaApi

96
uptime_kuma_api/api.py
View file

@ -18,9 +18,11 @@ from . import (AuthMethod,
Event,
IncidentStyle,
MaintenanceStrategy,
MonitorStatus,
MonitorType,
NotificationType,
ProxyProtocol,
Timeout,
UptimeKumaException,
notification_provider_conditions,
notification_provider_options)
@ -33,6 +35,7 @@ from .docstrings import (append_docstring,
proxy_docstring,
tag_docstring)
def int_to_bool(data, keys) -> None:
if isinstance(data, list):
for d in data:
@ -43,6 +46,20 @@ def int_to_bool(data, keys) -> None:
data[key] = True if data[key] == 1 else False
def parse_value(data, func) -> None:
if isinstance(data, list):
for d in data:
parse_value(d, func)
else:
func(data)
def parse_monitor_status(data) -> None:
def parse(x):
x["status"] = MonitorStatus(x["status"])
parse_value(data, parse)
def gen_secret(length: int) -> str:
chars = string.ascii_uppercase + string.ascii_lowercase + string.digits
return ''.join(random.choice(chars) for _ in range(length))
@ -147,7 +164,7 @@ def _build_status_page_data(
icon: str = "/icon.svg",
publicGroupList: list = None
) -> tuple(str, dict, str, list):
) -> tuple[str, dict, str, list]:
if theme not in ["light", "dark"]:
raise ValueError
if not domainNameList:
@ -381,7 +398,8 @@ class UptimeKumaApi(object):
)
:param str url: The url to the Uptime Kuma instance. For example ``http://127.0.0.1:3001``
:param float wait_timeout: How many seconds the client should wait for the connection., defaults to 1
:param float timeout: How many seconds the client should wait for the connection, an expected event or a server
response. Default is ``10``.
:param dict headers: Headers that are passed to the socketio connection, defaults to None
:param bool ssl_verify: ``True`` to verify SSL certificates, or ``False`` to skip SSL certificate
verification, allowing connections to servers with self signed certificates.
@ -395,13 +413,13 @@ class UptimeKumaApi(object):
def __init__(
self,
url: str,
wait_timeout: float = 1,
timeout: float = 10,
headers: dict = None,
ssl_verify: bool = True,
wait_events: float = 0.2
) -> None:
self.url = url
self.wait_timeout = wait_timeout
self.timeout = timeout
self.headers = headers
self.wait_events = wait_events
self.sio = socketio.Client(ssl_verify=ssl_verify)
@ -453,26 +471,25 @@ class UptimeKumaApi(object):
@contextmanager
def wait_for_event(self, event: Event) -> None:
# 200 * 0.05 seconds = 10 seconds
retries = 200
sleep = 0.05
# waits for the first event of the given type to arrive
try:
yield
except:
raise
else:
counter = 0
timestamp = time.time()
while self._event_data[event] is None:
time.sleep(sleep)
counter += 1
if counter >= retries:
print(f"wait_for_event {event} timeout")
break
if time.time() - timestamp > self.timeout:
raise Timeout(f"Timed out while waiting for event {event}")
time.sleep(0.01)
def _get_event_data(self, event) -> Any:
monitor_events = [Event.AVG_PING, Event.UPTIME, Event.HEARTBEAT_LIST, Event.IMPORTANT_HEARTBEAT_LIST, Event.CERT_INFO, Event.HEARTBEAT]
timestamp = time.time()
while self._event_data[event] is None:
if time.time() - timestamp > self.timeout:
raise Timeout(f"Timed out while waiting for event {event}")
# do not wait for events that are not sent
if self._event_data[Event.MONITOR_LIST] == {} and event in monitor_events:
return []
@ -481,7 +498,7 @@ class UptimeKumaApi(object):
return deepcopy(self._event_data[event].copy())
def _call(self, event, data=None) -> Any:
r = self.sio.call(event, data)
r = self.sio.call(event, data, timeout=self.timeout)
if isinstance(r, dict) and "ok" in r:
if not r["ok"]:
raise UptimeKumaException(r.get("msg"))
@ -599,7 +616,7 @@ class UptimeKumaApi(object):
"""
url = self.url.rstrip("/")
try:
self.sio.connect(f'{url}/socket.io/', wait_timeout=self.wait_timeout, headers=self.headers)
self.sio.connect(f'{url}/socket.io/', wait_timeout=self.timeout, headers=self.headers)
except:
raise UptimeKumaException("unable to connect")
@ -860,7 +877,7 @@ class UptimeKumaApi(object):
timeRange: list = None,
cron: str = "30 3 * * *",
durationMinutes: int = 60,
timezone: str = None
timezoneOption: str = None
) -> dict:
if not dateRange:
dateRange = [
@ -895,7 +912,7 @@ class UptimeKumaApi(object):
data.update({
"cron": cron,
"durationMinutes": durationMinutes,
"timezone": timezone,
"timezoneOption": timezoneOption,
})
return data
@ -1136,7 +1153,7 @@ class UptimeKumaApi(object):
'monitor_id': 1,
'msg': '200 - OK',
'ping': 201,
'status': True,
'status': <MonitorStatus.UP: 1>,
'time': '2022-12-15 12:38:42.661'
},
{
@ -1147,14 +1164,15 @@ class UptimeKumaApi(object):
'monitor_id': 1,
'msg': '200 - OK',
'ping': 193,
'status': True,
'status': <MonitorStatus.UP: 1>,
'time': '2022-12-15 12:39:42.878'
},
...
]
"""
r = self._call('getMonitorBeats', (id_, hours))["data"]
int_to_bool(r, ["important", "status"])
int_to_bool(r, ["important"])
parse_monitor_status(r)
return r
def get_game_list(self) -> list[dict]:
@ -1739,7 +1757,10 @@ class UptimeKumaApi(object):
}
"""
r1 = self._call('getStatusPage', slug)
r2 = requests.get(f"{self.url}/api/status-page/{slug}").json()
try:
r2 = requests.get(f"{self.url}/api/status-page/{slug}", timeout=self.timeout).json()
except requests.exceptions.Timeout as e:
raise Timeout(e)
config = r1["config"]
config.update(r2["config"])
@ -1975,7 +1996,8 @@ class UptimeKumaApi(object):
"""
r = self._get_event_data(Event.HEARTBEAT_LIST)
for i in r:
int_to_bool(r[i], ["important", "status"])
int_to_bool(r[i], ["important"])
parse_monitor_status(r[i])
return r
def get_important_heartbeats(self) -> dict:
@ -2004,7 +2026,8 @@ class UptimeKumaApi(object):
"""
r = self._get_event_data(Event.IMPORTANT_HEARTBEAT_LIST)
for i in r:
int_to_bool(r[i], ["important", "status"])
int_to_bool(r[i], ["important"])
parse_monitor_status(r[i])
return r
def get_heartbeat(self) -> dict:
@ -2032,7 +2055,8 @@ class UptimeKumaApi(object):
}
"""
r = self._get_event_data(Event.HEARTBEAT)
int_to_bool(r, ["important", "status"])
int_to_bool(r, ["important"])
parse_monitor_status(r)
return r
# avg ping
@ -3054,7 +3078,7 @@ class UptimeKumaApi(object):
],
"cron": "",
"durationMinutes": null,
"timezone": "Europe/Berlin",
"timezoneOption": "Europe/Berlin",
"timezoneOffset": "+02:00",
"status": "ended"
}
@ -3106,7 +3130,7 @@ class UptimeKumaApi(object):
"cron": null,
"duration": null,
"durationMinutes": 0,
"timezone": "Europe/Berlin",
"timezoneOption": "Europe/Berlin",
"timezoneOffset": "+02:00",
"status": "ended"
}
@ -3155,7 +3179,7 @@ class UptimeKumaApi(object):
... ],
... weekdays=[],
... daysOfMonth=[],
... timezone="Europe/Berlin"
... timezoneOption="Europe/Berlin"
... )
{
"msg": "Added Successfully.",
@ -3188,7 +3212,7 @@ class UptimeKumaApi(object):
... ],
... weekdays=[],
... daysOfMonth=[],
... timezone="Europe/Berlin"
... timezoneOption="Europe/Berlin"
... )
{
"msg": "Added Successfully.",
@ -3226,7 +3250,7 @@ class UptimeKumaApi(object):
... 0
... ],
... daysOfMonth=[],
... timezone="Europe/Berlin"
... timezoneOption="Europe/Berlin"
... )
{
"msg": "Added Successfully.",
@ -3265,7 +3289,7 @@ class UptimeKumaApi(object):
... 30,
... "lastDay1"
... ],
... timezone="Europe/Berlin"
... timezoneOption="Europe/Berlin"
... )
{
"msg": "Added Successfully.",
@ -3288,7 +3312,7 @@ class UptimeKumaApi(object):
... daysOfMonth=[],
... cron="50 5 * * *",
... durationMinutes=120,
... timezone="Europe/Berlin"
... timezoneOption="Europe/Berlin"
... )
{
"msg": "Added Successfully.",
@ -3680,3 +3704,13 @@ class UptimeKumaApi(object):
"""
with self.wait_for_event(Event.API_KEY_LIST):
return self._call('deleteAPIKey', id_)
# helper methods
def get_monitor_status(self, monitor_id: int) -> MonitorStatus:
heartbeats = self.get_heartbeats()
for heartbeat in heartbeats:
if int(heartbeat["id"]) == monitor_id:
status = heartbeat["data"][-1]["status"]
return MonitorStatus(status)
raise UptimeKumaException("monitor does not exist")

7
uptime_kuma_api/exceptions.py
View file

@ -2,4 +2,9 @@ class UptimeKumaException(Exception):
"""
There was an exception that occurred while communicating with Uptime Kuma.
"""
pass
class Timeout(UptimeKumaException):
"""
A timeout has occurred while communicating with Uptime Kuma.
"""

17
uptime_kuma_api/monitor_status.py Normal file
View file

@ -0,0 +1,17 @@
from enum import Enum
class MonitorStatus(int, Enum):
"""Enumerate monitor statuses."""
DOWN = 0
"""DOWN"""
UP = 1
"""UP"""
PENDING = 2
"""PENDING"""
MAINTENANCE = 3
"""MAINTENANCE"""