51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
import logging
|
|
from typing import Any, Dict, Tuple, Union
|
|
|
|
from aiohttp import ClientSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Updater:
|
|
def __init__(self, client_session: Union[ClientSession, None] = None) -> None:
|
|
self.client_session: Union[ClientSession, None] = client_session
|
|
|
|
async def check_updates(
|
|
self, version_current: Tuple[int, int, int], api_url: str
|
|
) -> bool:
|
|
if not self.client_session:
|
|
self.client_session = ClientSession()
|
|
|
|
response = await self.client_session.get(api_url)
|
|
|
|
if response.status != 200:
|
|
return False
|
|
|
|
try:
|
|
version_latest = (await response.json())["tag_name"][1:].split(".")
|
|
except Exception as exc:
|
|
logger.error("Error parsing latest version: %s", exc)
|
|
return False
|
|
|
|
return any(
|
|
version_current[index] < int(subversion)
|
|
for index, subversion in enumerate(version_latest)
|
|
)
|
|
|
|
async def get_latest_release(self, api_url: str) -> Dict[str, Any]:
|
|
if not self.client_session:
|
|
self.client_session = ClientSession()
|
|
|
|
response = await self.client_session.get(api_url)
|
|
|
|
if response.status != 200:
|
|
raise RuntimeError(f"Could not fetch latest release: {response.status}")
|
|
|
|
try:
|
|
return await response.json()
|
|
except Exception as exc:
|
|
logger.error("Error parsing latest release: %s", exc)
|
|
raise RuntimeError(
|
|
f"Error parsing latest release: {response.status}"
|
|
) from exc
|