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

feat: add support for uptime kuma 1.19.2

Browse source
This commit is contained in:
lucasheld 2022-12-29 00:22:53 +01:00
parent 1e4be04ad7
commit d01ff6d80e
17 changed files with 1128 additions and 79 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
docs/api.rst
View file

@ -31,6 +31,9 @@ Enums
.. autoclass:: DockerType
:members:
.. autoclass:: MaintenanceStrategy
:members:
Exceptions
----------

10
run_tests.sh
View file

@ -1,6 +1,14 @@
#!/bin/sh
for version in 1.18.5 1.17.1
version="$1"
if [ $version ]
then
versions=("$version")
else
versions=(1.19.2 1.18.5 1.17.1)
fi
for version in ${versions[*]}
do
echo "Starting uptime kuma $version..."
docker run -d -it --rm -p 3001:3001 --name uptimekuma "louislam/uptime-kuma:$version" > /dev/null

1
scripts/.gitignore vendored
View file

@ -1 +1,2 @@
uptime-kuma
uptime-kuma-old

22
scripts/build_inputs.py
View file

@ -4,6 +4,9 @@ import re
from utils import deduplicate_list, parse_vue_template
ROOT = "uptime-kuma"
def parse_model_keys(content, object_name):
match = re.findall(object_name + r"\.[0-9a-zA-Z_$]+", content)
keys = []
@ -15,13 +18,13 @@ def parse_model_keys(content, object_name):
def parse_proxy_keys():
content = parse_vue_template("uptime-kuma/src/components/ProxyDialog.vue")
content = parse_vue_template(f"{ROOT}/src/components/ProxyDialog.vue")
keys = parse_model_keys(content, "proxy")
return keys
def parse_notification_keys():
content = parse_vue_template("uptime-kuma/src/components/NotificationDialog.vue")
content = parse_vue_template(f"{ROOT}/src/components/NotificationDialog.vue")
keys = parse_model_keys(content, "notification")
return keys
@ -37,7 +40,7 @@ def parse_settings_keys():
def parse_monitor_keys():
content = parse_vue_template("uptime-kuma/src/pages/EditMonitor.vue")
content = parse_vue_template(f"{ROOT}/src/pages/EditMonitor.vue")
keys = parse_model_keys(content, "monitor")
return keys
@ -45,11 +48,11 @@ def parse_monitor_keys():
def parse_status_page_keys():
all_keys = ["id"]
content = parse_vue_template("uptime-kuma/src/pages/StatusPage.vue")
content = parse_vue_template(f"{ROOT}/src/pages/StatusPage.vue")
keys = parse_model_keys(content, "config")
all_keys.extend(keys)
content = parse_vue_template("uptime-kuma/src/pages/ManageStatusPage.vue")
content = parse_vue_template(f"{ROOT}/src/pages/ManageStatusPage.vue")
keys = parse_model_keys(content, "statusPage")
all_keys.extend(keys)
@ -57,6 +60,12 @@ def parse_status_page_keys():
return all_keys
def parse_maintenance_keys():
content = parse_vue_template(f"{ROOT}/src/pages/EditMaintenance.vue")
keys = parse_model_keys(content, "maintenance")
return keys
def main():
proxy_keys = parse_proxy_keys()
print("proxy:", proxy_keys)
@ -73,6 +82,9 @@ def main():
status_page_keys = parse_status_page_keys()
print("status_page:", status_page_keys)
maintenance_keys = parse_maintenance_keys()
print("maintenance:", maintenance_keys)
if __name__ == "__main__":
main()

37
scripts/build_notifications.py
View file

@ -6,9 +6,9 @@ from bs4 import BeautifulSoup
from utils import deduplicate_list, write_to_file
def build_notification_providers():
def build_notification_providers(root):
providers = []
for path in glob.glob('uptime-kuma/server/notification-providers/*'):
for path in glob.glob(f'{root}/server/notification-providers/*'):
with open(path) as f:
content = f.read()
match = re.search(r'class [^ ]+ extends NotificationProvider {', content)
@ -27,9 +27,9 @@ def build_notification_providers():
return providers
def build_notification_provider_conditions():
def build_notification_provider_conditions(root):
conditions = {}
for path in glob.glob('uptime-kuma/src/components/notifications/*'):
for path in glob.glob(f'{root}/src/components/notifications/*'):
if path.endswith("index.js"):
continue
with open(path) as f:
@ -54,11 +54,26 @@ def build_notification_provider_conditions():
return conditions
notification_providers = build_notification_providers()
notification_provider_conditions = build_notification_provider_conditions()
def diff(old, new):
for i in new:
if i not in old:
print("+", i)
for i in old:
if i not in new:
print("-", i)
print("")
write_to_file(
"notification_providers.py.j2", "./../uptime_kuma_api/notification_providers.py",
notification_providers=notification_providers,
notification_provider_conditions=notification_provider_conditions
)
# write_to_file(
# "notification_providers.py.j2", "./../uptime_kuma_api/notification_providers.py",
# notification_providers=notification_providers,
# notification_provider_conditions=notification_provider_conditions
# )
notification_providers_old = build_notification_providers("uptime-kuma-old")
notification_providers_new = build_notification_providers("uptime-kuma")
diff(notification_providers_old, notification_providers_new)
notification_provider_conditions_old = build_notification_provider_conditions("uptime-kuma-old")
notification_provider_conditions_new = build_notification_provider_conditions("uptime-kuma")
diff(notification_provider_conditions_old, notification_provider_conditions_new)

284
tests/test_maintenance.py Normal file
View file

@ -0,0 +1,284 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException, MaintenanceStrategy
from uptime_kuma_test_case import UptimeKumaTestCase
class TestMaintenance(UptimeKumaTestCase):
def setUp(self):
super(TestMaintenance, self).setUp()
if parse_version(self.api.version) < parse_version("1.19"):
super(TestMaintenance, self).tearDown()
self.skipTest("Unsupported in this Uptime Kuma version")
def test_maintenance(self):
expected_maintenance = {
"title": "maintenance 1",
"description": "test",
"strategy": "single",
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:36:00",
"2022-12-29 22:36:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": []
}
# add maintenance
r = self.api.add_maintenance(**expected_maintenance)
self.assertEqual(r["msg"], "Added Successfully.")
maintenance_id = r["maintenanceID"]
# get maintenance
maintenance = self.api.get_maintenance(maintenance_id)
self.compare(maintenance, expected_maintenance)
# get maintenances
maintenances = self.api.get_maintenances()
maintenance = self.find_by_id(maintenances, maintenance_id)
self.assertIsNotNone(maintenance)
self.compare(maintenance, expected_maintenance)
# edit maintenance
expected_maintenance["strategy"] = MaintenanceStrategy.RECURRING_INTERVAL
expected_maintenance["title"] = "maintenance 1 new"
r = self.api.edit_maintenance(maintenance_id, **expected_maintenance)
self.assertEqual(r["msg"], "Saved.")
maintenance = self.api.get_maintenance(maintenance_id)
self.compare(maintenance, expected_maintenance)
# pause maintenance
r = self.api.pause_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Paused Successfully.")
# resume maintenance
r = self.api.resume_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Resume Successfully")
# add monitor maintenance
monitor_name = "monitor 1"
monitor_id = self.add_monitor(monitor_name)
monitors = [
{
"id": monitor_id,
"name": monitor_name
},
]
r = self.api.add_monitor_maintenance(maintenance_id, monitors)
self.assertEqual(r["msg"], "Added Successfully.")
# get monitor maintenance
monitors = self.api.get_monitor_maintenance(maintenance_id)
monitor = self.find_by_id(monitors, monitor_id)
self.assertIsNotNone(monitor)
# add status page maintenance
status_page_title = "status page 1"
status_page_id = self.add_status_page(status_page_title)
status_pages = [
{
"id": status_page_id,
"name": status_page_title
}
]
r = self.api.add_status_page_maintenance(maintenance_id, status_pages)
self.assertEqual(r["msg"], "Added Successfully.")
# get status page maintenance
status_pages = self.api.get_status_page_maintenance(maintenance_id)
status_page = self.find_by_id(status_pages, status_page_id)
self.assertIsNotNone(status_page)
# delete maintenance
r = self.api.delete_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_maintenance(maintenance_id)
def test_maintenance_strategy_manual(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": "manual",
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 00:00:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_single(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": "single",
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:36:00",
"2022-12-29 22:36:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_recurring_interval(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": "recurring-interval",
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:37:00",
"2022-12-31 22:37:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_recurring_weekday(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": "recurring-weekday",
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:38:00",
"2022-12-31 22:38:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [
1,
3,
5,
0
],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_recurring_day_of_month(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": "recurring-day-of-month",
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:39:00",
"2022-12-31 22:39:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": [
1,
10,
20,
30,
"lastDay4",
"lastDay2"
]
}
self.do_test_maintenance_strategy(expected_maintenance)
def do_test_maintenance_strategy(self, expected_maintenance):
# add maintenance
r = self.api.add_maintenance(**expected_maintenance)
self.assertEqual(r["msg"], "Added Successfully.")
maintenance_id = r["maintenanceID"]
# get maintenance
maintenance = self.api.get_maintenance(maintenance_id)
self.compare(maintenance, expected_maintenance)
# get maintenances
maintenances = self.api.get_maintenances()
maintenance = self.find_by_id(maintenances, maintenance_id)
self.assertIsNotNone(maintenance)
self.compare(maintenance, expected_maintenance)
# edit maintenance
r = self.api.edit_maintenance(maintenance_id, title="name 2")
self.assertEqual(r["msg"], "Saved.")
maintenance = self.api.get_maintenance(maintenance_id)
expected_maintenance["title"] = "name 2"
self.compare(maintenance, expected_maintenance)
# delete maintenance
r = self.api.delete_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_maintenance(maintenance_id)
if __name__ == '__main__':
unittest.main()

38
tests/test_monitor.py
View file

@ -158,6 +158,20 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_grpc_keyword(self):
if parse_version(self.api.version) < parse_version("1.19"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.GRPC_KEYWORD,
"name": "monitor 1",
"grpcUrl": "127.0.0.1",
"keyword": "healthy",
"grpcServiceName": "health",
"grpcMethod": "check",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_dns(self):
expected_monitor = {
"type": MonitorType.DNS,
@ -169,6 +183,19 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_docker(self):
if parse_version(self.api.version) < parse_version("1.18"):
self.skipTest("Unsupported in this Uptime Kuma version")
docker_host_id = self.add_docker_host()
expected_monitor = {
"type": MonitorType.DOCKER,
"name": "monitor 1",
"docker_container": "test",
"docker_host": docker_host_id
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_push(self):
expected_monitor = {
"type": MonitorType.PUSH,
@ -223,16 +250,15 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_docker(self):
if parse_version(self.api.version) < parse_version("1.18"):
def test_monitor_type_mysql(self):
if parse_version(self.api.version) < parse_version("1.19"):
self.skipTest("Unsupported in this Uptime Kuma version")
docker_host_id = self.add_docker_host()
expected_monitor = {
"type": MonitorType.DOCKER,
"type": MonitorType.MYSQL,
"name": "monitor 1",
"docker_container": "test",
"docker_host": docker_host_id
"databaseConnectionString": "mysql://username:password@host:port/database",
"databaseQuery": "select getdate()"
}
self.do_test_monitor_type(expected_monitor)

6
tests/test_settings.py
View file

@ -24,6 +24,12 @@ class TestSettings(UptimeKumaTestCase):
"trustProxy": False
})
if parse_version(self.api.version) >= parse_version("1.19"):
expected_settings.update({
"serverTimezone": "Europe/Berlin",
"dnsCache": True,
})
# set settings
r = self.api.set_settings(self.password, **expected_settings)
self.assertEqual(r["msg"], "Saved")

12
tests/uptime_kuma_test_case.py
View file

@ -2,7 +2,7 @@ import unittest
import warnings
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaApi, MonitorType, DockerType
from uptime_kuma_api import UptimeKumaApi, MonitorType, DockerType, UptimeKumaException
token = None
@ -96,10 +96,10 @@ class UptimeKumaTestCase(unittest.TestCase):
if obj[key] == value:
return obj
def add_monitor(self):
def add_monitor(self, name="monitor 1"):
r = self.api.add_monitor(
type=MonitorType.HTTP,
name="monitor 1",
name=name,
url="http://127.0.0.1"
)
monitor_id = r["monitorID"]
@ -139,3 +139,9 @@ class UptimeKumaTestCase(unittest.TestCase):
)
docker_host_id = r["id"]
return docker_host_id
def add_status_page(self, title="status page 1"):
slug = "statuspage1"
self.api.add_status_page(slug, title)
r = self.api.get_status_page(slug)
return r["id"]

1
uptime_kuma_api/__init__.py
View file

@ -5,6 +5,7 @@ from .notification_providers import NotificationType, notification_provider_opti
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 .event import Event
from .api import UptimeKumaApi

718
uptime_kuma_api/api.py
View file

@ -1,10 +1,11 @@
import datetime
import json
import random
import string
import time
from contextlib import contextmanager
from copy import deepcopy
from typing import Any
from typing import Any, Union
import requests
import socketio
@ -14,12 +15,13 @@ from . import AuthMethod
from . import DockerType
from . import Event
from . import IncidentStyle
from . import MaintenanceStrategy
from . import MonitorType
from . import NotificationType, notification_provider_options, notification_provider_conditions
from . import ProxyProtocol
from . import UptimeKumaException
from .docstrings import append_docstring, monitor_docstring, notification_docstring, proxy_docstring, \
docker_host_docstring
docker_host_docstring, maintenance_docstring
def int_to_bool(data, keys) -> None:
@ -63,11 +65,11 @@ def _convert_monitor_input(kwargs) -> None:
def _build_notification_data(
name: str,
type: NotificationType,
isDefault: bool = False,
applyExisting: bool = False,
**kwargs
name: str,
type: NotificationType,
isDefault: bool = False,
applyExisting: bool = False,
**kwargs
) -> dict:
allowed_kwargs = []
for keys in notification_provider_options.values():
@ -88,15 +90,15 @@ def _build_notification_data(
def _build_proxy_data(
protocol: ProxyProtocol,
host: str,
port: str,
auth: bool = False,
username: str = None,
password: str = None,
active: bool = True,
default: bool = False,
applyExisting: bool = False,
protocol: ProxyProtocol,
host: str,
port: str,
auth: bool = False,
username: str = None,
password: str = None,
active: bool = True,
default: bool = False,
applyExisting: bool = False,
) -> dict:
data = {
"protocol": protocol,
@ -113,22 +115,22 @@ def _build_proxy_data(
def _build_status_page_data(
slug: str,
slug: str,
# config
id: int,
title: str,
description: str = None,
theme: str = "light",
published: bool = True,
showTags: bool = False,
domainNameList: list = None,
customCSS: str = "",
footerText: str = None,
showPoweredBy: bool = True,
# config
id: int,
title: str,
description: str = None,
theme: str = "light",
published: bool = True,
showTags: bool = False,
domainNameList: list = None,
customCSS: str = "",
footerText: str = None,
showPoweredBy: bool = True,
icon: str = "/icon.svg",
publicGroupList: list = None
icon: str = "/icon.svg",
publicGroupList: list = None
) -> (str, dict, str, list):
if theme not in ["light", "dark"]:
raise ValueError
@ -162,9 +164,9 @@ def _convert_docker_host_input(kwargs) -> None:
def _build_docker_host_data(
name: str,
dockerType: DockerType,
dockerDaemon: str = None
name: str,
dockerType: DockerType,
dockerDaemon: str = None
) -> dict:
data = {
"name": name,
@ -174,6 +176,49 @@ def _build_docker_host_data(
return data
def _build_maintenance_data(
title: str,
strategy: MaintenanceStrategy,
active: bool = True,
description: str = "",
dateRange: list[str] = None,
intervalDay: int = 1,
weekdays: list[int] = None,
daysOfMonth: list[Union[int, str]] = None,
timeRange: list[dict] = None
) -> dict:
if not dateRange:
dateRange = [
datetime.date.today().strftime("%Y-%m-%d 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 _check_missing_arguments(required_params, kwargs) -> None:
missing_arguments = []
for required_param in required_params:
@ -214,13 +259,15 @@ def _check_arguments_monitor(kwargs) -> None:
MonitorType.PORT: ["hostname", "port"],
MonitorType.PING: ["hostname"],
MonitorType.KEYWORD: ["url", "keyword", "maxredirects"],
MonitorType.GRPC_KEYWORD: ["grpcUrl", "keyword", "grpcServiceName", "grpcMethod"],
MonitorType.DNS: ["hostname", "dns_resolve_server", "port"],
MonitorType.DOCKER: ["docker_container", "docker_host"],
MonitorType.PUSH: [],
MonitorType.STEAM: ["hostname", "port"],
MonitorType.MQTT: ["hostname", "port", "mqttTopic"],
MonitorType.SQLSERVER: [],
MonitorType.POSTGRES: [],
MonitorType.DOCKER: ["docker_container", "docker_host"],
MonitorType.MYSQL: [],
MonitorType.RADIUS: ["radiusUsername", "radiusPassword", "radiusSecret", "radiusCalledStationId", "radiusCallingStationId"]
}
type_ = kwargs["type"]
@ -274,6 +321,23 @@ def _check_arguments_proxy(kwargs) -> None:
_check_argument_conditions(conditions, kwargs)
def _check_arguments_maintenance(kwargs) -> None:
required_args = ["title", "strategy"]
_check_missing_arguments(required_args, kwargs)
strategy = kwargs["strategy"]
if strategy in [MaintenanceStrategy.RECURRING_INTERVAL, MaintenanceStrategy.RECURRING_WEEKDAY, MaintenanceStrategy.RECURRING_DAY_OF_MONTH]:
required_args = ["dateRange"]
_check_missing_arguments(required_args, kwargs)
conditions = dict(
intervalDay=dict(
min=1,
max=3650,
)
)
_check_argument_conditions(conditions, kwargs)
class UptimeKumaApi(object):
"""This class is used to communicate with Uptime Kuma.
@ -327,7 +391,8 @@ class UptimeKumaApi(object):
Event.INFO: None,
Event.CERT_INFO: None,
Event.DOCKER_HOST_LIST: None,
Event.AUTO_LOGIN: None
Event.AUTO_LOGIN: None,
Event.MAINTENANCE_LIST: None
}
self.sio.on(Event.CONNECT, self._event_connect)
@ -345,6 +410,8 @@ class UptimeKumaApi(object):
self.sio.on(Event.CERT_INFO, self._event_cert_info)
self.sio.on(Event.DOCKER_HOST_LIST, self._event_docker_host_list)
self.sio.on(Event.AUTO_LOGIN, self._event_auto_login)
self.sio.on(Event.INIT_SERVER_TIMEZONE, self._event_init_server_timezone)
self.sio.on(Event.MAINTENANCE_LIST, self._event_maintenance_list)
self.connect()
@ -461,6 +528,12 @@ class UptimeKumaApi(object):
def _event_auto_login(self) -> None:
self._event_data[Event.AUTO_LOGIN] = True
def _event_init_server_timezone(self) -> None:
pass
def _event_maintenance_list(self, data) -> None:
self._event_data[Event.MAINTENANCE_LIST] = data
# connection
def connect(self) -> None:
@ -522,11 +595,20 @@ class UptimeKumaApi(object):
# KEYWORD
keyword: str = None,
# GRPC_KEYWORD
grpcUrl: str = None,
grpcEnableTls: bool = False,
grpcServiceName: str = None,
grpcMethod: str = None,
grpcProtobuf: str = None,
grpcBody: str = None,
grpcMetadata: str = None,
# DNS, PING, STEAM, MQTT
hostname: str = None,
# DNS, STEAM, MQTT
port: int = 53,
# DNS, STEAM, MQTT, RADIUS
port: int = None,
# DNS
dns_resolve_server: str = "1.1.1.1",
@ -568,7 +650,7 @@ class UptimeKumaApi(object):
"resendInterval": resendInterval
})
if type == MonitorType.KEYWORD:
if type in [MonitorType.KEYWORD, MonitorType.GRPC_KEYWORD]:
data.update({
"keyword": keyword,
})
@ -599,12 +681,29 @@ class UptimeKumaApi(object):
"authWorkstation": authWorkstation,
})
# GRPC_KEYWORD
if type == MonitorType.GRPC_KEYWORD:
data.update({
"grpcUrl": grpcUrl,
"grpcEnableTls": grpcEnableTls,
"grpcServiceName": grpcServiceName,
"grpcMethod": grpcMethod,
"grpcProtobuf": grpcProtobuf,
"grpcBody": grpcBody,
"grpcMetadata": grpcMetadata,
})
# PORT, PING, DNS, STEAM, MQTT
data.update({
"hostname": hostname,
})
# PORT, DNS, STEAM, MQTT
# PORT, DNS, STEAM, MQTT, RADIUS
if not port:
if type == MonitorType.DNS:
port = 53
elif type == MonitorType.RADIUS:
port = 1812
data.update({
"port": port,
})
@ -627,7 +726,7 @@ class UptimeKumaApi(object):
data.update({
"databaseConnectionString": databaseConnectionString
})
if type in [MonitorType.SQLSERVER, MonitorType.POSTGRES]:
if type in [MonitorType.SQLSERVER, MonitorType.POSTGRES, MonitorType.MYSQL]:
data.update({
"databaseQuery": databaseQuery,
})
@ -695,6 +794,7 @@ class UptimeKumaApi(object):
'proxyId': None,
'notificationIDList': [],
'tags': [],
'maintenance': False,
'mqttUsername': None,
'mqttPassword': None,
'mqttTopic': None,
@ -765,6 +865,7 @@ class UptimeKumaApi(object):
'proxyId': None,
'notificationIDList': [],
'tags': [],
'maintenance': False,
'mqttUsername': None,
'mqttPassword': None,
'mqttTopic': None,
@ -1398,12 +1499,14 @@ class UptimeKumaApi(object):
'style': 'danger',
'title': 'title 1'
},
'maintenanceList': [],
'publicGroupList': [
{
'id': 1,
'monitorList': [
{
'id': 1,
'maintenance': False,
'name': 'monitor 1',
'sendUrl': 0
}
@ -1425,11 +1528,17 @@ class UptimeKumaApi(object):
config = r1["config"]
config.update(r2["config"])
return {
data = {
**config,
"incident": r2["incident"],
"publicGroupList": r2["publicGroupList"]
"publicGroupList": r2["publicGroupList"],
}
if parse_version(self.version) >= parse_version("1.19"):
data.update({
"maintenanceList": r2["maintenanceList"]
})
return data
def add_status_page(self, slug: str, title: str) -> dict:
"""
@ -1540,6 +1649,8 @@ class UptimeKumaApi(object):
"""
status_page = self.get_status_page(slug)
status_page.pop("incident")
if parse_version(self.version) >= parse_version("1.19"):
status_page.pop("maintenanceList")
status_page.update(kwargs)
data = _build_status_page_data(**status_page)
r = self._call('saveStatusPage', data)
@ -1804,9 +1915,11 @@ class UptimeKumaApi(object):
>>> api.info()
{
'version': '1.18.5',
'latestVersion': '1.18.5',
'primaryBaseURL': None
'version': '1.19.2',
'latestVersion': '1.19.2',
'primaryBaseURL': None,
'serverTimezone': 'Europe/Berlin',
'serverTimezoneOffset': '+01:00'
}
"""
r = self._get_event_data(Event.INFO)
@ -1977,19 +2090,21 @@ class UptimeKumaApi(object):
>>> api.get_settings()
{
'checkUpdate': False,
'checkBeta': False,
'keepDataPeriodDays': 180,
'checkUpdate': False,
'disableAuth': False,
'dnsCache': True,
'entryPage': 'dashboard',
'searchEngineIndex': False,
'keepDataPeriodDays': 180,
'primaryBaseURL': '',
'searchEngineIndex': False,
'serverTimezone': 'Europe/Berlin',
'steamAPIKey': '',
'tlsExpiryNotifyDays': [
7,
14,
21
],
'disableAuth': False,
'trustProxy': False
}
"""
@ -2008,10 +2123,12 @@ class UptimeKumaApi(object):
keepDataPeriodDays: int = 180,
# general
serverTimezone: str = "",
entryPage: str = "dashboard",
searchEngineIndex: bool = False,
primaryBaseURL: str = "",
steamAPIKey: str = "",
dnsCache: bool = False,
# notifications
tlsExpiryNotifyDays: list = None,
@ -2029,10 +2146,12 @@ class UptimeKumaApi(object):
:param bool, optional checkUpdate: Show update if available, defaults to True
:param bool, optional checkBeta: Also check beta release, defaults to False
:param int, optional keepDataPeriodDays: Keep monitor history data for X days., defaults to 180
:param str, optional serverTimezone: Server Timezone, defaults to ""
:param str, optional entryPage: Entry Page, defaults to "dashboard"
:param bool, optional searchEngineIndex: Search Engine Visibility, defaults to False
:param str, optional primaryBaseURL: Primary Base URL, defaults to ""
:param str, optional steamAPIKey: Steam API Key. For monitoring a Steam Game Server you need a Steam Web-API key., defaults to ""
:param bool, optional dnsCache: True to enable DNS Cache. It may be not working in some IPv6 environments, disable it if you encounter any issues., defaults to False
:param list, optional tlsExpiryNotifyDays: TLS Certificate Expiry. HTTPS Monitors trigger notification when TLS certificate expires in., defaults to None
:param bool, optional disableAuth: Disable Authentication, defaults to False
:param bool, optional trustProxy: Trust Proxy. Trust 'X-Forwarded-\*' headers. If you want to get the correct client IP and your Uptime Kuma is behind such as Nginx or Apache, you should enable this., defaults to False
@ -2046,11 +2165,17 @@ class UptimeKumaApi(object):
... checkUpdate=False,
... checkBeta=False,
... keepDataPeriodDays=180,
... serverTimezone="Europe/Berlin",
... entryPage="dashboard",
... searchEngineIndex=False,
... primaryBaseURL="",
... steamAPIKey="",
... tlsExpiryNotifyDays=[7, 14, 21],
... dnsCache=False,
... tlsExpiryNotifyDays=[
... 7,
... 14,
... 21
... ],
... disableAuth=False,
... trustProxy=False
... )
@ -2066,10 +2191,12 @@ class UptimeKumaApi(object):
"checkUpdate": checkUpdate,
"checkBeta": checkBeta,
"keepDataPeriodDays": keepDataPeriodDays,
"serverTimezone": serverTimezone,
"entryPage": entryPage,
"searchEngineIndex": searchEngineIndex,
"primaryBaseURL": primaryBaseURL,
"steamAPIKey": steamAPIKey,
"dnsCache": dnsCache,
"tlsExpiryNotifyDays": tlsExpiryNotifyDays,
"disableAuth": disableAuth
}
@ -2525,3 +2652,494 @@ class UptimeKumaApi(object):
"""
with self.wait_for_event(Event.DOCKER_HOST_LIST):
return self._call('deleteDockerHost', id_)
def get_maintenances(self) -> list:
"""
Get all maintenances.
:return: All maintenances.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.get_maintenances()
[
{
"id": 1,
"title": "title",
"description": "description",
"strategy": "single",
"intervalDay": 1,
"active": true,
"dateRange": [
"2022-12-27 15:39:00",
"2022-12-30 15:39:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0,
"seconds": 0
},
{
"hours": 3,
"minutes": 0,
"seconds": 0
}
],
"weekdays": [],
"daysOfMonth": [],
"timeslotList": [
{
"id": 1,
"startDate": "2022-12-27 14:39: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"
}
]
"""
return list(self._get_event_data(Event.MAINTENANCE_LIST).values())
def get_maintenance(self, id_: int) -> dict:
"""
Get a maintenance.
:param id_: Id of the maintenance to get.
:return: The maintenance.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.get_maintenance(1)
{
"id": 1,
"title": "title",
"description": "description",
"strategy": "single",
"intervalDay": 1,
"active": true,
"dateRange": [
"2022-12-27 15:39:00",
"2022-12-30 15:39:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0,
"seconds": 0
},
{
"hours": 3,
"minutes": 0,
"seconds": 0
}
],
"weekdays": [],
"daysOfMonth": [],
"timeslotList": [
{
"id": 1,
"startDate": "2022-12-27 14:39: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"
}
"""
return self._call('getMaintenance', id_)["maintenance"]
@append_docstring(maintenance_docstring("add"))
def add_maintenance(self, **kwargs) -> dict:
"""
Adds a maintenance.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example (strategy: :attr:`~.MaintenanceStrategy.MANUAL`)::
>>> api.add_maintenance(
... title="test",
... description="test",
... strategy=MaintenanceStrategy.MANUAL,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 00:00:00"
... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0
... },
... {
... "hours": 3,
... "minutes": 0
... }
... ],
... weekdays=[],
... daysOfMonth=[]
... )
{
"msg": "Added Successfully.",
"maintenanceID": 1
}
Example (strategy: :attr:`~.MaintenanceStrategy.SINGLE`)::
>>> api.add_maintenance(
... title="test",
... description="test",
... strategy=MaintenanceStrategy.SINGLE,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 22:36:00",
... "2022-12-29 22:36:00"
... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0
... },
... {
... "hours": 3,
... "minutes": 0
... }
... ],
... weekdays=[],
... daysOfMonth=[]
... )
{
"msg": "Added Successfully.",
"maintenanceID": 1
}
Example (strategy: :attr:`~.MaintenanceStrategy.RECURRING_INTERVAL`)::
>>> api.add_maintenance(
... title="test",
... description="test",
... strategy=MaintenanceStrategy.RECURRING_INTERVAL,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 22:37:00",
... "2022-12-31 22:37:00"
... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0
... },
... {
... "hours": 3,
... "minutes": 0
... }
... ],
... weekdays=[],
... daysOfMonth=[]
... )
{
"msg": "Added Successfully.",
"maintenanceID": 1
}
Example (strategy: :attr:`~.MaintenanceStrategy.RECURRING_WEEKDAY`)::
>>> api.add_maintenance(
... title="test",
... description="test",
... strategy=MaintenanceStrategy.RECURRING_WEEKDAY,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 22:38:00",
... "2022-12-31 22:38:00"
... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0
... },
... {
... "hours": 3,
... "minutes": 0
... }
... ],
... weekdays=[
... 1,
... 3,
... 5,
... 0
... ],
... daysOfMonth=[]
... )
{
"msg": "Added Successfully.",
"maintenanceID": 1
}
Example (strategy: :attr:`~.MaintenanceStrategy.RECURRING_DAY_OF_MONTH`)::
>>> api.add_maintenance(
... title="test",
... description="test",
... strategy=MaintenanceStrategy.RECURRING_DAY_OF_MONTH,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 22:39:00",
... "2022-12-31 22:39:00"
... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0
... },
... {
... "hours": 3,
... "minutes": 0
... }
... ],
... weekdays=[],
... daysOfMonth=[
... 1,
... 10,
... 20,
... 30,
... "lastDay2"
... ]
... )
{
"msg": "Added Successfully.",
"maintenanceID": 1
}
"""
data = _build_maintenance_data(**kwargs)
_check_arguments_maintenance(data)
return self._call('addMaintenance', data)
@append_docstring(maintenance_docstring("edit"))
def edit_maintenance(self, id_: int, **kwargs) -> dict:
"""
Edits a maintenance.
:param id_: Id of the maintenance to edit.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.edit_maintenance(
... 1,
... title="test",
... description="test",
... strategy=MaintenanceStrategy.RECURRING_INTERVAL,
... active=True,
... intervalDay=1,
... dateRange=[
... "2022-12-27 22:37:00",
... "2022-12-31 22:37:00"
... ],
... timeRange=[
... {
... "hours": 2,
... "minutes": 0
... },
... {
... "hours": 3,
... "minutes": 0
... }
... ],
... weekdays=[],
... daysOfMonth=[]
... )
{
"msg": "Saved.",
"maintenanceID": 1
}
"""
maintenance = self.get_maintenance(id_)
maintenance.update(kwargs)
_check_arguments_maintenance(maintenance)
return self._call('editMaintenance', maintenance)
def delete_maintenance(self, id_: int) -> dict:
"""
Deletes a maintenance.
:param id_: Id of the maintenance to delete.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.delete_maintenance(1)
{
"msg": "Deleted Successfully."
}
"""
return self._call('deleteMaintenance', id_)
def pause_maintenance(self, id_: int) -> dict:
"""
Pauses a maintenance.
:param id_: Id of the maintenance to pause.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.pause_maintenance(1)
{
"msg": "Paused Successfully."
}
"""
return self._call('pauseMaintenance', id_)
def resume_maintenance(self, id_: int) -> dict:
"""
Resumes a maintenance.
:param id_: Id of the maintenance to resume.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.resume_maintenance(1)
{
"msg": "Resume Successfully"
}
"""
return self._call('resumeMaintenance', id_)
def get_monitor_maintenance(self, id_: int) -> list:
"""
Gets all monitors of a maintenance.
:param id_: Id of the maintenance to get the monitors from.
:return: All monitors of the maintenance.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.get_monitor_maintenance(1)
[
{
"id": 1,
"name": "monitor 1"
},
{
"id": 2,
"name": "monitor 2"
}
]
"""
return self._call('getMonitorMaintenance', id_)["monitors"]
def add_monitor_maintenance(
self,
id_: int,
monitors: list[dict],
) -> dict:
"""
Adds monitors to a maintenance.
:param id_: Id of the maintenance to add the monitors to.
:param monitors: The list of monitors to add to the maintenance.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> monitors = [
... {
... "id": 1,
... "name": "monitor 1"
... },
... {
... "id": 2,
... "name": "monitor 2"
... }
... ]
>>> api.add_monitor_maintenance(1, monitors)
{
"msg": "Added Successfully."
}
"""
return self._call('addMonitorMaintenance', (id_, monitors))
def get_status_page_maintenance(self, id_: int) -> list:
"""
Gets all status pages of a maintenance.
:param id_: Id of the maintenance to get the status pages from.
:return: All status pages of the maintenance.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.get_status_page_maintenance(1)
[
{
"id": 1,
"title": "test"
}
]
"""
return self._call('getMaintenanceStatusPage', id_)["statusPages"]
def add_status_page_maintenance(
self,
id_: int,
status_pages: list,
) -> dict:
"""
Adds status pages to a maintenance.
:param id_: Id of the maintenance to add the monitors to.
:param status_pages: The list of status pages to add to the maintenance.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> status_pages = [
... {
... "id": 1,
... "name": "status page 1"
... },
... {
... "id": 2,
... "name": "status page 2"
... }
... ]
>>> api.add_status_page_maintenance(1, status_pages)
{
"msg": "Added Successfully."
}
"""
return self._call('addMaintenanceStatusPage', (id_, status_pages))

23
uptime_kuma_api/docstrings.py
View file

@ -40,7 +40,7 @@ def monitor_docstring(mode) -> str:
:param str, optional authWorkstation: Workstation, defaults to None
:param str, optional keyword: Keyword. Search keyword in plain HTML or JSON response. The search is case-sensitive., defaults to None
:param str, optional hostname: Hostname, defaults to None
:param int, optional port: Port, defaults to 53
:param int, optional port: Port, ``type`` :attr:`~.MonitorType.DNS` defaults to ``53`` and ``type`` :attr:`~.MonitorType.RADIUS` defaults to ``1812``
:param str, optional dns_resolve_server: Resolver Server, defaults to "1.1.1.1"
:param str, optional dns_resolve_type: Resource Record Type, defaults to "A"
:param str, optional mqttUsername: MQTT Username, defaults to None
@ -171,6 +171,7 @@ def notification_docstring(mode) -> str:
:param str, optional telegramBotToken: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param str, optional telegramChatID: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param str, optional webhookContentType: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`
:param str, optional webhookAdditionalHeaders: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`
:param str, optional webhookURL: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`
:param str, optional weComBotKey: Notification option for ``type`` :attr:`~.NotificationType.WECOM`
:param str, optional alertNowWebhookURL: Notification option for ``type`` :attr:`~.NotificationType.ALERTNOW`
@ -205,7 +206,14 @@ def notification_docstring(mode) -> str:
:param str, optional ntfypassword: Notification option for ``type`` :attr:`~.NotificationType.NTFY`
:param str, optional ntfytopic: Notification option for ``type`` :attr:`~.NotificationType.NTFY`
:param int, optional ntfyPriority: Notification option for ``type`` :attr:`~.NotificationType.NTFY`
:param int, optional ntfyIcon: Notification option for ``type`` :attr:`~.NotificationType.NTFY`
:param str, optional ntfyserverurl: Notification option for ``type`` :attr:`~.NotificationType.NTFY`
:param bool, optional smseagleEncoding: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. True to send messages in unicode.
:param int, optional smseaglePriority: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Message priority (0-9, default = 0).
:param str, optional smseagleRecipientType: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Recipient type. Available values are "smseagle-to" (Phone number(s)), "smseagle-group" (Phonebook group name(s)) and "smseagle-contact" (Phonebook contact name(s)).
:param str, optional smseagleToken: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. API Access token.
:param str, optional smseagleRecipient: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Recipient(s) (multiple must be separated with comma).
:param str, optional smseagleUrl: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Your SMSEagle device URL.
"""
def proxy_docstring(mode) -> str:
@ -227,3 +235,16 @@ def docker_host_docstring(mode) -> str:
:param DockerType{", optional" if mode == "edit" else ""} dockerType: Connection Type
:param str, optional dockerDaemon: Docker Daemon, defaults to None
"""
def maintenance_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} title: Title
:param MaintenanceStrategy{", optional" if mode == "edit" else ""} strategy: Strategy
:param bool, optional active: True if maintenance is active, defaults to ``True``
:param str, optional description: Description, defaults to ``""``
:param list[str], optional dateRange: DateTime Range, defaults to ``["<current date>"]``
:param int, optional intervalDay: Interval (Run once every day), defaults to ``1``
:param list[int], 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[Union[int, str]], 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[dict], optional timeRange: Maintenance Time Window of a Day, defaults to ``[{{"hours": 2, "minutes": 0}}, {{"hours": 3, "minutes": 0}}]``.
"""

2
uptime_kuma_api/event.py
View file

@ -17,3 +17,5 @@ class Event(str, Enum):
CERT_INFO = "certInfo"
DOCKER_HOST_LIST = "dockerHostList"
AUTO_LOGIN = "autoLogin"
INIT_SERVER_TIMEZONE = "initServerTimezone"
MAINTENANCE_LIST = "maintenanceList"

20
uptime_kuma_api/maintenance_strategy.py Normal file
View file

@ -0,0 +1,20 @@
from enum import Enum
class MaintenanceStrategy(str, Enum):
"""Enumerate maintenance strategies."""
MANUAL = "manual"
"""Active/Inactive Manually"""
SINGLE = "single"
"""Single Maintenance Window"""
RECURRING_INTERVAL = "recurring-interval"
"""Recurring - Interval"""
RECURRING_WEEKDAY = "recurring-weekday"
"""Recurring - Day of Week"""
RECURRING_DAY_OF_MONTH = "recurring-day-of-month"
"""Recurring - Day of Month"""

6
uptime_kuma_api/monitor_type.py
View file

@ -16,6 +16,9 @@ class MonitorType(str, Enum):
KEYWORD = "keyword"
"""HTTP(s) - Keyword"""
GRPC_KEYWORD = "grpc-keyword"
"""gRPC(s) - Keyword"""
DNS = "dns"
"""DNS"""
@ -37,5 +40,8 @@ class MonitorType(str, Enum):
POSTGRES = "postgres"
"""PostgreSQL"""
MYSQL = "mysql"
"""MySQL/MariaDB"""
RADIUS = "radius"
"""Radius"""

17
uptime_kuma_api/notification_providers.py
View file

@ -133,6 +133,9 @@ class NotificationType(str, Enum):
NTFY = "ntfy"
"""ntfy"""
SMSEAGLE = "SMSEagle"
"""SMSEagle"""
notification_provider_options = {
NotificationType.ALERTA: dict(
@ -302,6 +305,7 @@ notification_provider_options = {
),
NotificationType.WEBHOOK: dict(
webhookContentType=dict(type="str"),
webhookAdditionalHeaders=dict(type="str"),
webhookURL=dict(type="str"),
),
NotificationType.WECOM: dict(
@ -362,8 +366,17 @@ notification_provider_options = {
ntfypassword=dict(type="str"),
ntfytopic=dict(type="str"),
ntfyPriority=dict(type="int"),
ntfyIcon=dict(type="str"),
ntfyserverurl=dict(type="str"),
),
NotificationType.SMSEAGLE: dict(
smseagleEncoding=dict(type="bool"),
smseaglePriority=dict(type="int"),
smseagleRecipientType=dict(type="str"),
smseagleToken=dict(type="str"),
smseagleRecipient=dict(type="str"),
smseagleUrl=dict(type="str")
)
}
notification_provider_conditions = dict(
@ -379,4 +392,8 @@ notification_provider_conditions = dict(
min=1,
max=5
),
smseaglePriority=dict(
min=0,
max=9
)
)

7
uptime_kuma_api/proxy_protocol.py
View file

@ -14,7 +14,10 @@ class ProxyProtocol(str, Enum):
"""SOCKS"""
SOCKS5 = "socks5"
"""SOCKS5"""
"""SOCKS v5"""
SOCKS5H = "socks5h"
"""SOCKS v5 (+DNS)"""
SOCKS4 = "socks4"
"""SOCKS4"""
"""SOCKS v4"""