4
4
5
5
import copy
6
6
from functools import cache
7
- from typing import Any , Dict , List , Union
7
+ from typing import Any
8
+
8
9
import requests
9
10
import taskcluster
10
- from taskgraph .util .taskcluster import get_task_url , get_session
11
11
from taskgraph .util .memoize import memoize
12
+ from taskgraph .util .taskcluster import get_session , get_task_url
12
13
13
14
FIREFOXCI_ROOT_URL = "https://firefox-ci-tc.services.mozilla.com"
14
15
STAGING_ROOT_URL = "https://stage.taskcluster.nonprod.cloudops.mozgcp.net"
15
16
17
+
16
18
def _do_request (url , method = None , ** kwargs ):
17
19
if method is None :
18
20
method = "post" if kwargs else "get"
@@ -50,9 +52,7 @@ def _get_deps(task_ids, use_proxy):
50
52
return upstream_tasks
51
53
52
54
53
- def get_ancestors (
54
- task_ids : Union [List [str ], str ], use_proxy : bool = False
55
- ) -> Dict [str , str ]:
55
+ def get_ancestors (task_ids : list [str ] | str , use_proxy : bool = False ) -> dict [str , str ]:
56
56
"""Gets the ancestor tasks of the given task_ids as a dictionary of label -> taskid.
57
57
58
58
Args:
@@ -62,7 +62,7 @@ def get_ancestors(
62
62
Returns:
63
63
dict: A dict whose keys are task labels and values are task ids.
64
64
"""
65
- upstream_tasks : Dict [str , str ] = {}
65
+ upstream_tasks : dict [str , str ] = {}
66
66
67
67
if isinstance (task_ids , str ):
68
68
task_ids = [task_ids ]
@@ -74,14 +74,17 @@ def get_ancestors(
74
74
75
75
return copy .deepcopy (upstream_tasks )
76
76
77
+
77
78
@cache
78
79
def get_taskcluster_client (service : str ):
79
80
options = {"rootUrl" : FIREFOXCI_ROOT_URL }
80
81
return getattr (taskcluster , service .capitalize ())(options )
81
82
82
83
83
84
@cache
84
- def find_tasks (decision_index_path : str , include_deps : bool = False ) -> list [dict [str , Any ]]:
85
+ def find_tasks (
86
+ decision_index_path : str , include_deps : bool = False
87
+ ) -> list [dict [str , Any ]]:
85
88
"""Find tasks targeted by the Decision task pointed to by `decision_index_path`."""
86
89
queue = get_taskcluster_client ("queue" )
87
90
index = get_taskcluster_client ("index" )
@@ -115,6 +118,7 @@ def find_tasks(decision_index_path: str, include_deps: bool = False) -> list[dic
115
118
if include_deps :
116
119
# TODO: remove hack
117
120
import os
121
+
118
122
orig = os .environ ["TASKCLUSTER_ROOT_URL" ]
119
123
os .environ ["TASKCLUSTER_ROOT_URL" ] = FIREFOXCI_ROOT_URL
120
124
for label , task_id in get_ancestors (task ["task_id" ]).items ():
0 commit comments