forked from aws/deep-learning-containers
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbuildspec.py
More file actions
148 lines (118 loc) · 4.8 KB
/
Copy pathbuildspec.py
File metadata and controls
148 lines (118 loc) · 4.8 KB
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
"""
Copyright 2019-2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
Licensed under the Apache License, Version 2.0 (the "License"). You
may not use this file except in compliance with the License. A copy of
the License is located at
http://aws.amazon.com/apache2.0/
or in the "license" file accompanying this file. This file is
distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
ANY KIND, either express or implied. See the License for the specific
language governing permissions and limitations under the License.
"""
import os
import warnings
import ruamel.yaml
class Buildspec:
"""
The Buildspec class is responsible for parsing the buildspec file.
It is used to standardize the ruamel.yaml configurations, add
special constructors and load yaml files.
"""
def __init__(self):
self.yaml = ruamel.yaml.YAML()
self.yaml.allow_duplicate_keys = True
self.yaml.Constructor.add_constructor("!join", self.join)
self._buildspec = None
def load(self, path):
"""
This function loads the buildspec file and
populates the buildspec object.
Parameters:
path: str
Returns:
None
"""
# Check to see if buildspec file is a pointer
with open(path, "r") as bf:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
pointer_check = self.yaml.load(bf)
pointer = pointer_check.get("buildspec_pointer")
if pointer:
if os.getenv("BUILD_CONTEXT") != "PR":
raise RuntimeError(
f"Detected pointer in buildspec: {path} - this is only supported in PRs"
)
print(f"Buildspec {path} points to another buildspec file {pointer}")
path = os.path.join(os.path.dirname(path), pointer)
print(f"Inferring buildspec path to be {path}")
with open(path, "r") as buildspec_file:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
self._buildspec = self.yaml.load(buildspec_file)
self._buildspec = self.override(self._buildspec)
def override(self, yaml_object):
"""
This method overrides anchors in a scalar string with
values from the environment
"""
# If the yaml object is a PlainScalarString or ScalarFloat and an environment variable
# with the same name exists, return the environment variable otherwise,
# return the original yaml_object
scalar_types = (
ruamel.yaml.scalarstring.ScalarString,
ruamel.yaml.scalarfloat.ScalarFloat,
ruamel.yaml.scalarstring.PlainScalarString,
ruamel.yaml.scalarbool.ScalarBoolean,
)
if isinstance(yaml_object, ruamel.yaml.comments.CommentedMap):
for key in yaml_object:
yaml_object[key] = self.override(yaml_object[key])
elif isinstance(yaml_object, scalar_types):
if yaml_object.anchor is not None:
if yaml_object.anchor.value is not None:
yaml_object = os.environ.get(yaml_object.anchor.value, yaml_object)
# If the yaml object is not a PlainScalarString, does not have an anchor,
# or it's anchor does not have a value, return
# the original yaml object
return yaml_object
def join(self, loader, node):
"""
This method is used to perform string concatenation
in the yaml file. Specifying !join [x,y,z] should
result in the string xyz
Parameters:
loader: ruamel.yaml.constructor.RoundTripConstructor
node: ruamel.yaml.nodes.SequenceNode
Returns:
str
"""
seq = [self.override(scalar_string) for scalar_string in loader.construct_sequence(node)]
seq = "".join([str(scalar_string) for scalar_string in seq])
seq = ruamel.yaml.scalarstring.PlainScalarString(seq)
if node.anchor is not None:
seq.anchor.value = node.anchor
return seq
def get(self, name, default=None):
"""
Returns default if there's no such key in the buildspec.
Parameters:
name: str
default: str - default value to return
Returns:
object of if object is None - default
"""
try:
return self._buildspec[name]
except KeyError:
return default
def __getitem__(self, name):
"""
This method adds dictionary style access to an object of the
Buildspec class.
Parameters:
name: str
Returns:
object
"""
return self._buildspec[name]