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

feat: add param wait_events

Browse source
This commit is contained in:
lucasheld 2023-04-30 16:48:33 +02:00
parent 31fe4466a2
commit 391e5a3077
1 changed files with 8 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
uptime_kuma_api/api.py
View file

@ -360,6 +360,10 @@ class UptimeKumaApi(object):
:param bool ssl_verify: ``True`` to verify SSL certificates, or ``False`` to skip SSL certificate :param bool ssl_verify: ``True`` to verify SSL certificates, or ``False`` to skip SSL certificate
verification, allowing connections to servers with self signed certificates. verification, allowing connections to servers with self signed certificates.
Default is ``True``. Default is ``True``.
:param float wait_events: How many seconds the client should wait for the next event of the same type.
There is no way to determine when the last message of a certain type has arrived.
Therefore, a timeout is required. If no further message has arrived within this time,
it is assumed that it was the last message. Defaults is ``0.2``.
:raises UptimeKumaException: When connection to server failed. :raises UptimeKumaException: When connection to server failed.
""" """
def __init__( def __init__(
@ -367,11 +371,13 @@ class UptimeKumaApi(object):
url: str, url: str,
wait_timeout: float = 1, wait_timeout: float = 1,
headers: dict = None, headers: dict = None,
ssl_verify: bool = True ssl_verify: bool = True,
wait_events: float = 0.2
) -> None: ) -> None:
self.url = url self.url = url
self.wait_timeout = wait_timeout self.wait_timeout = wait_timeout
self.headers = headers self.headers = headers
self.wait_events = wait_events
self.sio = socketio.Client(ssl_verify=ssl_verify) self.sio = socketio.Client(ssl_verify=ssl_verify)
self._event_data: dict = { self._event_data: dict = {
@ -439,7 +445,7 @@ class UptimeKumaApi(object):
if self._event_data[Event.MONITOR_LIST] == {} and event in monitor_events: if self._event_data[Event.MONITOR_LIST] == {} and event in monitor_events:
return [] return []
time.sleep(0.01) time.sleep(0.01)
time.sleep(0.05) # wait for multiple messages time.sleep(self.wait_events) # wait for multiple messages
return deepcopy(self._event_data[event]) return deepcopy(self._event_data[event])
def _call(self, event, data=None) -> Any: def _call(self, event, data=None) -> Any: