-
-
Notifications
You must be signed in to change notification settings - Fork 13
/
version.py
85 lines (76 loc) · 2.8 KB
/
version.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
"""System Bridge Connector: Version"""
from __future__ import annotations
from aiohttp import ClientSession
from pkg_resources import parse_version
from systembridgeconnector.base import Base
from systembridgeconnector.exceptions import ConnectionErrorException
from systembridgeconnector.http_client import HTTPClient
from systembridgeconnector.models.system import System
SUPPORTED_VERSION = "3.1.2"
class Version(Base):
"""Version"""
def __init__(
self,
api_host: str,
api_port: int,
api_key: str,
session: ClientSession | None = None,
) -> None:
"""Initialize the client."""
super().__init__()
self._http_client = HTTPClient(
api_host,
api_port,
api_key,
session,
)
async def check_supported(self) -> bool:
"""Check if the system is running a supported version."""
if (
await self.check_version_2() is None
and (version := await self.check_version_3()) is not None
):
return parse_version(version) >= parse_version(SUPPORTED_VERSION)
return False
async def check_version_2(self) -> str | None:
"""Check if the system is running v2.x.x version."""
try:
information = await self._http_client.get("/information")
if (
information
and information.get("version")
and (
information.get("version").startswith("2")
or information.get("version").startswith("v2")
)
):
return information["version"]
except ConnectionErrorException as exception:
error: dict = exception.args[0]
if (
error is not None # pylint: disable=invalid-sequence-index
and error["status"] == 404 # pylint: disable=invalid-sequence-index
):
return None
raise exception
return None
async def check_version_3(self) -> str | None:
"""Check if the system is running v3.x.x version."""
try:
response = await self._http_client.get("/api/data/system")
system = System(**response)
if (
system
and system.version is not None
and parse_version(system.version) >= parse_version("3.0.0")
):
return system.version
except ConnectionErrorException as exception:
error: dict = exception.args[0]
if (
error is not None # pylint: disable=invalid-sequence-index
and error["status"] == 404 # pylint: disable=invalid-sequence-index
):
return None
raise exception
return None