Download the latest asset from Github

majordome/__init__.py

@@ -0,0 +1,5 @@
+import logging.config
+from pathlib import Path
+
+current_path = Path.cwd()
+logging.config.fileConfig(str(current_path / "logging.conf"))

majordome/errors.py

@@ -0,0 +1,14 @@
+class NoReleaseFound(Exception):
+    pass
+
+
+class NoAssetMatchingMnemonic(Exception):
+    pass
+
+
+class AssetNotFound(Exception):
+    pass
+
+
+class DownloadFailure(Exception):
+    pass

majordome/github_service.py

@@ -0,0 +1,85 @@
+from typing import Any
+import logging
+import requests
+
+from majordome.errors import NoReleaseFound, AssetNotFound, DownloadFailure
+
+logger = logging.getLogger(__name__)
+
+
+class GithubConnector:
+    def __init__(self, token: str):
+        self.base_url = "https://api.github.com"
+        self.token = token
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return {
+            "Accept": "application/vnd.github.v3+json",
+            "Authorization": f"Bearer {self.token}",
+            "X-GitHub-Api-Version": "2022-11-28",
+        }
+
+    def get_latest_release(self, owner: str, repo: str) -> dict[str, Any]:
+        logger.info("Getting latest release from Github")
+        logger.debug(
+            "GET %s with headers %s", self.latest_release_url(owner, repo), self.headers
+        )
+        response = requests.get(
+            self.latest_release_url(owner, repo), headers=self.headers
+        )
+        if response.ok:
+            logger.debug("Latest release found")
+            return response.json()[0]
+        logger.error(
+            "Failed to get latest release from Github. Response code was: %s",
+            response.status_code,
+        )
+        raise NoReleaseFound(
+            f"No release found on Github for the repo {repo} from owner {owner}"
+        )
+
+    @staticmethod
+    def latest_release_url(owner: str, repo: str) -> str:
+        return f"https://api.github.com/repos/{owner}/{repo}/releases?per_page=1"
+
+    def download_asset(self, owner: str, repo: str, asset_id: int) -> bytes:
+        logger.info(
+            "Downloading asset with id %s from Github owner %s and repo %s",
+            asset_id,
+            owner,
+            repo,
+        )
+        asset_response = requests.get(
+            self.asset_url(owner, repo, asset_id), headers=self.headers
+        )
+        if asset_response.ok:
+            download_url = asset_response.json().get("browser_download_url")
+            downloaded_response = requests.get(download_url, headers=self.headers)
+            if downloaded_response.ok:
+                return downloaded_response.content
+
+            logger.error(
+                "Failed to download asset from Github repo %s of owner %s. Response code was: %s",
+                repo,
+                owner,
+                downloaded_response.status_code,
+            )
+            raise DownloadFailure(
+                f"The asset {asset_id} was not found on Github for the repo {repo} from owner {owner}"
+            )
+
+        logger.error(
+            "Failed to find asset with id %s from Github repo %s of owner %s. Response code was: %s",
+            asset_id,
+            repo,
+            owner,
+            asset_response.status_code,
+        )
+        raise AssetNotFound(
+            f"The asset {asset_id} was not found on Github for the repo {repo} from owner {owner}"
+        )
+
+    @staticmethod
+    def asset_url(owner: str, repo: str, asset_id: int) -> str:
+        return f"https://api.github.com/repos/{owner}/{repo}/releases/assets/{asset_id}"

majordome/settings.py

@@ -0,0 +1,8 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic import Field
+
+
+class GithubSettings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="github_", env_file=".env")
+
+    token: str = Field(default="")

majordome/software_repo.py

@@ -0,0 +1,51 @@
+import logging
+
+from majordome.github_service import GithubConnector
+from majordome.errors import NoAssetMatchingMnemonic
+
+logger = logging.getLogger(__name__)
+
+
+class SoftwareRepo:
+    def __init__(
+        self,
+        url: str,
+        github_connector: GithubConnector,
+        tag_name_mnemonic: str = "prefix-{{VERSION}}-suffix",
+        asset_mnemonic: str = "prefix-{{VERSION}}-suffix",
+    ):
+        self.url = url
+        self.github_connector = github_connector
+        self.owner, self.repo = url.removeprefix("https://github.com/").split("/")
+        self.tag_prefix, self.tag_suffix = tag_name_mnemonic.split("{{VERSION}}")
+        self.asset_mnemonic = asset_mnemonic
+
+    def get_latest_version(self) -> str:
+        logger.info("Getting latest version of %s", self.url)
+        release = self.github_connector.get_latest_release(self.owner, self.repo)
+
+        tag_name = release.get("tag_name")
+        version = tag_name.removeprefix(self.tag_prefix).removesuffix(self.tag_suffix)
+        logger.info("Latest version for %s is %s", self.url, version)
+        return version
+
+    def get_latest_asset_id(self) -> int:
+        logger.info("Getting asset version of %s", self.url)
+        latest_release = self.github_connector.get_latest_release(self.owner, self.repo)
+        assets = latest_release.get("assets", [])
+
+        version = self.get_latest_version()
+        asset_name = self.asset_mnemonic.replace("{{VERSION}}", version)
+
+        for asset in assets:
+            if asset.get("name") == asset_name:
+                return asset.get("id")
+
+        raise NoAssetMatchingMnemonic(
+            f"No asset found matching {self.asset_mnemonic} on {self.url}"
+        )
+
+    def download_latest_asset(self) -> bytes:
+        logger.info("Downloading latest asset for %s", self.url)
+        id_ = self.get_latest_asset_id()
+        return self.github_connector.download_asset(self.owner, self.repo, id_)