diff --git a/taskcluster/fxci_config_taskgraph/util/integration.py b/taskcluster/fxci_config_taskgraph/util/integration.py index a12b1e55..015b819e 100644 --- a/taskcluster/fxci_config_taskgraph/util/integration.py +++ b/taskcluster/fxci_config_taskgraph/util/integration.py @@ -4,16 +4,17 @@ import copy from functools import cache -from typing import Any, Dict, List, Union +from typing import Any import requests import taskcluster -from taskgraph.util.taskcluster import logging, get_task_url, get_session from taskgraph.util.memoize import memoize +from taskgraph.util.taskcluster import get_session, get_task_url, logging FIREFOXCI_ROOT_URL = "https://firefox-ci-tc.services.mozilla.com" STAGING_ROOT_URL = "https://stage.taskcluster.nonprod.cloudops.mozgcp.net" + def _do_request(url, method=None, **kwargs): if method is None: method = "post" if kwargs else "get" @@ -60,9 +61,7 @@ def _get_deps(task_ids, use_proxy): return upstream_tasks -def get_ancestors( - task_ids: Union[List[str], str], use_proxy: bool = False -) -> Dict[str, str]: +def get_ancestors(task_ids: list[str] | str, use_proxy: bool = False) -> dict[str, str]: """Gets the ancestor tasks of the given task_ids as a dictionary of label -> taskid. Args: @@ -72,7 +71,7 @@ def get_ancestors( Returns: dict: A dict whose keys are task labels and values are task ids. """ - upstream_tasks: Dict[str, str] = {} + upstream_tasks: dict[str, str] = {} logging.info("A") if isinstance(task_ids, str): @@ -87,6 +86,7 @@ def get_ancestors( return copy.deepcopy(upstream_tasks) + @cache def get_taskcluster_client(service: str): options = {"rootUrl": FIREFOXCI_ROOT_URL} @@ -94,7 +94,9 @@ def get_taskcluster_client(service: str): @cache -def find_tasks(decision_index_path: str, include_deps: bool = False) -> list[dict[str, Any]]: +def find_tasks( + decision_index_path: str, include_deps: bool = False +) -> list[dict[str, Any]]: """Find tasks targeted by the Decision task pointed to by `decision_index_path`.""" queue = get_taskcluster_client("queue") index = get_taskcluster_client("index") @@ -130,6 +132,7 @@ def find_tasks(decision_index_path: str, include_deps: bool = False) -> list[dic logging.info("including deps") # TODO: remove hack import os + orig = os.environ["TASKCLUSTER_ROOT_URL"] os.environ["TASKCLUSTER_ROOT_URL"] = FIREFOXCI_ROOT_URL logging.info(f"fetching ancestors for {task['task_id']}")