Merge branch 'master' into bugfix/memory-leak

This commit is contained in:
lucasheld 2023-05-01 20:26:49 +02:00
commit 7771d51f12
10 changed files with 298 additions and 182 deletions

View file

@ -1,5 +1,19 @@
## Changelog ## Changelog
### Release 0.13.0
#### Feature
- add support for uptime kuma 1.21.2
- implement custom socketio headers
#### Bugfix
- do not wait for events that have already arrived
### Release 0.12.0
#### Feature
- add support for uptime kuma 1.21.1
### Release 0.11.0 ### Release 0.11.0
#### Feature #### Feature

View file

@ -6,9 +6,9 @@ uptime-kuma-api is a Python wrapper for the [Uptime Kuma](https://github.com/lou
This package was developed to configure Uptime Kuma with Ansible. The Ansible collection can be found at https://github.com/lucasheld/ansible-uptime-kuma. This package was developed to configure Uptime Kuma with Ansible. The Ansible collection can be found at https://github.com/lucasheld/ansible-uptime-kuma.
Python version 3.6+ is required. Python version 3.7+ is required.
Supported Uptime Kuma versions: 1.17.0 - 1.21.0 Supported Uptime Kuma versions: 1.17.0 - 1.21.2
Installation Installation
--- ---

View file

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

View file

@ -24,20 +24,15 @@ class TestMaintenance(UptimeKumaTestCase):
"2022-12-27 22:36:00", "2022-12-27 22:36:00",
"2022-12-29 22:36:00" "2022-12-29 22:36:00"
], ],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [], "weekdays": [],
"daysOfMonth": [] "daysOfMonth": []
} }
if parse_version(self.api.version) >= parse_version("1.21.2"):
expected_maintenance.update({
"timezone": "Europe/Berlin"
})
# add maintenance # add maintenance
r = self.api.add_maintenance(**expected_maintenance) r = self.api.add_maintenance(**expected_maintenance)
self.assertEqual(r["msg"], "Added Successfully.") self.assertEqual(r["msg"], "Added Successfully.")
@ -119,18 +114,6 @@ class TestMaintenance(UptimeKumaTestCase):
"dateRange": [ "dateRange": [
"2022-12-27 00:00:00" "2022-12-27 00:00:00"
], ],
"timeRange": [
{
"hours": 2,
"minutes": 0,
"seconds": 0
},
{
"hours": 3,
"minutes": 0,
"seconds": 0
}
],
"weekdays": [], "weekdays": [],
"daysOfMonth": [] "daysOfMonth": []
} }
@ -147,16 +130,6 @@ class TestMaintenance(UptimeKumaTestCase):
"2022-12-27 22:36:00", "2022-12-27 22:36:00",
"2022-12-29 22:36:00" "2022-12-29 22:36:00"
], ],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [], "weekdays": [],
"daysOfMonth": [] "daysOfMonth": []
} }
@ -246,12 +219,33 @@ class TestMaintenance(UptimeKumaTestCase):
10, 10,
20, 20,
30, 30,
"lastDay4", "lastDay1"
"lastDay2"
] ]
} }
self.do_test_maintenance_strategy(expected_maintenance) self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_cron(self):
if parse_version(self.api.version) < parse_version("1.21.2"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.CRON,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:37:00",
"2022-12-31 22:37:00"
],
"weekdays": [],
"daysOfMonth": [],
"cron": "50 5 * * *",
"durationMinutes": 120,
"timezone": "Europe/Berlin"
}
self.do_test_maintenance_strategy(expected_maintenance)
def do_test_maintenance_strategy(self, expected_maintenance): def do_test_maintenance_strategy(self, expected_maintenance):
# add maintenance # add maintenance
r = self.api.add_maintenance(**expected_maintenance) r = self.api.add_maintenance(**expected_maintenance)

View file

@ -80,6 +80,18 @@ class UptimeKumaTestCase(unittest.TestCase):
for docker_host in docker_hosts: for docker_host in docker_hosts:
self.api.delete_docker_host(docker_host["id"]) self.api.delete_docker_host(docker_host["id"])
if parse_version(self.api.version) >= parse_version("1.19"):
# delete maintenances
maintenances = self.api.get_maintenances()
for maintenance in maintenances:
self.api.delete_maintenance(maintenance["id"])
if parse_version(self.api.version) >= parse_version("1.21"):
# delete api keys
api_keys = self.api.get_api_keys()
for api_key in api_keys:
self.api.delete_api_key(api_key["id"])
# login again to receive initial messages # login again to receive initial messages
self.api.disconnect() self.api.disconnect()
self.api = UptimeKumaApi(self.url) self.api = UptimeKumaApi(self.url)

