tubearchivist/tubearchivist
1
0
Fork 0
mirror of https://github.com/tubearchivist/tubearchivist.git synced 2024-12-22 18:00:13 +00:00
Code Issues Projects Releases Wiki Activity

add type hints to ReleaseVersion

Browse Source
This commit is contained in:
Simon 2023-12-22 10:41:10 +07:00
parent b84bf78974
commit 94295cdbd4
No known key found for this signature in database
GPG Key ID: 2C15AA5E89985DD4
1 changed files with 17 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
tubearchivist/home/src/ta/config.py
View File

@ -269,14 +269,13 @@ class ReleaseVersion:
REMOTE_URL = "https://www.tubearchivist.com/api/release/latest/" REMOTE_URL = "https://www.tubearchivist.com/api/release/latest/"
NEW_KEY = "versioncheck:new" NEW_KEY = "versioncheck:new"
def __init__(self): def __init__(self) -> None:
self.local_version = settings.TA_VERSION self.local_version: str = settings.TA_VERSION
self.is_unstable = settings.TA_VERSION.endswith("-unstable") self.is_unstable: bool = settings.TA_VERSION.endswith("-unstable")
self.remote_version = False self.remote_version: str = ""
self.is_breaking = False self.is_breaking: bool = False
self.response = False
def check(self): def check(self) -> None:
"""check version""" """check version"""
print(f"[{self.local_version}]: look for updates") print(f"[{self.local_version}]: look for updates")
self.get_remote_version() self.get_remote_version()
@ -290,18 +289,18 @@ class ReleaseVersion:
RedisArchivist().set_message(self.NEW_KEY, message) RedisArchivist().set_message(self.NEW_KEY, message)
print(f"[{self.local_version}]: found new version {new_version}") print(f"[{self.local_version}]: found new version {new_version}")
def get_local_version(self): def get_local_version(self) -> str:
"""read version from local""" """read version from local"""
return self.local_version return self.local_version
def get_remote_version(self): def get_remote_version(self) -> None:
"""read version from remote""" """read version from remote"""
sleep(randint(0, 60)) sleep(randint(0, 60))
self.response = requests.get(self.REMOTE_URL, timeout=20).json() response = requests.get(self.REMOTE_URL, timeout=20).json()
self.remote_version = self.response["release_version"] self.remote_version = response["release_version"]
self.is_breaking = self.response["breaking_changes"] self.is_breaking = response["breaking_changes"]
def _has_update(self): def _has_update(self) -> str | bool:
"""check if there is an update""" """check if there is an update"""
remote_parsed = self._parse_version(self.remote_version) remote_parsed = self._parse_version(self.remote_version)
local_parsed = self._parse_version(self.local_version) local_parsed = self._parse_version(self.local_version)
@ -314,12 +313,12 @@ class ReleaseVersion:
return False return False
@staticmethod @staticmethod
def _parse_version(version): def _parse_version(version) -> tuple[int, ...]:
"""return version parts""" """return version parts"""
clean = version.rstrip("-unstable").lstrip("v") clean = version.rstrip("-unstable").lstrip("v")
return tuple((int(i) for i in clean.split("."))) return tuple((int(i) for i in clean.split(".")))
def is_updated(self): def is_updated(self) -> str | bool:
"""check if update happened in the mean time""" """check if update happened in the mean time"""
message = self.get_update() message = self.get_update()
if not message: if not message:
@ -334,15 +333,15 @@ class ReleaseVersion:
return False return False
def get_update(self): def get_update(self) -> dict:
"""return new version dict if available""" """return new version dict if available"""
message = RedisArchivist().get_message(self.NEW_KEY) message = RedisArchivist().get_message(self.NEW_KEY)
if not message.get("status"): if not message.get("status"):
return False return {}
return message return message
def clear_fail(self): def clear_fail(self) -> None:
"""clear key, catch previous error in v0.4.5""" """clear key, catch previous error in v0.4.5"""
message = self.get_update() message = self.get_update()
if not message: if not message: