def get_pipelines()

in analytics/circleci_analyze.py [0:0]

• 17 lines of code
• 8 McCabe index (conditional complexity)

    def get_pipelines(self, project: str = 'github/pytorch/pytorch', branch: Optional[str] = None, item_count: Optional[int] = None) -> List:
        if self.is_offline():
            c = self.db.cursor()
            cmd = "SELECT json from pipelines"
            if branch is not None:
                cmd += f" WHERE branch='{branch}'"
            if item_count is not None and item_count > 0:
                cmd += f" LIMIT {item_count}"
            c.execute(cmd)
            return [json.loads(val[0]) for val in c.fetchall()]
        rc = self._get_paged_items_list(f'{self.url_prefix}/project/{project}/pipeline', {'branch': branch} if branch is not None else {}, item_count)
        for pipeline in rc:
            vcs = pipeline['vcs']
            pid, branch, revision, pser = pipeline['id'], vcs['branch'], vcs['revision'], json.dumps(pipeline)
            self.db.execute("INSERT OR REPLACE INTO pipelines(id, branch, revision, json) VALUES (?, ?, ?, ?)", (pid, branch, revision, pser))
        self.db.commit()
        return rc