View file

@ -1,5 +1,5 @@
__title__ = "uptime_kuma_api" __title__ = "uptime_kuma_api"
__version__ = "0.11.0" __version__ = "0.13.0"
__author__ = "Lucas Held" __author__ = "Lucas Held"
__license__ = "MIT" __license__ = "MIT"
__copyright__ = "Copyright 2023 Lucas Held" __copyright__ = "Copyright 2023 Lucas Held"

View file

@ -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:
@ -184,49 +193,6 @@ def _build_docker_host_data(
return data return data
def _build_maintenance_data(
title: str,
strategy: MaintenanceStrategy,
active: bool = True,
description: str = "",
dateRange: list = None,
intervalDay: int = 1,
weekdays: list = None,
daysOfMonth: list = None,
timeRange: list = None
) -> dict:
if not dateRange:
dateRange = [
datetime.date.today().strftime("%Y-%m-%d 00:00:00")
]
if not timeRange:
timeRange = [
{
"hours": 2,
"minutes": 0,
}, {
"hours": 3,
"minutes": 0,
}
]
if not weekdays:
weekdays = []
if not daysOfMonth:
daysOfMonth = []
data = {
"title": title,
"active": active,
"intervalDay": intervalDay,
"dateRange": dateRange,
"description": description,
"strategy": strategy,
"weekdays": weekdays,
"daysOfMonth": daysOfMonth,
"timeRange": timeRange
}
return data
def _build_tag_data( def _build_tag_data(
name: str, name: str,
color: str color: str
@ -380,6 +346,9 @@ class UptimeKumaApi(object):
>>> from uptime_kuma_api import UptimeKumaApi >>> from uptime_kuma_api import UptimeKumaApi
>>> api = UptimeKumaApi('INSERT_URL') >>> api = UptimeKumaApi('INSERT_URL')
>>> api.login('INSERT_USERNAME', 'INSERT_PASSWORD') >>> api.login('INSERT_USERNAME', 'INSERT_PASSWORD')
{
'token': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6ImFkbWluIiwiaWF0IjoxNjgyOTU4OTU4fQ.Xb81nuKXeNyE1D_XoQowYgsgZHka-edONdwHmIznJdk'
}
Now you can call one of the existing methods of the instance. For example create a new monitor: Now you can call one of the existing methods of the instance. For example create a new monitor:
@ -397,18 +366,45 @@ class UptimeKumaApi(object):
>>> api.disconnect() >>> api.disconnect()
With a context manager, the disconnect method is called automatically:
.. code-block:: 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"
)
:param str url: The url to the Uptime Kuma instance. For example ``http://127.0.0.1:3001`` :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 wait_timeout: How many seconds the client should wait for the connection., defaults to 1
: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.
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__(
self, self,
url: str, url: str,
wait_timeout: float = 1 wait_timeout: float = 1,
headers: dict = None,
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.sio = socketio.Client() self.headers = headers
self.wait_events = wait_events
self.sio = socketio.Client(ssl_verify=ssl_verify)
self._event_data: dict = { self._event_data: dict = {
Event.MONITOR_LIST: None, Event.MONITOR_LIST: None,
@ -449,10 +445,17 @@ class UptimeKumaApi(object):
self.connect() self.connect()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.disconnect()
@contextmanager @contextmanager
def wait_for_event(self, event: Event) -> None: def wait_for_event(self, event: Event) -> None:
# 200 * 0.05 seconds = 10 seconds
retries = 200 retries = 200
event_data_before = deepcopy(self._event_data) sleep = 0.05
try: try:
yield yield
@ -460,11 +463,11 @@ class UptimeKumaApi(object):
raise raise
else: else:
counter = 0 counter = 0
while event_data_before[event] == self._event_data[event]: while self._event_data[event] is None:
time.sleep(0.01) time.sleep(sleep)
counter += 1 counter += 1
if counter >= retries: if counter >= retries:
print("wait_for_event timeout") print(f"wait_for_event {event} timeout")
break break
def _get_event_data(self, event) -> Any: def _get_event_data(self, event) -> Any:
@ -474,14 +477,14 @@ 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].copy()) return deepcopy(self._event_data[event].copy())
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
@ -596,7 +599,7 @@ class UptimeKumaApi(object):
""" """
url = self.url.rstrip("/") url = self.url.rstrip("/")
try: try:
self.sio.connect(f'{url}/socket.io/', wait_timeout=self.wait_timeout) self.sio.connect(f'{url}/socket.io/', wait_timeout=self.wait_timeout, headers=self.headers)
except: except:
raise UptimeKumaException("unable to connect") raise UptimeKumaException("unable to connect")
@ -613,7 +616,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,
@ -623,7 +626,7 @@ class UptimeKumaApi(object):
interval: int = 60, interval: int = 60,
retryInterval: int = 60, retryInterval: int = 60,
resendInterval: int = 0, resendInterval: int = 0,
maxretries: int = 0, maxretries: int = 1,
upsideDown: bool = False, upsideDown: bool = False,
notificationIDList: list = None, notificationIDList: list = None,
httpBodyEncoding: str = "json", httpBodyEncoding: str = "json",
@ -698,6 +701,10 @@ class UptimeKumaApi(object):
# GAMEDIG # GAMEDIG
game: str = None game: str = None
) -> dict: ) -> dict:
# https://github.com/louislam/uptime-kuma/compare/1.21.1...1.21.2#diff-f672603317047f3e6f27b0d7a44f6f244b7dbb5d0d0a85f1059a6b0bc2cb9aa0L910
if parse_version(self.version) < parse_version("1.21.2"):
maxretries = 0
data = { data = {
"type": type, "type": type,
"name": name, "name": name,
@ -840,9 +847,61 @@ class UptimeKumaApi(object):
return data return data
def _build_maintenance_data(
self,
title: str,
strategy: MaintenanceStrategy,
active: bool = True,
description: str = "",
dateRange: list = None,
intervalDay: int = 1,
weekdays: list = None,
daysOfMonth: list = None,
timeRange: list = None,
cron: str = "30 3 * * *",
durationMinutes: int = 60,
timezone: str = None
) -> dict:
if not dateRange:
dateRange = [
datetime.date.today().strftime("%Y-%m-%d 00:00:00")
]
if not timeRange:
timeRange = [
{
"hours": 2,
"minutes": 0,
}, {
"hours": 3,
"minutes": 0,
}
]
if not weekdays:
weekdays = []
if not daysOfMonth:
daysOfMonth = []
data = {
"title": title,
"active": active,
"intervalDay": intervalDay,
"dateRange": dateRange,
"description": description,
"strategy": strategy,
"weekdays": weekdays,
"daysOfMonth": daysOfMonth,
"timeRange": timeRange
}
if parse_version(self.version) >= parse_version("1.21.2"):
data.update({
"cron": cron,
"durationMinutes": durationMinutes,
"timezone": timezone,
})
return data
# monitor # monitor
def get_monitors(self) -> list: def get_monitors(self) -> list[dict]:
""" """
Get all monitors. Get all monitors.
@ -887,7 +946,7 @@ class UptimeKumaApi(object):
'keyword': None, 'keyword': None,
'maintenance': False, 'maintenance': False,
'maxredirects': 10, 'maxredirects': 10,
'maxretries': 0, 'maxretries': 1,
'method': 'GET', 'method': 'GET',
'mqttPassword': None, 'mqttPassword': None,
'mqttSuccessMessage': None, 'mqttSuccessMessage': None,
@ -969,7 +1028,7 @@ class UptimeKumaApi(object):
'keyword': None, 'keyword': None,
'maintenance': False, 'maintenance': False,
'maxredirects': 10, 'maxredirects': 10,
'maxretries': 0, 'maxretries': 1,
'method': 'GET', 'method': 'GET',
'mqttPassword': None, 'mqttPassword': None,
'mqttSuccessMessage': None, 'mqttSuccessMessage': None,
@ -1055,7 +1114,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.
@ -1098,7 +1157,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.
@ -1139,7 +1198,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:
@ -1257,7 +1316,7 @@ class UptimeKumaApi(object):
# notification # notification
def get_notifications(self) -> list: def get_notifications(self) -> list[dict]:
""" """
Get all notifications. Get all notifications.
@ -1452,7 +1511,7 @@ class UptimeKumaApi(object):
# proxy # proxy
def get_proxies(self) -> list: def get_proxies(self) -> list[dict]:
""" """
Get all proxies. Get all proxies.
@ -1510,7 +1569,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")
@ -1596,7 +1655,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.
@ -2252,7 +2311,7 @@ class UptimeKumaApi(object):
# tags # tags
def get_tags(self) -> list: def get_tags(self) -> list[dict]:
""" """
Get all tags. Get all tags.
@ -2807,7 +2866,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.
@ -2952,7 +3011,7 @@ class UptimeKumaApi(object):
# maintenance # maintenance
def get_maintenances(self) -> list: def get_maintenances(self) -> list[dict]:
""" """
Get all maintenances. Get all maintenances.
@ -2977,29 +3036,27 @@ class UptimeKumaApi(object):
], ],
"timeRange": [ "timeRange": [
{ {
"hours": 2, "hours": 0,
"minutes": 0, "minutes": 0
"seconds": 0
}, },
{ {
"hours": 3, "hours": 0,
"minutes": 0, "minutes": 0
"seconds": 0
} }
], ],
"weekdays": [], "weekdays": [],
"daysOfMonth": [], "daysOfMonth": [],
"timeslotList": [ "timeslotList": [
{ {
"id": 1, "startDate": "2022-12-27 22:36:00",
"startDate": "2022-12-27 14:39:00", "endDate": "2022-12-29 22:36:00"
"endDate": "2022-12-30 14:39:00",
"startDateServerTimezone": "2022-12-27 15:39",
"endDateServerTimezone": "2022-12-30 15:39",
"serverTimezoneOffset": "+01:00"
} }
], ],
"status": "under-maintenance" "cron": "",
"durationMinutes": null,
"timezone": "Europe/Berlin",
"timezoneOffset": "+02:00",
"status": "ended"
} }
] ]
""" """
@ -3030,29 +3087,28 @@ class UptimeKumaApi(object):
], ],
"timeRange": [ "timeRange": [
{ {
"hours": 2, "hours": 0,
"minutes": 0, "minutes": 0
"seconds": 0
}, },
{ {
"hours": 3, "hours": 0,
"minutes": 0, "minutes": 0
"seconds": 0
} }
], ],
"weekdays": [], "weekdays": [],
"daysOfMonth": [], "daysOfMonth": [],
"timeslotList": [ "timeslotList": [
{ {
"id": 1, "startDate": "2022-12-27 22:36:00",
"startDate": "2022-12-27 14:39:00", "endDate": "2022-12-29 22:36:00"
"endDate": "2022-12-30 14:39:00",
"startDateServerTimezone": "2022-12-27 15:39",
"endDateServerTimezone": "2022-12-30 15:39",
"serverTimezoneOffset": "+01:00"
} }
], ],
"status": "under-maintenance" "cron": null,
"duration": null,
"durationMinutes": 0,
"timezone": "Europe/Berlin",
"timezoneOffset": "+02:00",
"status": "ended"
} }
""" """
return self._call('getMaintenance', id_)["maintenance"] return self._call('getMaintenance', id_)["maintenance"]
@ -3077,18 +3133,6 @@ class UptimeKumaApi(object):
... dateRange=[ ... dateRange=[
... "2022-12-27 00:00:00" ... "2022-12-27 00:00:00"
... ], ... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0,
... "seconds": 0
... },
... {
... "hours": 3,
... "minutes": 0,
... "seconds": 0
... }
... ],
... weekdays=[], ... weekdays=[],
... daysOfMonth=[] ... daysOfMonth=[]
... ) ... )
@ -3109,20 +3153,9 @@ class UptimeKumaApi(object):
... "2022-12-27 22:36:00", ... "2022-12-27 22:36:00",
... "2022-12-29 22:36:00" ... "2022-12-29 22:36:00"
... ], ... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0,
... "seconds": 0
... },
... {
... "hours": 3,
... "minutes": 0,
... "seconds": 0
... }
... ],
... weekdays=[], ... weekdays=[],
... daysOfMonth=[] ... daysOfMonth=[],
... timezone="Europe/Berlin"
... ) ... )
{ {
"msg": "Added Successfully.", "msg": "Added Successfully.",
@ -3154,7 +3187,8 @@ class UptimeKumaApi(object):
... } ... }
... ], ... ],
... weekdays=[], ... weekdays=[],
... daysOfMonth=[] ... daysOfMonth=[],
... timezone="Europe/Berlin"
... ) ... )
{ {
"msg": "Added Successfully.", "msg": "Added Successfully.",
@ -3191,7 +3225,8 @@ class UptimeKumaApi(object):
... 5, ... 5,
... 0 ... 0
... ], ... ],
... daysOfMonth=[] ... daysOfMonth=[],
... timezone="Europe/Berlin"
... ) ... )
{ {
"msg": "Added Successfully.", "msg": "Added Successfully.",
@ -3228,15 +3263,39 @@ class UptimeKumaApi(object):
... 10, ... 10,
... 20, ... 20,
... 30, ... 30,
... "lastDay2" ... "lastDay1"
... ] ... ],
... timezone="Europe/Berlin"
... )
{
"msg": "Added Successfully.",
"maintenanceID": 1
}
Example (strategy: :attr:`~.MaintenanceStrategy.CRON`)::
>>> api.add_maintenance(
... title="test",
... description="test",
... strategy=MaintenanceStrategy.CRON,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 22:39:00",
... "2022-12-31 22:39:00"
... ],
... weekdays=[],
... daysOfMonth=[],
... cron="50 5 * * *",
... durationMinutes=120,
... timezone="Europe/Berlin"
... ) ... )
{ {
"msg": "Added Successfully.", "msg": "Added Successfully.",
"maintenanceID": 1 "maintenanceID": 1
} }
""" """
data = _build_maintenance_data(**kwargs) data = self._build_maintenance_data(**kwargs)
_check_arguments_maintenance(data) _check_arguments_maintenance(data)
return self._call('addMaintenance', data) return self._call('addMaintenance', data)
@ -3341,7 +3400,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.
@ -3399,7 +3458,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.
@ -3455,7 +3514,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.

