"""Module to manage GitHub repository actions through the GitHub API.""" import collections.abc import random import string import time import urllib.parse from dataclasses import dataclass from json import JSONDecodeError import requests class TokenRetrievalError(Exception): """Exception raised when there is an error retrieving a token from GitHub.""" # Public exception name kept as-is for backwards compatibility with callers. class MissingRunnerLabel(Exception): # noqa: N818 - existing public name """Exception raised when a runner does not exist in the repository.""" class RunnerListError(Exception): """Exception raised when there is an error getting the list of runners.""" @dataclass class SelfHostedRunner: id: int name: str os: str labels: list[str] class GitHubInstance: """Class to manage GitHub repository actions through the GitHub API. Parameters ---------- token : str GitHub API token for authentication. repo : str Full name of the GitHub repository in the format "owner/repo". Attributes ---------- headers : dict Headers for HTTP requests to GitHub API. github : Github Instance of Github object for interacting with the GitHub API. """ BASE_URL = "https://api.github.com" def __init__(self, token: str, repo: str): self.token = token self.headers = self._headers({}) self.repo = repo def _headers(self, header_kwargs): """Generate headers for API requests, adding authorization and specific API version. Can be removed if this is added into PyGitHub. Parameters ---------- header_kwargs : dict Additional headers to include in the request. Returns ------- dict Headers including authorization, API version, and any additional headers. """ headers = { "Authorization": f"Bearer {self.token}", "X-Github-Api-Version": "2022-11-28", "Accept": "application/vnd.github+json", } headers.update(header_kwargs) return headers def _do_request(self, func, endpoint, **kwargs): """Make a request to the GitHub API. This can be removed if this is added into PyGitHub. """ endpoint_url = urllib.parse.urljoin(self.BASE_URL, endpoint) headers = self.headers resp: requests.Response = func(endpoint_url, headers=headers, **kwargs) if not resp.ok: raise RuntimeError( f"Error in API call for {endpoint_url}: " f"{resp.status_code} {resp.reason}: {resp.content}" ) else: try: return resp.json() except JSONDecodeError: return resp.content def create_runner_tokens(self, count: int) -> list[str]: """Generate registration tokens for GitHub Actions runners. This can be removed if this is added into PyGitHub. Parameters ---------- count : int The number of runner tokens to generate. Returns ------- list[str] A list of runner registration tokens. Raises ------ TokenCreationError If there is an error generating the tokens. """ tokens = [] for _ in range(count): token = self.create_runner_token() tokens.append(token) return tokens def create_runner_token(self) -> str: """Generate a registration token for GitHub Actions runners. This can be removed if this is added into PyGitHub. Returns ------- str A runner registration token. Raises ------ TokenRetrievalError If there is an error generating the token. """ try: res = self.post( f"repos/{self.repo}/actions/runners/registration-token" ) return res["token"] except Exception as e: raise TokenRetrievalError(f"Error creating runner token: {e}") def post(self, endpoint, **kwargs): """Make a POST request to the GitHub API. This can be removed if this is added into PyGitHub. Parameters ---------- endpoint : str The endpoint to make the request to. **kwargs : dict, optional Additional keyword arguments to pass to the request. See the requests.post documentation for more information. """ return self._do_request(requests.post, endpoint, **kwargs) def get(self, endpoint, **kwargs): """Make a GET request to the GitHub API. Parameters ---------- endpoint : str The endpoint to make the request to. **kwargs : dict, optional Additional keyword arguments to pass to the request. See the requests.get documentation for more information. """ return self._do_request(requests.get, endpoint, **kwargs) def delete(self, endpoint, **kwargs): """Make a DELETE request to the GitHub API. Parameters ---------- endpoint : str The endpoint to make the request to. **kwargs : dict, optional Additional keyword arguments to pass to the request. See the requests.delete documentation for more information. """ return self._do_request(requests.delete, endpoint, **kwargs) def get_runners(self) -> list[SelfHostedRunner] | None: """Get a list of self-hosted runners in the repository. Returns ------- list[SelfHostedRunner] | None A list of self-hosted runners in the repository if they exist, otherwise None. Raises ------ RunnerListError If there is an error getting the list of runners. Either because of an error in the request or the response is not a mapping object. """ runners = [] per_page = 30 # GH API default page = 1 total_runners = float("inf") # paginate through the pages until we have all the runners while (len(runners) < total_runners): try: res = self.get( f"repos/{self.repo}/actions/runners" f"?per_page={per_page}&page={page}" ) # This allows for arbitrary mappable objects to be used if not isinstance(res, collections.abc.Mapping): # This could be related to the API or the request itself. # ie the response is not a JSON object raise RunnerListError( f"Did not receive mapping object: {res}" ) total_runners = res["total_count"] page += 1 # protect from bug/issue where total_count is higher than # actual # of runners if len(res["runners"]) < 1: break for runner in res["runners"]: id = runner["id"] name = runner["name"] os = runner["os"] labels = [label["name"] for label in runner["labels"]] runners.append(SelfHostedRunner(id, name, os, labels)) except RuntimeError as e: # This occurs when we receive a status code is > 400 raise RunnerListError(f"Error getting runners: {e}") # Other exceptions are bubbled up to the caller return runners if len(runners) > 0 else None def get_runner(self, label: str) -> SelfHostedRunner: """Get a runner by a given label for a repository. Returns ------- SelfHostedRunner The runner with the given label. Raises ------ MissingRunnerLabel If the runner with the given label is not found. """ runners = self.get_runners() if runners is not None: for runner in runners: if label in runner.labels: return runner raise MissingRunnerLabel(f"Runner {label} not found") def wait_for_runner( self, label: str, timeout: int, wait: int = 15 ) -> SelfHostedRunner: """Wait for the runner with the given label to be online. Parameters ---------- label : str The label of the runner to wait for. wait : int The time in seconds to wait between checks. Defaults to 15 seconds. timeout : int The maximum time in seconds to wait for the runner to be online. Returns ------- SelfHostedRunner The runner with the given label. """ max = time.time() + timeout try: runner = self.get_runner(label) return runner except MissingRunnerLabel: print(f"Waiting for runner {label}...") while True: if time.time() > max: raise RuntimeError( f"Timeout reached: Runner {label} not found" ) try: runner = self.get_runner(label) return runner except MissingRunnerLabel: print(f"Runner {label} not found. Waiting...") time.sleep(wait) def remove_runner(self, label: str): """Remove a runner by a given label. Parameters ---------- label : str The label of the runner to remove. Raises ------ RuntimeError If there is an error removing the runner or the runner is not found. """ runner = self.get_runner(label) try: self.delete(f"repos/{self.repo}/actions/runners/{runner.id}") except Exception as e: raise RuntimeError(f"Error removing runner {label}. Error: {e}") @staticmethod def generate_random_label() -> str: """Generate a random label for a runner. Returns ------- str A random label for a runner. The label is in the format "runner-". The random string is 8 characters long and consists of lowercase letters and digits. """ letters = string.ascii_lowercase + string.digits result_str = "".join(random.choice(letters) for i in range(8)) return f"runner-{result_str}" def _get_latest_release(self, repo: str) -> dict: """Get the latest release for a repository. Parameters ---------- repo : str The repository to get the latest release for. Returns ------- str The tag name of the latest release. """ try: release = self.get(f"repos/{repo}/releases/latest") return release except Exception as e: raise RuntimeError(f"Error getting latest release: {e}") def get_latest_runner_release( self, platform: str, architecture: str ) -> str: """Return the latest runner for the given platform and architecture. Parameters ---------- platform : str The platform of the runner to download. architecture : str The architecture of the runner to download. Returns ------- str The download URL of the runner. Raises ------ RuntimeError If the runner is not found for the given platform and architecture. ValueError If the platform or architecture is not supported. """ repo = "actions/runner" supported_platforms = {"linux": ["x64", "arm", "arm64"]} if platform not in supported_platforms: raise ValueError( f"Platform '{platform}' not supported. " f"Supported platforms are {list(supported_platforms)}" ) if architecture not in supported_platforms[platform]: raise ValueError( f"Architecture '{architecture}' not supported for " f"platform '{platform}'. " f"Supported architectures are {supported_platforms[platform]}" ) release = self._get_latest_release(repo) assets = release["assets"] for asset in assets: if platform in asset["name"] and architecture in asset["name"]: return asset["browser_download_url"] raise RuntimeError( f"Runner release not found for platform {platform} " f"and architecture {architecture}" )