44
55import copy
66from functools import cache
7- from typing import Any , Dict , List , Union
7+ from typing import Any
88
99import requests
1010import taskcluster
11- from taskgraph .util .taskcluster import logging , get_task_url , get_session
1211from taskgraph .util .memoize import memoize
12+ from taskgraph .util .taskcluster import get_session , get_task_url , logging
1313
1414FIREFOXCI_ROOT_URL = "https://firefox-ci-tc.services.mozilla.com"
1515STAGING_ROOT_URL = "https://stage.taskcluster.nonprod.cloudops.mozgcp.net"
1616
17+
1718def _do_request (url , method = None , ** kwargs ):
1819 if method is None :
1920 method = "post" if kwargs else "get"
@@ -60,9 +61,7 @@ def _get_deps(task_ids, use_proxy):
6061 return upstream_tasks
6162
6263
63- def get_ancestors (
64- task_ids : Union [List [str ], str ], use_proxy : bool = False
65- ) -> Dict [str , str ]:
64+ def get_ancestors (task_ids : list [str ] | str , use_proxy : bool = False ) -> dict [str , str ]:
6665 """Gets the ancestor tasks of the given task_ids as a dictionary of label -> taskid.
6766
6867 Args:
@@ -72,7 +71,7 @@ def get_ancestors(
7271 Returns:
7372 dict: A dict whose keys are task labels and values are task ids.
7473 """
75- upstream_tasks : Dict [str , str ] = {}
74+ upstream_tasks : dict [str , str ] = {}
7675 logging .info ("A" )
7776
7877 if isinstance (task_ids , str ):
@@ -87,14 +86,17 @@ def get_ancestors(
8786
8887 return copy .deepcopy (upstream_tasks )
8988
89+
9090@cache
9191def get_taskcluster_client (service : str ):
9292 options = {"rootUrl" : FIREFOXCI_ROOT_URL }
9393 return getattr (taskcluster , service .capitalize ())(options )
9494
9595
9696@cache
97- def find_tasks (decision_index_path : str , include_deps : bool = False ) -> list [dict [str , Any ]]:
97+ def find_tasks (
98+ decision_index_path : str , include_deps : bool = False
99+ ) -> list [dict [str , Any ]]:
98100 """Find tasks targeted by the Decision task pointed to by `decision_index_path`."""
99101 queue = get_taskcluster_client ("queue" )
100102 index = get_taskcluster_client ("index" )
@@ -130,6 +132,7 @@ def find_tasks(decision_index_path: str, include_deps: bool = False) -> list[dic
130132 logging .info ("including deps" )
131133 # TODO: remove hack
132134 import os
135+
133136 orig = os .environ ["TASKCLUSTER_ROOT_URL" ]
134137 os .environ ["TASKCLUSTER_ROOT_URL" ] = FIREFOXCI_ROOT_URL
135138 logging .info (f"fetching ancestors for { task ['task_id' ]} " )
0 commit comments