def list_classification_tasks()

in mozci/console/commands/push.py [0:0]

• 28 lines of code
• 12 McCabe index (conditional complexity)

    def list_classification_tasks(self, group_id):
        # Check cache first
        cache_key = f"perf/task_group/{group_id}"
        tasks = config.cache.get(cache_key, [])

        if not tasks:
            queue = taskcluster.Queue(get_taskcluster_options())
            token = False
            try:
                # Support pagination using continuation token
                while token is not None:
                    query = {"continuationToken": token} if token else {}
                    results = queue.listTaskGroup(group_id, query=query)
                    tasks += results.get("tasks")
                    token = results.get("continuationToken")
            except TaskclusterRestFailure as e:
                # Skip expired task groups
                if e.status_code == 404:
                    return

                raise
        else:
            logger.debug("From cache", cache_key)

        for task_status in tasks:
            task_id = task_status["status"]["taskId"]

            # Skip decision task
            if task_id == group_id:
                continue

            # Only provide completed tasks
            if task_status["status"]["state"] != "completed":
                logger.debug(f"Skip not completed task {task_id}")
                continue

            yield task_status

        # Cache all tasks if all completed
        if all(t["status"]["state"] == "completed" for t in tasks):
            config.cache.add(cache_key, tasks, int(config["cache"]["retention"]))