### Start GitHub Actions Runner Instances Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Starts runner instances and waits for them to be ready. It creates instances, maps them to GitHub labels, and confirms their registration. ```python def start_runner_instances(self): """Start the runner instances. This function starts the runner instances and waits for them to be ready. """ print("Starting up...") # Create a GitHub instance print("Creating GitHub Actions Runner") mappings = self.provider.create_instances() instance_ids = list(mappings.keys()) github_labels = list(mappings.values()) # Output the instance mapping and labels so the stop action can use them self.provider.set_instance_mapping(mappings) # Wait for the instance to be ready print("Waiting for instance to be ready...") self.provider.wait_until_ready(instance_ids) print("Instance is ready!") # Confirm the runner is registered with GitHub for label in github_labels: print(f"Waiting for {label}...") self.gh.wait_for_runner(label, self.timeout) ``` -------------------------------- ### Install gha-runner Source: https://gha-runner.readthedocs.io/en/latest/index Install the gha-runner library using pip. ```bash pip install gha-runner ``` -------------------------------- ### start_runner_instances Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Starts the runner instances and waits for them to be ready. This function handles the creation of instances, mapping them to GitHub labels, and confirming their registration. ```APIDOC ## start_runner_instances ### Description Starts the runner instances and waits for them to be ready. This function handles the creation of instances, mapping them to GitHub labels, and confirming their registration. ### Method Not specified (likely a method within a class) ### Endpoint Not applicable (Python method) ### Parameters None ### Request Example ```python start_runner_instances() ``` ### Response None explicitly defined, but prints status messages. ``` -------------------------------- ### get Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Make a GET request to the GitHub API. This method allows for general GET requests to specified endpoints with additional keyword arguments. ```APIDOC ## get ### Description Make a GET request to the GitHub API. This method allows for general GET requests to specified endpoints with additional keyword arguments. ### Method GET ### Endpoint [ENDPOINT] ### Parameters #### Query Parameters - **endpoint** (str) - Required - 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. ### Request Example ```python client.get("/some/endpoint", params={"key": "value"}) ``` ### Response #### Success Response (200) - **response** (dict) - The JSON response from the GitHub API. #### Response Example ```json { "example": "response body" } ``` ``` -------------------------------- ### Abstract Base Class for Cloud Instance Creation Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Defines the interface for starting a cloud instance. Implementations must provide methods for creating instances, waiting for them to be ready, and setting their mapping. ```python class CreateCloudInstance(ABC): """Abstract base class for starting a cloud instance. This class defines the interface for starting a cloud instance. """ @abstractmethod def create_instances(self) -> dict[str, str]: """Create instances in the cloud provider and return their IDs. The number of instances to create is defined by the implementation. Returns ------- dict[str, str] A dictionary of instance IDs and their corresponding github runner labels. """ raise NotImplementedError @abstractmethod def wait_until_ready(self, ids: list[str], **kwargs): """Wait until instances are in a ready state. Parameters ---------- ids : list[str] A list of instance IDs to wait for. **kwargs : dict, optional Additional arguments to pass to the waiter. """ raise NotImplementedError @abstractmethod def set_instance_mapping(self, mapping: dict[str, str]): """Set the instance mapping in the environment. Parameters ---------- mapping : dict[str, str] A dictionary of instance IDs and their corresponding github runner labels. """ raise NotImplementedError ``` -------------------------------- ### GitHub Actions Workflow for AWS Self-Hosted Runner Source: https://gha-runner.readthedocs.io/en/latest/aws/index This workflow demonstrates how to start, use, and stop AWS self-hosted runners. It includes steps for configuring AWS credentials, creating a runner instance, running jobs on the self-hosted runner, and stopping the instance. Ensure you have the necessary IAM role and GH_PAT secret configured. ```yaml name: Test Self-Hosted Runner on: workflow_dispatch: jobs: start-aws-runner: runs-on: ubuntu-latest permissions: id-token: write contents: read outputs: mapping: ${{ steps.aws-start.outputs.mapping }} instances: ${{ steps.aws-start.outputs.instances }} steps: - name: Configure AWS credentials uses: aws-actions/configure-aws-credentials@v4 with: role-to-assume: aws-region: - name: Create cloud runner id: aws-start uses: omsf/start-aws-gha-runner@v1.0.0 with: aws_image_id: aws_instance_type: aws_home_dir: /home/ubuntu env: GH_PAT: ${{ secrets.GH_PAT }} self-hosted-test: runs-on: ${{ fromJSON(needs.start-aws-runner.outputs.instances) }} # This ensures that you only run on the instances you just provisioned needs: - start-aws-runner steps: - uses: actions/checkout@v5 - name: Print disk usage run: "df -h" - name: Print Docker details run: "docker version || true" stop-aws-runner: runs-on: ubuntu-latest permissions: id-token: write contents: read needs: - start-aws-runner - self-hosted-test if: ${{ always() }} steps: - name: Configure AWS credentials uses: aws-actions/configure-aws-credentials@v4 with: role-to-assume: aws-region: - name: Stop instances uses: omsf/stop-aws-gha-runner@v1.0.0 with: instance_mapping: ${{ needs.start-aws-runner.outputs.mapping }} env: GH_PAT: ${{ secrets.GH_PAT }} notify_failure: needs: [self-hosted-test] runs-on: ubuntu-latest permissions: contents: read issues: write if: always() && (needs.stop-aws-runner.result == 'failure') steps: - name: Create issue # Customize this to your liking, add assignees if needed run: 'gh issue create --title "AWS job failed to stop correctly" --body "Job failed, please check AWS for running instances." -R omsf-eco-infra/openmm-gpu-test' env: GH_TOKEN: ${{ github.token }} ``` -------------------------------- ### Get Latest Release Information Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Retrieves the latest release information for a given repository. Raises a RuntimeError if there is an error fetching the release. ```python 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}") ``` -------------------------------- ### Make a GET request to the GitHub API Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Use this method to make GET requests to the GitHub API. It accepts an endpoint and additional keyword arguments for the request. ```python 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) ``` -------------------------------- ### Get Instance Mapping Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Retrieves the mapping of instance IDs to their corresponding GitHub runner labels. This is an abstract method that must be implemented by subclasses. ```python @abstractmethod def get_instance_mapping(self) -> dict[str, str]: """Get the instance mapping from the environment. Returns ------- dict[str, str] A dictionary of instance IDs and their corresponding github runner labels. """ raise NotImplementedError ``` -------------------------------- ### Get a runner by label Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Finds and returns a self-hosted runner matching the specified label within the repository. Raises MissingRunnerLabel if no runner with the given label is found. ```python 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") ``` -------------------------------- ### Get Latest Runner Download URL Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Fetches the download URL for the latest GitHub Actions runner for a specified platform and architecture. Raises ValueError for unsupported platforms/architectures and RuntimeError if the runner release is not found. ```python 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 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} and architecture {architecture}" ) ``` -------------------------------- ### create_runner_tokens Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Generates a specified number of registration tokens for GitHub Actions runners. This method is useful for automating the setup of new runners. ```APIDOC ## create_runner_tokens ### Description Generate registration tokens for GitHub Actions runners. This can be removed if this is added into PyGitHub. ### Method Signature create_runner_tokens(count: int) -> list[str] ### Parameters #### Path Parameters - **`count`** (int) - Required - 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. ``` -------------------------------- ### Get a list of self-hosted runners Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Retrieves all self-hosted runners for a repository, paginating through results if necessary. Raises RunnerListError if the API response is not a mapping object or if there's an HTTP error. ```python 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?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 ``` -------------------------------- ### Get Latest Runner Release Download URL Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Retrieves the download URL for the latest GitHub Actions runner for a specified platform and architecture. It validates the provided platform and architecture against supported options and raises errors if they are invalid or if the runner is not found. ```python 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 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} and architecture {architecture}" ) ``` -------------------------------- ### Initialize GitHubInstance Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Instantiate the GitHubInstance class with your GitHub API token and repository name. The repository should be in the format 'owner/repo'. ```python from gha_runner.gh import GitHubInstance instance = GitHubInstance(token="YOUR_GITHUB_TOKEN", repo="owner/repo") ``` -------------------------------- ### Initialize and Use EnvVarBuilder Source: https://gha-runner.readthedocs.io/en/latest/api/helper/input/index Demonstrates how to initialize EnvVarBuilder with environment variables and chain methods to parse and build a parameter dictionary. Use this to configure and extract multiple environment variables with type hinting and JSON parsing. ```python env = {"MY_VAR": "123", "JSON_VAR": '{"key": "value"}'} builder = EnvVarBuilder(env) result = ( builder .with_var("MY_VAR", "my_key", type_hint=int) .with_var("JSON_VAR", "json_key", is_json=True) .build() ) ``` -------------------------------- ### Create Cloud Instances Method Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Abstract method to create instances in the cloud provider. Implementations should return a dictionary mapping instance IDs to their GitHub runner labels. ```python @abstractmethod def create_instances(self) -> dict[str, str]: """Create instances in the cloud provider and return their IDs. The number of instances to create is defined by the implementation. Returns ------- dict[str, str] A dictionary of instance IDs and their corresponding github runner labels. """ raise NotImplementedError ``` -------------------------------- ### CreateCloudInstance.wait_until_ready Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Waits until the specified cloud instances are in a ready state. Additional arguments can be passed to customize the waiting behavior. ```APIDOC ## CreateCloudInstance.wait_until_ready ### Description Wait until instances are in a ready state. ### Method Abstract Method ### Parameters #### Path Parameters - **ids** (list[str]) - Required - A list of instance IDs to wait for. #### Query Parameters - **kwargs** (dict) - Optional - Additional arguments to pass to the waiter. ### Request Example ```json { "ids": ["instance_id_1", "instance_id_2"], "timeout": 300 } ``` ### Response Example ```json { "example": "Not applicable for abstract method" } ``` ``` -------------------------------- ### DeployInstance Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Class used to deploy instances and runners. It initializes the cloud provider and manages the lifecycle of runner instances. ```APIDOC ## Class DeployInstance ### Description Class that is used to deploy instances and runners. It initializes the cloud provider and manages the lifecycle of runner instances. ### Parameters - **`provider_type`** (Type[CreateCloudInstance]) – Required - The type of cloud provider to use. - **`cloud_params`** (dict) – Required - The parameters to pass to the cloud provider. - **`gh`** (GitHubInstance) – Required - The GitHub instance to use. - **`count`** (int) – Required - The number of instances to create. - **`timeout`** (int) – Required - The timeout to use when waiting for the runner to come online ### Attributes - **`provider`** (CreateCloudInstance) – The cloud provider instance. - **`provider_type`** (Type[CreateCloudInstance]) - **`cloud_params`** (dict) - **`gh`** (GitHubInstance) - **`count`** (int) - **`timeout`** (int) ### Methods #### `__post_init__(self)` Initializes the cloud provider after the object is created. #### `start_runner_instances(self)` Starts the runner instances and waits for them to be ready. This includes creating GitHub Actions runners, setting up instance mappings, waiting for instances to be ready, and confirming runner registration with GitHub. ``` -------------------------------- ### Set Instance Mapping Method Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Abstract method to set the instance mapping in the environment. Requires a dictionary mapping instance IDs to their GitHub runner labels. ```python @abstractmethod def set_instance_mapping(self, mapping: dict[str, str]): """Set the instance mapping in the environment. Parameters ---------- mapping : dict[str, str] A dictionary of instance IDs and their corresponding github runner labels. """ raise NotImplementedError ``` -------------------------------- ### CreateCloudInstance.create_instances Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Creates instances in the cloud provider and returns their IDs. The number of instances to create is determined by the specific implementation. ```APIDOC ## CreateCloudInstance.create_instances ### Description Create instances in the cloud provider and return their IDs. The number of instances to create is defined by the implementation. ### Method Abstract Method ### Returns - **dict[str, str]** - A dictionary of instance IDs and their corresponding github runner labels. ### Request Example ```json { "example": "Not applicable for abstract method" } ``` ### Response Example ```json { "example": "{'instance_id_1': 'label1', 'instance_id_2': 'label2'}" } ``` ``` -------------------------------- ### Wait Until Ready Method Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Abstract method to wait until instances are in a ready state. Accepts a list of instance IDs and optional keyword arguments for the waiter. ```python @abstractmethod def wait_until_ready(self, ids: list[str], **kwargs): """Wait until instances are in a ready state. Parameters ---------- ids : list[str] A list of instance IDs to wait for. **kwargs : dict, optional Additional arguments to pass to the waiter. """ raise NotImplementedError ``` -------------------------------- ### DeployInstance Class Definition Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Defines the DeployInstance class for managing cloud-based runner deployments. It initializes the cloud provider and configures necessary parameters for runner creation. ```python from dataclasses import dataclass, field from typing import Type # Assuming GitHubInstance and CreateCloudInstance are defined elsewhere # class GitHubInstance: # def create_runner_tokens(self, count: int) -> list: # pass # def get_latest_runner_release(self, platform: str, architecture: str) -> str: # pass # def wait_for_runner(self, label: str, timeout: int): # pass # # class CreateCloudInstance: # def __init__(self, **kwargs): # pass # def create_instances(self) -> dict: # pass # def set_instance_mapping(self, mappings: dict): # pass # def wait_until_ready(self, instance_ids: list): # pass @dataclass class DeployInstance: """Class that is used to deploy instances and runners. Parameters ---------- provider_type : Type[CreateCloudInstance] The type of cloud provider to use. cloud_params : dict The parameters to pass to the cloud provider. gh : GitHubInstance The GitHub instance to use. count : int The number of instances to create. timeout : int The timeout to use when waiting for the runner to come online Attributes ---------- provider : CreateCloudInstance The cloud provider instance provider_type : Type[CreateCloudInstance] cloud_params : dict gh : GitHubInstance count : int timeout : int """ provider_type: Type[CreateCloudInstance] cloud_params: dict gh: GitHubInstance count: int timeout: int provider: CreateCloudInstance = field(init=False) def __post_init__(self): """Initialize the cloud provider. This function is called after the object is created to correctly init the provider. """ # We need to create runner tokens for use by the provider runner_tokens = self.gh.create_runner_tokens(self.count) self.cloud_params["gh_runner_tokens"] = runner_tokens architecture = self.cloud_params.get("arch", "x64") release = self.gh.get_latest_runner_release( platform="linux", architecture=architecture ) self.cloud_params["runner_release"] = release self.provider = self.provider_type(**self.cloud_params) def start_runner_instances(self): """Start the runner instances. This function starts the runner instances and waits for them to be ready. """ print("Starting up...") # Create a GitHub instance print("Creating GitHub Actions Runner") mappings = self.provider.create_instances() instance_ids = list(mappings.keys()) github_labels = list(mappings.values()) # Output the instance mapping and labels so the stop action can use them self.provider.set_instance_mapping(mappings) # Wait for the instance to be ready print("Waiting for instance to be ready...") self.provider.wait_until_ready(instance_ids) print("Instance is ready!") # Confirm the runner is registered with GitHub for label in github_labels: print(f"Waiting for {label}...") self.gh.wait_for_runner(label, self.timeout) ``` -------------------------------- ### GitHubInstance Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Manages GitHub repository actions through the GitHub API. It requires a token for authentication and the repository name. ```APIDOC ## Class GitHubInstance ### Description Manages GitHub repository actions through the GitHub API. It requires a token for authentication and the repository name. ### Parameters #### Constructor 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. ``` -------------------------------- ### Wait Until Cloud Instances Are Removed Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Abstract method to wait until instances are removed. Accepts a list of instance IDs and optional keyword arguments for the waiter. ```python @abstractmethod def wait_until_removed(self, ids: list[str], **kwargs): """Wait until instances are removed. Parameters ---------- ids : list[str] A list of instance IDs to wait for. **kwargs : dict, optional Additional arguments to pass to the waiter. """ raise NotImplementedError ``` -------------------------------- ### Output Workflow Command Source: https://gha-runner.readthedocs.io/en/latest/api/helper/workflow_cmds/index Writes a key-value pair to the GitHub Actions output file. This allows subsequent steps in the workflow to access the value. ```python def output(name: str, value: str): with open(os.environ["GITHUB_OUTPUT"], "a") as output: output.write(f"{name}={value}\n") ``` -------------------------------- ### CreateCloudInstance.set_instance_mapping Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Sets the mapping between instance IDs and their corresponding GitHub runner labels within the environment. ```APIDOC ## CreateCloudInstance.set_instance_mapping ### Description Set the instance mapping in the environment. ### Method Abstract Method ### Parameters #### Path Parameters - **mapping** (dict[str, str]) - Required - A dictionary of instance IDs and their corresponding github runner labels. ### Request Example ```json { "mapping": {"instance_id_1": "label1", "instance_id_2": "label2"} } ``` ### Response Example ```json { "example": "Not applicable for abstract method" } ``` ``` -------------------------------- ### GitHubInstance Class Definition Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index The source code for the GitHubInstance class, showing its constructor and internal methods for API interaction. Note that methods like `_headers`, `_do_request`, and `post` are internal helpers. ```python 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.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. ``` -------------------------------- ### Check Required Environment Variables Source: https://gha-runner.readthedocs.io/en/latest/api/helper/input/index Verifies if all specified environment variables are present in the provided environment dictionary. Raises a ValueError if any are missing. ```python def check_required(env: Dict[str, str], required: list[str]): """Check if required environment variables are set. Parameters ---------- env : Dict[str, str] The environment variables. required : list[str] A list of required environment variables. Raises ------ ValueError If any required environment variables are missing. """ missing = [var for var in required if not env.get(var)] if missing: raise ValueError(f"Missing required environment variables: {missing}") ``` -------------------------------- ### StopCloudInstance.get_instance_mapping Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Retrieves the mapping of instance IDs to their corresponding GitHub runner labels from the environment. This is an abstract method. ```APIDOC ## get_instance_mapping ### Description Retrieves the mapping of instance IDs to their corresponding GitHub runner labels from the environment. This is an abstract method. ### Method Not specified (abstract method) ### Endpoint Not applicable (abstract method) ### Parameters None ### Request Example ```python get_instance_mapping() ``` ### Response #### Success Response - **dict[str, str]** - A dictionary of instance IDs and their corresponding github runner labels. ### Response Example ```json { "instance-1": "label-a", "instance-2": "label-b" } ``` ``` -------------------------------- ### Create GitHub Actions Runner Registration Token Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Generates a registration token required for registering GitHub Actions runners. Raises a TokenRetrievalError if token generation fails. ```python 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}") ``` -------------------------------- ### Create GitHub Actions Runner Tokens Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Generates a specified number of registration tokens for GitHub Actions runners. This method is useful for provisioning new runners. ```python 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 ``` -------------------------------- ### post Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Makes a POST request to the GitHub API. ```APIDOC ## POST [endpoint] ### Description Makes a POST request to the GitHub API. ### Method POST ### Endpoint [endpoint] ### Parameters #### Path Parameters - **`endpoint`** (str) - The endpoint to make the request to. #### Request Body - **`**kwargs`** (dict) - Additional keyword arguments to pass to the request. See the requests.post documentation for more information. ``` -------------------------------- ### TeardownInstance Class Definition Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Defines the TeardownInstance class for tearing down cloud instances and runners. It includes initialization logic for the cloud provider and methods to stop runners and remove instances. ```python from dataclasses import dataclass, field from typing import Type from gha_runner.cloud.cloudprovider import StopCloudInstance from gha_runner.github.github import GitHubInstance, MissingRunnerLabel from gha_runner.logging import error, warning @dataclass class TeardownInstance: """Class that is used to teardown instances and runners. Parameters ---------- provider_type : Type[StopCloudInstance] The type of cloud provider to use. cloud_params : dict The parameters to pass to the cloud provider. gh : GitHubInstance The GitHub instance to use. Attributes ---------- provider : StopCloudInstance The cloud provider instance provider_type : Type[StopCloudInstance] cloud_params : dict gh : GitHub """ provider_type: Type[StopCloudInstance] cloud_params: dict gh: GitHubInstance provider: StopCloudInstance = field(init=False) def __post_init__(self): """Initialize the cloud provider. This function is called after the object is created to correctly stop the provider. """ self.provider = self.provider_type(**self.cloud_params) def stop_runner_instances(self): """Stop the runner instances. This function stops the runner instances and waits for them to be removed. """ print("Shutting down...") try: # Get the instance mapping from our input mappings = self.provider.get_instance_mapping() except Exception as e: error(title="Malformed instance mapping", message=e) exit(1) # Remove the runners and instances print("Removing GitHub Actions Runner") instance_ids = list(mappings.keys()) labels = list(mappings.values()) for label in labels: try: print(f"Removing runner {label}") self.gh.remove_runner(label) # This occurs when we have a runner that might already be shutdown. # Since we are mainly using the ephemeral runners, we expect this to happen except MissingRunnerLabel: print(f"Runner {label} does not exist, skipping...") continue # This is more of the case when we have a failure to remove the runner # This is not a concern for the user (because we will remove the instance anyways), # but we should log it for debugging purposes. except Exception as e: warning(title="Failed to remove runner", message=e) print("Removing instances...") self.provider.remove_instances(instance_ids) print("Waiting for instance to be removed...") try: self.provider.wait_until_removed(instance_ids) except Exception as e: # Print to stdout print( f"Failed to remove instances check your provider console: {e}" ) # Print to Annotations error( title="Failed to remove instances, check your provider console", message=e, ) exit(1) else: print("Instances removed!") ``` -------------------------------- ### Generate Multiple Runner Tokens Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Use the `create_runner_tokens` method to generate a specified number of registration tokens for GitHub Actions runners. This is useful for setting up multiple runners simultaneously. ```python tokens = instance.create_runner_tokens(count=3) print(tokens) ``` -------------------------------- ### EnvVarBuilder Source: https://gha-runner.readthedocs.io/en/latest/api/helper/input/index A builder class for constructing a dictionary of parameters from environment variables. It provides a fluent interface for parsing and transforming environment variables into a structured dictionary, with support for JSON parsing and type conversion. ```APIDOC ## EnvVarBuilder ### Description A builder class for constructing a dictionary of parameters from environment variables. This class provides a fluent interface for parsing and transforming environment variables into a structured dictionary, with support for JSON parsing and type conversion. ### Parameters #### `env` (Dict[str, str]) - Required The environment variables. ### Methods #### `__init__(self, env: Dict[str, str])` Initializes the EnvVarBuilder with a dictionary of environment variables. #### `update_state(self, var_name: str, key: str, is_json: bool = False, allow_empty: bool = False, type_hint: Type = str) -> "EnvVarBuilder"` Updates the state of the builder with a single parameter. This method supports method chaining. - **`var_name`** (str) - Required - The name of the environment variable to parse. - **`key`** (str) - Required - The key to use for the parsed parameter in the resulting dictionary. - **`is_json`** (bool) - Optional - If True, parse the value as JSON. Defaults to False. - **`allow_empty`** (bool) - Optional - If True, include empty strings in the result. Defaults to False. - **`type_hint`** (Type) - Optional - The type to convert the value to (if not JSON). Defaults to str. ### Properties #### `params` (dict) Returns a copy of the dictionary of parsed parameters. ``` -------------------------------- ### EnvVarBuilder Source: https://gha-runner.readthedocs.io/en/latest/api/helper/input/index A builder class for constructing a dictionary of parameters from environment variables. It provides a fluent interface for parsing and transforming environment variables into a structured dictionary, with support for JSON parsing and type conversion. ```APIDOC ## EnvVarBuilder ### Description A builder class for constructing a dictionary of parameters from environment variables. This class provides a fluent interface for parsing and transforming environment variables into a structured dictionary, with support for JSON parsing and type conversion. ### Attributes - **`env`** (`Dict[str, str]`) – The environment variables. - **`params`** (`dict`) – The dictionary of parsed parameters. ### Parameters - **`env`** (`Dict[str, str]`) – The environment variables. ### Methods #### `with_var(self, name: str, alias: str, type_hint: Optional[type] = None, is_json: bool = False, allow_empty: bool = False)` Adds an environment variable to the builder, with optional type hinting and JSON parsing. #### `build(self) -> dict` Builds and returns the dictionary of parsed parameters. ### Example ```python env = {"MY_VAR": "123", "JSON_VAR": '{"key": "value"}'} builder = EnvVarBuilder(env) result = ( builder .with_var("MY_VAR", "my_key", type_hint=int) .with_var("JSON_VAR", "json_key", is_json=True) .build() ) # result = {"my_key": 123, "json_key": {"key": "value"}} ``` ### Notes - The builder creates deep copies of values to prevent mutation. - JSON parsing is performed before type conversion. - Empty strings are ignored by default unless allow_empty is True. ``` -------------------------------- ### Remove Cloud Instances Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Abstract method to remove instances from the cloud provider. Requires a list of instance IDs to be removed. ```python @abstractmethod def remove_instances(self, ids: list[str]): """Remove instances from the cloud provider. Parameters ---------- ids : list[str] A list of instance IDs to remove. """ raise NotImplementedError ``` -------------------------------- ### Build Dictionary from Environment Variables Source: https://gha-runner.readthedocs.io/en/latest/api/helper/input/index Use EnvVarBuilder to parse environment variables into a dictionary, with support for type hinting and JSON parsing. The builder creates deep copies of values and performs JSON parsing before type conversion. Empty strings are ignored by default. ```python >>> env = {"MY_VAR": "123", "JSON_VAR": '{"key": "value"}'} >>> builder = EnvVarBuilder(env) >>> result = (builder ... .with_var("MY_VAR", "my_key", type_hint=int) ... .with_var("JSON_VAR", "json_key", is_json=True) ... .build()) >>> # result = {"my_key": 123, "json_key": {"key": "value"}} ``` -------------------------------- ### Abstract Base Class for Stopping Cloud Instances Source: https://gha-runner.readthedocs.io/en/latest/api/clouddeployment/index Defines the interface for stopping a cloud instance. Subclasses must implement methods for removing instances and waiting for their removal. ```python class StopCloudInstance(ABC): """Abstract base class for stopping a cloud instance. This class defines the interface for stopping a cloud instance. """ @abstractmethod def remove_instances(self, ids: list[str]): """Remove instances from the cloud provider. Parameters ---------- ids : list[str] A list of instance IDs to remove. """ raise NotImplementedError @abstractmethod def wait_until_removed(self, ids: list[str], **kwargs): """Wait until instances are removed. Parameters ---------- ids : list[str] A list of instance IDs to wait for. **kwargs : dict, optional Additional arguments to pass to the waiter. """ raise NotImplementedError @abstractmethod def get_instance_mapping(self) -> dict[str, str]: """Get the instance mapping from the environment. Returns ------- dict[str, str] A dictionary of instance IDs and their corresponding github runner labels. """ raise NotImplementedError ``` -------------------------------- ### create_runner_tokens Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Generates a specified number of registration tokens for GitHub Actions runners. ```APIDOC ## POST /repos/{owner}/{repo}/actions/runners/registration-token ### Description Generates a specified number of registration tokens for GitHub Actions runners. ### Method POST ### Endpoint `/repos/{self.repo}/actions/runners/registration-token` ### Parameters #### Query 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. ``` -------------------------------- ### Warning Workflow Command Source: https://gha-runner.readthedocs.io/en/latest/api/helper/workflow_cmds/index Prints a warning message to the GitHub Actions workflow log with an optional title. Use this to indicate potential issues without failing the workflow. ```python def warning(title: str, message): print(f"::warning title={title}::{message}") ``` -------------------------------- ### post Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Make a POST request to the GitHub API. This method can be used to send data to various GitHub API endpoints. Additional keyword arguments can be passed to customize the request, such as headers or data payloads. ```APIDOC ## post ### Description Make a POST request to the GitHub API. This can be removed if this is added into PyGitHub. ### Method POST ### Endpoint `/endpoint` ### Parameters #### Path Parameters - **endpoint** (str) - The endpoint to make the request to. #### Query Parameters None #### Request Body None (Additional keyword arguments are passed to the requests.post function) ### Request Example ```python # Example usage (assuming 'gh' is an instance of the GitHub API client) gh.post("some/endpoint", data={"key": "value"}) ``` ### Response #### Success Response (200) Depends on the specific GitHub API endpoint called. #### Response Example ```json { "message": "Success" } ``` ``` -------------------------------- ### create_runner_token Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Generates a single registration token for GitHub Actions runners. ```APIDOC ## POST /repos/{owner}/{repo}/actions/runners/registration-token ### Description Generates a single registration token for GitHub Actions runners. ### Method POST ### Endpoint `/repos/{self.repo}/actions/runners/registration-token` ### Returns - **`str`** - A runner registration token. ### Raises - **`TokenRetrievalError`** - If there is an error generating the token. ``` -------------------------------- ### Make a POST request to the GitHub API Source: https://gha-runner.readthedocs.io/en/latest/api/gh/index Use this method to make POST requests to the GitHub API. It accepts an endpoint and additional keyword arguments for the request. ```python def post(self, endpoint, **kwargs): """Make a POST 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.post documentation for more information. """ return self._do_request(requests.post, endpoint, **kwargs) ``` -------------------------------- ### output Source: https://gha-runner.readthedocs.io/en/latest/api/helper/workflow_cmds/index Sets an output variable for the GitHub Actions workflow. This output can be used by subsequent steps. ```APIDOC ## output ### Description Sets an output variable for the GitHub Actions workflow. This output can be used by subsequent steps. ### Signature ```python output(name: str, value: str) ``` ### Parameters * **name** (str) - The name of the output variable. * **value** (str) - The value to set for the output variable. ```