1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import re
from taskgraph.util.taskcluster import get_artifact_url
from taskgraph.util.time import json_time_from_now
TASK_REFERENCE_PATTERN = re.compile("<([^>]+)>")
ARTIFACT_REFERENCE_PATTERN = re.compile("<([^/]+)/([^>]+)>")
def _recurse(val, param_fns):
def recurse(val):
if isinstance(val, list):
return [recurse(v) for v in val]
elif isinstance(val, dict):
if len(val) == 1:
for param_key, param_fn in param_fns.items():
if set(val.keys()) == {param_key}:
if isinstance(val[param_key], dict):
# handle `{"task-reference": {"<foo>": "bar"}}`
return {
param_fn(key): recurse(v)
for key, v in val[param_key].items()
}
return param_fn(val[param_key])
return {k: recurse(v) for k, v in val.items()}
else:
return val
return recurse(val)
def resolve_timestamps(now, task_def):
"""Resolve all instances of `{'relative-datestamp': '..'}` in the given task definition"""
return _recurse(
task_def,
{
"relative-datestamp": lambda v: json_time_from_now(v, now),
},
)
def resolve_task_references(label, task_def, task_id, decision_task_id, dependencies):
"""Resolve all instances of ``{'task-reference': '..<..>..'} ``
and ``{'artifact-reference`: '..<dependency/artifact/path>..'}``
in the given task definition, using the given dependencies.
"""
def task_reference(val):
def repl(match):
key = match.group(1)
if key == "self":
return task_id
elif key == "decision":
return decision_task_id
try:
return dependencies[key]
except KeyError:
# handle escaping '<'
if key == "<":
return key
raise KeyError(f"task '{label}' has no dependency named '{key}'")
return TASK_REFERENCE_PATTERN.sub(repl, val)
def artifact_reference(val):
def repl(match):
dependency, artifact_name = match.group(1, 2)
if dependency == "self":
raise KeyError(f"task '{label}' can't reference artifacts of self")
elif dependency == "decision":
task_id = decision_task_id
else:
try:
task_id = dependencies[dependency]
except KeyError:
raise KeyError(
f"task '{label}' has no dependency named '{dependency}'"
)
use_proxy = False
if not artifact_name.startswith("public/"):
use_proxy = True
return get_artifact_url(task_id, artifact_name, use_proxy=use_proxy)
return ARTIFACT_REFERENCE_PATTERN.sub(repl, val)
return _recurse(
task_def,
{
"task-reference": task_reference,
"artifact-reference": artifact_reference,
},
)
|
