circleci-wait-completion.py

import os
from time import sleep
import requests
from typing import Any


class CircleCI:
    def __init__(self, token: str, project_username: str, project_reponame: str):
        self._token = token
        self._project = f'{project_username}/{project_reponame}'

    @staticmethod
    def from_env() -> 'CircleCI':
        circle_token = os.environ['CIRCLE_TOKEN']
        circle_project_username = os.environ['CIRCLE_PROJECT_USERNAME']
        circle_project_reponame = os.environ['CIRCLE_PROJECT_REPONAME']

        return CircleCI(circle_token, circle_project_username, circle_project_reponame)

    def _request(self, url: str, tries: int = 5) -> Any:
        try:
            return requests.get(url, params={'circle-token': self._token}).json()
        except Exception as e:
            print(f"Exception when fetching '{url}': {e}")

            if tries > 1:
                sleep(1)
                return self._request(url, tries - 1)

    def job_details(self, job_id: str) -> Any:
        url = f"https://circleci.com/api/v2/project/gh/{self._project}/job/{job_id}"
        return self._request(url)


def running_parallel_runs_count(ci: CircleCI, circle_build_num: str) -> int:
    runs = ci.job_details(circle_build_num)['parallel_runs']
    return len(list(filter(lambda run: run['status'] == 'running', runs)))


#
# Wait for other parallel runs to finish.
#
def wait_for_completion() -> None:
    ci = CircleCI.from_env()
    circle_build_num = os.environ['CIRCLE_BUILD_NUM']

    while count := running_parallel_runs_count(ci, circle_build_num) - 1:
        print(f'Running parallel runs: {count + 1}')
        sleep(5)


if __name__ == "__main__":
    wait_for_completion()