def get_pipeline_jobs()

in analytics/circleci_analyze.py [0:0]

• 6 lines of code
• 4 McCabe index (conditional complexity)

    def get_pipeline_jobs(self, project: str = 'github/pytorch/pytorch', branch: Optional[str] = None, item_count: Optional[int] = None) -> Generator:
        for pipeline in self.get_pipelines(project, branch, item_count):
            for workflow in self.get_pipeline_workflows(pipeline['id']):
                in_progress = is_workflow_in_progress(workflow)
                for job in self.get_workflow_jobs(workflow['id'], should_cache=not in_progress):
                    yield (pipeline, workflow, job)