def update_pending()

in tools/scripts/update_pending_hud.py [0:0]

• 41 lines of code
• 11 McCabe index (conditional complexity)

def update_pending(branch: str = "master") -> None:
    commit_index = s3_get_json(bucket_name, f'{branch}/index.json', [])
    for idx, item in enumerate(commit_index):
        commit_id = item['id']
        title = item['message'].split("\n")[0]
        timestamp = datetime.fromisoformat(item['timestamp']).replace(tzinfo=None)
        if datetime.utcnow() - timestamp < timedelta(hours=5):
            print(f"[{idx}/{len(commit_index)}] {title} ( {commit_id} ) is skipped as it was merged less than 5 hours ago")
            continue
        has_pending, has_updates = False, False
        job_statuses = s3_get_json(bucket_name, f'{branch}/{commit_id}.json', {})
        for (name, value) in job_statuses.items():
            status = value['status']
            build_url = value['build_url']
            if status not in ['success', 'skipped', 'error', 'failure']:
                circle_match = re.match("https://circleci.com/gh/pytorch/pytorch/(\\d+)\\?", build_url)
                ghrun_match = re.match("https://github.com/pytorch/pytorch/runs/(\\d+)", build_url)
                if circle_match is not None:
                    job_id = int(circle_match.group(1))
                    job_status = circleci_get_job_status("pytorch", "pytorch", job_id)
                    circle_status = map_circle_status(job_status['status'])
                    if status != circle_status:
                        job_statuses[name]['status'] = circle_status
                        has_updates = True
                        continue
                if ghrun_match is not None:
                    run_id = int(ghrun_match.group(1))
                    check_status = gh_get_runs_status("pytorch", "pytorch", run_id)
                    ghrun_status = map_ghrun_status(check_status['status'])
                    if status != ghrun_status:
                        job_statuses[name]['status'] = ghrun_status
                        has_updates = True
                        continue
                has_pending = True
        if has_updates:
            print(f"[{idx}/{len(commit_index)}] {title} ( {commit_id} ) has updates")
            s3.Object(bucket_name, f'{branch}/{commit_id}.json').put(Body=json_dumps(job_statuses))
        elif has_pending:
            print(f"[{idx}/{len(commit_index)}] {title} ( {commit_id} ) has pending statuses")
        else:
            print(f"[{idx}/{len(commit_index)}] were processed")