in services/daily-ci-reports/report.py [0:0]
def all_pipelines(self) -> List[Pipeline]:
if self.pipelines is not None:
return self.pipelines
def map_pipeline(obj):
name = str(obj['name'])
if all(name not in job for job, _ in ENABLED_JOBS.items()):
logging.info(f'skipping pipeline {name}')
return
if 'MultiBranchPipelineImpl' in obj['_class']:
branches = list(filter(Pipeline.filter_branch_name, obj['branchNames']))
if not branches:
return
yield MultiBranchPipeline(name=str(obj['name']),
url=str(obj['_links']['self']['href']),
runs_url=str(obj['_links']['runs']['href']),
branches_url=str(obj['_links']['branches']['href']),
branch_names=branches)
return
if 'PipelineImpl' in obj['_class']:
yield Pipeline(name=str(obj['name']),
url=str(obj['_links']['self']['href']),
runs_url=str(obj['_links']['runs']['href']))
return
if 'io.jenkins.blueocean.service.embedded.rest.PipelineFolderImpl' in obj['_class']:
subfolders = [folder for folder in obj['pipelineFolderNames'] if folder is not None]
for child_name in subfolders:
child_url = f"{JENKINS_URL}{obj['_links']['self']['href']}pipelines/{child_name}"
child_pipeline = json.load(urllib.request.urlopen(child_url))
# Use recursive calls to map nested folders and children
mapped_child_pipelines = map_pipeline(child_pipeline)
for mapped_child_pipeline in mapped_child_pipelines:
# Rewrite children to preprend parent folder name
mapped_child_pipeline.name = f"{obj['name']}/{child_pipeline['name']}"
yield mapped_child_pipeline
return
logging.warning('Unsupported pipeline type %s for %s', obj['_class'], obj['name'])
return
self.pipelines = list()
for pipeline in self.query_org():
for mapped_pipeline in map_pipeline(pipeline):
self.pipelines.append(mapped_pipeline)
return self.pipelines