def update_tasks()

in sync/update.py [0:0]

• 26 lines of code
• 10 McCabe index (conditional complexity)

def update_tasks(git_gecko: Repo, git_wpt: Repo, pr_id: int | None = None,
                 sync: SyncProcess | None = None) -> None:
    logger.info("Running update_tasks%s" % ("for PR %s" % pr_id if pr_id else ""))

    syncs: Iterable[SyncProcess] = []
    if not sync:
        if pr_id is not None:
            pr_syncs = downstream.DownstreamSync.load_by_obj(git_gecko, git_wpt, pr_id)
            if not pr_syncs:
                logger.error("No sync for pr_id %s" % pr_id)
                return
            assert len(pr_syncs) == 1
            syncs = [pr_syncs.pop()]
        else:
            current_landing = landing.current(git_gecko, git_wpt)
            syncs = downstream.DownstreamSync.load_by_status(git_gecko, git_wpt, "open")
            if current_landing is not None:
                syncs.add(current_landing)
    else:
        syncs = [sync]

    for sync in syncs:
        try_push = sync.latest_try_push
        if try_push and try_push.taskgroup_id:
            try:
                handle_sync("taskgroup", {"taskGroupId": try_push.taskgroup_id})
            except AbortError:
                pass