Bases: BaseService
A service class for application updates.
This class provides methods for checking, downloading, and installing application updates.
The constructor for the UpdateService class.
Parameters:
-
config
(UpdateServiceConfig | None, default:
None
)
–
The update configuration.
-
request
(RequestService | None, default:
None
)
–
The request service for making HTTP requests.
Source code in dev_tool/services/update/service.py
| def __init__(self, config: UpdateServiceConfig | None = None, request: RequestService | None = None) -> None:
"""
The constructor for the UpdateService class.
:param config: The update configuration.
:param request: The request service for making HTTP requests.
"""
super().__init__()
self.config = config or UpdateServiceConfig()
self.request = request or RequestService()
github_token = self.get_dev_tool_token()
default_headers = {
'Authorization': f'token {github_token}',
'Accept': 'application/vnd.github.v3+json'
}
self.client = APIClientService(self.request, self.config.base_url, default_headers)
|
A method that downloads a binary asset from GitHub.
Parameters:
-
asset
(dict[str, str])
–
The asset information containing download URL.
Returns:
-
bool
–
True if download was successful.
Raises:
Source code in dev_tool/services/update/service.py
| def download_binary(self, asset: dict[str, str]) -> bool:
"""
A method that downloads a binary asset from GitHub.
:param asset: The asset information containing download URL.
:return: True if download was successful.
:raises BinaryDownloadError: If the download fails.
"""
if not asset:
message = 'No valid asset provided to download the binary'
log.exception(message)
raise BinaryDownloadError(message)
path = STRATUS / 'update.exe'
url = asset.get('url')
headers = {'Accept': 'application/octet-stream'}
try:
if url is None:
return False
with self.client.stream('GET', url, headers=headers) as response:
response.raise_for_status()
with open(path, 'wb') as file:
for chunk in response.iter_bytes(chunk_size=8192):
file.write(chunk)
message = f'The file was downloaded successfully to {path}'
self.notification.normal_text(message)
log.debug(message)
except Exception:
message = f'Failed to download binary from {url}'
log.exception(message)
raise BinaryDownloadError(message) from None
else:
return True
|
A method that finds the appropriate asset for the current platform.
Parameters:
-
release
(dict[str, str])
–
The GitHub release information.
Returns:
-
dict[str, str]
–
The asset information for the current platform.
Raises:
-
AssetNotFoundError
–
If no suitable asset is found.
Source code in dev_tool/services/update/service.py
| def find_asset(self, release: dict[str, str]) -> dict[str, str]:
"""
A method that finds the appropriate asset for the current platform.
:param release: The GitHub release information.
:return: The asset information for the current platform.
:raises AssetNotFoundError: If no suitable asset is found.
"""
try:
assets = release.get('assets', [])
platform = PLATFORM.get(sys.platform)
for asset in assets:
if not isinstance(asset, dict):
continue
name = asset.get('name', '').lower()
if platform is None:
continue
if 'cli' in name and platform.lower() in name:
return asset
message = f'No suitable asset found for platform: {sys.platform}'
log.debug(message)
raise AssetNotFoundError(message)
except (AttributeError, TypeError):
message = 'Failed to find appropriate asset for platform'
log.exception(message)
raise AssetNotFoundError(message) from None
|
A method that retrieves the GitHub token for authentication.
Returns:
Raises:
-
TokenError
–
If the token cannot be retrieved.
Source code in dev_tool/services/update/service.py
| def get_dev_tool_token(self) -> str:
"""
A method that retrieves the GitHub token for authentication.
:return: The GitHub token.
:raises TokenError: If the token cannot be retrieved.
"""
try:
with open(DEV_TOOL_TOKEN, 'r', encoding='utf-8') as handle:
return handle.read().strip()
except FileNotFoundError:
message = f'GitHub token file not found: {DEV_TOOL_TOKEN}'
log.exception(message)
raise TokenError(message) from None
except Exception:
message = f'Failed to read GitHub token from {DEV_TOOL_TOKEN}'
log.exception(message)
raise TokenError(message) from None
|
A method that retrieves GitHub releases for the configured repository.
Returns:
-
list
–
A list of GitHub releases.
Raises:
-
GitHubReleaseError
–
If releases cannot be retrieved.
Source code in dev_tool/services/update/service.py
| def get_github_releases(self) -> list:
"""
A method that retrieves GitHub releases for the configured repository.
:return: A list of GitHub releases.
:raises GitHubReleaseError: If releases cannot be retrieved.
"""
try:
endpoint = self.config.releases()
response = self.client.get(endpoint)
rate_limit_remaining_header = 'X-RateLimit-Remaining'
rate_limit_reset_header = 'X-RateLimit-Reset'
if rate_limit_remaining_header in response.headers:
remaining_requests = int(response.headers[rate_limit_remaining_header])
if remaining_requests == 0:
reset_time = int(response.headers.get(rate_limit_reset_header, 0))
message = f'Rate limit reached. Try again at {reset_time}'
log.exception(message)
raise GitHubReleaseError(message)
if response.status_code == HTTPStatus.OK:
return response.json()
message = f'GitHub API request failed with status {response.status_code}: {response.text}'
log.exception(message)
raise GitHubReleaseError(message)
except (AttributeError, KeyError, ValueError):
message = 'Failed to retrieve GitHub releases'
log.exception(message)
raise GitHubReleaseError(message) from None
|
A method that retrieves the latest version information from GitHub.
Returns:
-
tuple
–
A tuple containing the latest version tag and release metadata.
Raises:
-
VersionCheckError
–
If version information cannot be retrieved.
Source code in dev_tool/services/update/service.py
| def get_latest_version(self) -> tuple:
"""
A method that retrieves the latest version information from GitHub.
:return: A tuple containing the latest version tag and release metadata.
:raises VersionCheckError: If version information cannot be retrieved.
"""
try:
releases = self.get_github_releases()
if releases:
latest = releases[0]
return latest.get('tag_name'), latest
except GitHubReleaseError:
raise
except Exception:
message = 'Failed to get latest version information'
log.exception(message)
raise VersionCheckError(message) from None
else:
return None, None
|
A method that replaces the current executable with the updated version.
Returns:
-
bool
–
True if replacement was successful.
Raises:
-
ExecutableReplacementError
–
If executable replacement fails.
Source code in dev_tool/services/update/service.py
| @staticmethod
def replace_executable() -> bool:
"""
A method that replaces the current executable with the updated version.
:return: True if replacement was successful.
:raises ExecutableReplacementError: If executable replacement fails.
"""
cli = STRATUS / 'cli.exe'
update = STRATUS / 'update.exe'
backup = STRATUS / 'cli.bak'
try:
if backup.exists():
backup.unlink()
if cli.exists():
cli.rename(backup)
log.debug('Detected an existing CLI executable and created a backup before updating.')
else:
log.debug('No existing CLI executable found. Installing a new CLI.')
if not update.exists():
message = f'Update file {update} not found'
log.exception(message)
raise ExecutableReplacementError(message)
update.rename(cli)
if backup.exists():
log.debug('The CLI executable has been successfully updated.')
else:
log.debug('A new CLI executable has been successfully installed.')
if backup.exists():
backup.unlink()
log.debug('Update process completed.')
except (FileNotFoundError, OSError, PermissionError):
message = 'An error occurred during the update process'
log.exception(message)
try:
if backup.exists() and not cli.exists():
backup.rename(cli)
if update.exists():
update.unlink()
log.debug('Update failed and rollback was performed.')
except Exception:
log.exception('Failed to rollback after update error')
raise ExecutableReplacementError(message) from None
else:
return True
|
A method that runs the complete update process.
Returns:
-
bool
–
True if an update was downloaded, False if already up to date, None if failed.
Source code in dev_tool/services/update/service.py
| def run(self) -> bool:
"""
A method that runs the complete update process.
:return: True if an update was downloaded, False if already up to date, None if failed.
"""
executable = STRATUS / 'cli.exe'
current = CONTEXT.configuration.get_current_version()
try:
latest, metadata = self.get_latest_version() or (current, None)
except (GitHubReleaseError, VersionCheckError):
message = 'Failed to check for updates'
log.exception(message)
return False
if current == latest and executable.exists():
message = 'You are running the latest version.'
self.notification.normal_text(message)
log.debug(message)
return False
try:
if metadata is None:
message = 'No release metadata available'
self.notification.error_banner(message)
log.debug(message)
return False
asset = self.find_asset(metadata)
except AssetNotFoundError:
message = 'No suitable binary found for download.'
self.notification.warning_text(message)
log.exception(message)
return False
try:
message = 'Updating application...'
self.notification.normal_text(message)
log.debug(message)
success = self.download_binary(asset)
if success:
message = 'Update downloaded successfully. Restarting application...'
self.notification.normal_text(message)
log.debug(message)
return True
message = 'Failed to download the update.'
log.debug(message)
except BinaryDownloadError:
message = 'Failed to download update'
log.exception(message)
return False
else:
return False
|