# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import argparse import os import re import tarfile import tempfile import time import zipfile from functools import wraps from typing import Any, Callable, Dict, List, Optional import gitlab TR = Callable[..., Any] def retry(func: TR) -> TR: """ This wrapper will only catch several exception types associated with "network issues" and retry the whole function. """ @wraps(func) def wrapper(self: 'Gitlab', *args: Any, **kwargs: Any) -> Any: retried = 0 while True: try: res = func(self, *args, **kwargs) except (IOError, EOFError, gitlab.exceptions.GitlabError) as e: if isinstance(e, gitlab.exceptions.GitlabError): if e.response_code == 500: # retry on this error pass elif e.response_code == 404 and os.environ.get('LOCAL_GITLAB_HTTPS_HOST', None): # remove the environment variable "LOCAL_GITLAB_HTTPS_HOST" and retry os.environ.pop('LOCAL_GITLAB_HTTPS_HOST', None) else: # other GitlabErrors aren't retried raise e retried += 1 if retried > self.DOWNLOAD_ERROR_MAX_RETRIES: raise e # get out of the loop else: print('Network failure in {}, retrying ({})'.format(getattr(func, '__name__', '(unknown callable)'), retried)) time.sleep(2 ** retried) # wait a bit more after each retry continue else: break return res return wrapper class Gitlab(object): JOB_NAME_PATTERN = re.compile(r'(\w+)(\s+(\d+)/(\d+))?') DOWNLOAD_ERROR_MAX_RETRIES = 3 def __init__(self, project_id: Optional[int] = None): config_data_from_env = os.getenv('PYTHON_GITLAB_CONFIG') if config_data_from_env: # prefer to load config from env variable with tempfile.NamedTemporaryFile('w', delete=False) as temp_file: temp_file.write(config_data_from_env) config_files = [temp_file.name] # type: Optional[List[str]] else: # otherwise try to use config file at local filesystem config_files = None self._init_gitlab_inst(project_id, config_files) @retry def _init_gitlab_inst(self, project_id: Optional[int], config_files: Optional[List[str]]) -> None: gitlab_id = os.getenv('LOCAL_GITLAB_HTTPS_HOST') # if None, will use the default gitlab server self.gitlab_inst = gitlab.Gitlab.from_config(gitlab_id=gitlab_id, config_files=config_files) self.gitlab_inst.auth() if project_id: self.project = self.gitlab_inst.projects.get(project_id) else: self.project = None @retry def get_project_id(self, name: str, namespace: Optional[str] = None) -> int: """ search project ID by name :param name: project name :param namespace: namespace to match when we have multiple project with same name :return: project ID """ projects = self.gitlab_inst.projects.list(search=name) res = [] for project in projects: if namespace is None: if len(projects) == 1: res.append(project.id) break if project.namespace['path'] == namespace: if project.name == name: res.insert(0, project.id) else: res.append(project.id) if not res: raise ValueError("Can't find project") return int(res[0]) @retry def download_artifacts(self, job_id: int, destination: str) -> None: """ download full job artifacts and extract to destination. :param job_id: Gitlab CI job ID :param destination: extract artifacts to path. """ job = self.project.jobs.get(job_id) with tempfile.NamedTemporaryFile(delete=False) as temp_file: job.artifacts(streamed=True, action=temp_file.write) with zipfile.ZipFile(temp_file.name, 'r') as archive_file: archive_file.extractall(destination) @retry def download_artifact(self, job_id: int, artifact_path: str, destination: Optional[str] = None) -> List[bytes]: """ download specific path of job artifacts and extract to destination. :param job_id: Gitlab CI job ID :param artifact_path: list of path in artifacts (relative path to artifact root path) :param destination: destination of artifact. Do not save to file if destination is None :return: A list of artifact file raw data. """ job = self.project.jobs.get(job_id) raw_data_list = [] for a_path in artifact_path: try: data = job.artifact(a_path) # type: bytes except gitlab.GitlabGetError as e: print("Failed to download '{}' from job {}".format(a_path, job_id)) raise e raw_data_list.append(data) if destination: file_path = os.path.join(destination, a_path) try: os.makedirs(os.path.dirname(file_path)) except OSError: # already exists pass with open(file_path, 'wb') as f: f.write(data) return raw_data_list @retry def find_job_id(self, job_name: str, pipeline_id: Optional[str] = None, job_status: str = 'success') -> List[Dict]: """ Get Job ID from job name of specific pipeline :param job_name: job name :param pipeline_id: If None, will get pipeline id from CI pre-defined variable. :param job_status: status of job. One pipeline could have multiple jobs with same name after retry. job_status is used to filter these jobs. :return: a list of job IDs (parallel job will generate multiple jobs) """ job_id_list = [] if pipeline_id is None: pipeline_id = os.getenv('CI_PIPELINE_ID') pipeline = self.project.pipelines.get(pipeline_id) jobs = pipeline.jobs.list(all=True) for job in jobs: match = self.JOB_NAME_PATTERN.match(job.name) if match: if match.group(1) == job_name and job.status == job_status: job_id_list.append({'id': job.id, 'parallel_num': match.group(3)}) return job_id_list @retry def download_archive(self, ref: str, destination: str, project_id: Optional[int] = None, cache_dir: Optional[str] = None) -> str: """ Download archive of certain commit of a repository and extract to destination path :param ref: commit or branch name :param destination: destination path of extracted archive file :param project_id: download project of current instance if project_id is None :return: root path name of archive file """ if project_id is None: project = self.project else: project = self.gitlab_inst.projects.get(project_id) if cache_dir: local_archive_file = os.path.join(cache_dir, f'{ref}.tar.gz') os.makedirs(os.path.dirname(local_archive_file), exist_ok=True) if os.path.isfile(local_archive_file): print('Use cached archive file. Skipping download...') else: with open(local_archive_file, 'wb') as fw: try: project.repository_archive(sha=ref, streamed=True, action=fw.write) except gitlab.GitlabGetError as e: print('Failed to archive from project {}'.format(project_id)) raise e print('Downloaded archive size: {:.03f}MB'.format(float(os.path.getsize(local_archive_file)) / (1024 * 1024))) return self.decompress_archive(local_archive_file, destination) # no cache with tempfile.NamedTemporaryFile(delete=False) as temp_file: try: project.repository_archive(sha=ref, streamed=True, action=temp_file.write) except gitlab.GitlabGetError as e: print('Failed to archive from project {}'.format(project_id)) raise e print('Downloaded archive size: {:.03f}MB'.format(float(os.path.getsize(temp_file.name)) / (1024 * 1024))) return self.decompress_archive(temp_file.name, destination) @staticmethod def decompress_archive(path: str, destination: str) -> str: with tarfile.open(path, 'r') as archive_file: root_name = archive_file.getnames()[0] archive_file.extractall(destination) return os.path.join(os.path.realpath(destination), root_name) def get_job_tags(self, job_id: int) -> str: """ Get tags of a job :param job_id: job id :return: comma-separated tags of the job """ job = self.project.jobs.get(job_id) return ','.join(job.tag_list) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument('action') parser.add_argument('project_id', type=int) parser.add_argument('--pipeline_id', '-i', type=int, default=None) parser.add_argument('--ref', '-r', default='master') parser.add_argument('--job_id', '-j', type=int, default=None) parser.add_argument('--job_name', '-n', default=None) parser.add_argument('--project_name', '-m', default=None) parser.add_argument('--destination', '-d', default=None) parser.add_argument('--artifact_path', '-a', nargs='*', default=None) args = parser.parse_args() gitlab_inst = Gitlab(args.project_id) if args.action == 'download_artifacts': gitlab_inst.download_artifacts(args.job_id, args.destination) if args.action == 'download_artifact': gitlab_inst.download_artifact(args.job_id, args.artifact_path, args.destination) elif args.action == 'find_job_id': job_ids = gitlab_inst.find_job_id(args.job_name, args.pipeline_id) print(';'.join([','.join([str(j['id']), j['parallel_num']]) for j in job_ids])) elif args.action == 'download_archive': gitlab_inst.download_archive(args.ref, args.destination) elif args.action == 'get_project_id': ret = gitlab_inst.get_project_id(args.project_name) print('project id: {}'.format(ret)) elif args.action == 'get_job_tags': ret = gitlab_inst.get_job_tags(args.job_id) print(ret) if __name__ == '__main__': main()