def find_task_group_id()

in mozperftest_tools/mozperftest_tools/utils/utils.py [0:0]

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

def find_task_group_id(revision, branch, search_crons=False, cache=None):
    # Find the task IDs from this revision first
    task_ids_url = TASK_IDS.format(branch, revision)

    cached_filename = f"task-ids-data-{branch}-{revision}.json"
    task_ids_data = _retrieve_from_cache(cached_filename, cache)
    if task_ids_data is None:
        print("Downloading task ids from: %s" % task_ids_url)
        task_ids_data = get_json(task_ids_url)
        _store_to_cache(task_ids_data, cached_filename, cache)
    if "tasks" not in task_ids_data or len(task_ids_data["tasks"]) == 0:
        raise Exception("Cannot find any task IDs for %s!" % revision)

    task_group_ids = []
    for task in task_ids_data["tasks"]:
        # Only find the task group ID for the decision task if we
        # don't need to search for cron tasks
        if not search_crons and not task["namespace"].endswith("decision"):
            continue
        task_group_url = TASK_INFO.format(task["taskId"])

        cached_filename = f"task-group-{task['taskId']}-{branch}-{revision}.json"
        task_info = _retrieve_from_cache(cached_filename, cache)
        if task_info is None:
            print("Downloading task group id from: %s" % task_group_url)
            task_info = get_json(task_group_url)
            _store_to_cache(task_info, cached_filename, cache)

        task_group_ids.append(task_info["taskGroupId"])

    return task_group_ids