View file

@ -109,8 +109,8 @@ def notification_docstring(mode) -> str:
:param str, optional lineChannelAccessToken: Notification option for ``type`` :attr:`~.NotificationType.LINE` :param str, optional lineChannelAccessToken: Notification option for ``type`` :attr:`~.NotificationType.LINE`
:param str, optional lineUserID: Notification option for ``type`` :attr:`~.NotificationType.LINE` :param str, optional lineUserID: Notification option for ``type`` :attr:`~.NotificationType.LINE`
:param str, optional lunaseaTarget: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. Allowed values: "device", "user". :param str, optional lunaseaTarget: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. Allowed values: "device", "user".
:param str, optional lunaseaUserID: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA` :param str, optional lunaseaUserID: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. User ID.
:param str, optional lunaseaDevice: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA` :param str, optional lunaseaDevice: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. Device ID.
:param str, optional internalRoomId: Notification option for ``type`` :attr:`~.NotificationType.MATRIX` :param str, optional internalRoomId: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`
:param str, optional accessToken: Notification option for ``type`` :attr:`~.NotificationType.MATRIX` :param str, optional accessToken: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`
:param str, optional homeserverUrl: Notification option for ``type`` :attr:`~.NotificationType.MATRIX` :param str, optional homeserverUrl: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`
@ -275,6 +275,16 @@ def notification_docstring(mode) -> str:
- ``WARNING`` - ``WARNING``
- ``CRITICAL`` - ``CRITICAL``
:param str, optional splunkRestURL: Notification option for ``type`` :attr:`~.NotificationType.SPLUNK`. Splunk Rest URL. :param str, optional splunkRestURL: Notification option for ``type`` :attr:`~.NotificationType.SPLUNK`. Splunk Rest URL.
:param int, optional opsgeniePriority: Notification option for ``type`` :attr:`~.NotificationType.OPSGENIE`. Priority. Available values are numbers between ``1`` and ``5``.
:param str, optional opsgenieRegion: Notification option for ``type`` :attr:`~.NotificationType.OPSGENIE`. Region. Available values are:
- ``us``: US (Default)
- ``eu``: EU
:param str, optional opsgenieApiKey: Notification option for ``type`` :attr:`~.NotificationType.OPSGENIE`. API Key.
:param str, optional twilioAccountSID: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. Account SID.
:param str, optional twilioAuthToken: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. Auth Token.
:param str, optional twilioToNumber: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. To Number.
:param str, optional twilioFromNumber: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. From Number.
""" """
@ -309,8 +319,11 @@ def maintenance_docstring(mode) -> str:
:param list, optional dateRange: DateTime Range, defaults to ``["<current date>"]`` :param list, optional dateRange: DateTime Range, defaults to ``["<current date>"]``
:param int, optional intervalDay: Interval (Run once every day), defaults to ``1`` :param int, optional intervalDay: Interval (Run once every day), defaults to ``1``
:param list, optional weekdays: List that contains the days of the week on which the maintenance is enabled (Sun = ``0``, Mon = ``1``, ..., Sat = ``6``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_WEEKDAY`., defaults to ``[]``. :param list, optional weekdays: List that contains the days of the week on which the maintenance is enabled (Sun = ``0``, Mon = ``1``, ..., Sat = ``6``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_WEEKDAY`., defaults to ``[]``.
:param list, optional daysOfMonth: List that contains the days of the month on which the maintenance is enabled (Day 1 = ``1``, Day 2 = ``2``, ..., Day 31 = ``31``) and the last day of the month (Last Day of Month = ``"lastDay1"``, 2nd Last Day of Month = ``"lastDay2"``, 3rd Last Day of Month = ``"lastDay3"``, 4th Last Day of Month = ``"lastDay4"``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_DAY_OF_MONTH`., defaults to ``[]``. :param list, optional daysOfMonth: List that contains the days of the month on which the maintenance is enabled (Day 1 = ``1``, Day 2 = ``2``, ..., Day 31 = ``31``) and the last day of the month (``"lastDay1"``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_DAY_OF_MONTH`., defaults to ``[]``.
:param list, optional timeRange: Maintenance Time Window of a Day, defaults to ``[{{"hours": 2, "minutes": 0}}, {{"hours": 3, "minutes": 0}}]``. :param list, optional timeRange: Maintenance Time Window of a Day, defaults to ``[{{"hours": 2, "minutes": 0}}, {{"hours": 3, "minutes": 0}}]``.
:param str, optional cron: Cron Schedule. Required for ``strategy`` :attr:`~.MaintenanceStrategy.CRON`., defaults to ``"30 3 * * *"``
:param int, optional durationMinutes: Duration (Minutes). Required for ``strategy`` :attr:`~.MaintenanceStrategy.CRON`., defaults to ``60``
:param str, optional timezone: Timezone, defaults to ``None`` (Server Timezone)
""" """

View file

@ -18,3 +18,6 @@ class MaintenanceStrategy(str, Enum):
RECURRING_DAY_OF_MONTH = "recurring-day-of-month" RECURRING_DAY_OF_MONTH = "recurring-day-of-month"
"""Recurring - Day of Month""" """Recurring - Day of Month"""
CRON = "cron"
"""Cron Expression"""

View file

@ -148,6 +148,12 @@ class NotificationType(str, Enum):
SPLUNK = "Splunk" SPLUNK = "Splunk"
"""Splunk""" """Splunk"""
OPSGENIE = "Opsgenie"
"""Opsgenie"""
TWILIO = "twilio"
"""twilio"""
notification_provider_options = { notification_provider_options = {
NotificationType.ALERTA: dict( NotificationType.ALERTA: dict(
@ -411,6 +417,17 @@ notification_provider_options = {
splunkAutoResolve=dict(type="str"), splunkAutoResolve=dict(type="str"),
splunkSeverity=dict(type="str"), splunkSeverity=dict(type="str"),
splunkRestURL=dict(type="str") splunkRestURL=dict(type="str")
),
NotificationType.OPSGENIE: dict(
opsgeniePriority=dict(type="int"),
opsgenieRegion=dict(type="str"),
opsgenieApiKey=dict(type="str")
),
NotificationType.TWILIO: dict(
twilioAccountSID=dict(type="str"),
twilioAuthToken=dict(type="str"),
twilioToNumber=dict(type="str"),
twilioFromNumber=dict(type="str")
) )
} }
@ -430,5 +447,9 @@ notification_provider_conditions = dict(
smseaglePriority=dict( smseaglePriority=dict(
min=0, min=0,
max=9 max=9
),
opsgeniePriority=dict(
min=1,
max=5
) )
) )