Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_set_new_list():
dict = {
"a": [
]
}
dpath.util.new(dict, '/a/1', 1)
assert(dict['a'][1] == 1)
assert(dict['a'][0] == None)
dpath.util.new(dict, ['a', '1'], 1)
assert(dict['a'][1] == 1)
assert(dict['a'][0] == None)
def parse_statement(override_statement):
key, value = override_statement.split('=')
obj = {}
dpath.util.new(obj, key, value, separator='.')
return obj
regex = re.compile(r"^\[.*|^\{.*") # Regex used for detecting dicts/arrays in pack params
for key, value in pack_params:
if re.match(regex, value):
try:
value = ast.literal_eval(value)
except Exception as e:
raise AttributeError(Texts.CANT_PARSE_VALUE.format(value=value, error=e))
# Handle boolean params
elif value in {"true", "false"}:
value = str(_parse_yaml_boolean(value))
if key == WORK_CNT_PARAM:
workersCount = value
if key == P_SERV_CNT_PARAM:
pServersCount = value
dutil.new(v, key, value, '.')
# setting sum of replicas involved in multinode training if both pServersCount and workersCount are present in
# the pack or given in the cli
if (WORK_CNT_PARAM in v or workersCount) and (P_SERV_CNT_PARAM in v or pServersCount):
number_of_replicas = int(v.get(WORK_CNT_PARAM)) if not workersCount else int(workersCount)
number_of_replicas += int(v.get(P_SERV_CNT_PARAM)) if not pServersCount else int(pServersCount)
v[POD_COUNT_PARAM] = number_of_replicas
elif (WORK_CNT_PARAM in v or workersCount) and (POD_COUNT_PARAM not in v):
number_of_replicas = int(v.get(WORK_CNT_PARAM)) if not workersCount else int(workersCount)
v[POD_COUNT_PARAM] = number_of_replicas + 1
env_variables = env_variables if env_variables else []
parsed_envs = []
for variable in env_variables:
key, value = variable.split("=")
one_env_map = {"name": key, "value": value}
def prepareGroups(parameters):
'''group keyword by package'''
groups = {}
for keyword, value in parameters['language'].iteritems():
group, module = tuple(value['package'].split(':'))
if not Utilities.isDefined(groups, '/' + group + '/' + module):
dpath.util.new(groups, '/' + group + '/' + module, [])
groups[group][module].append(keyword)
return groups
def set(self, manifest, x, path, value):
qualified = '/%d/%s' % (x, path)
logging.debug("SET %s = %s" % (qualified, value))
dpath.util.new(manifest, qualified, value)
from django.template import loader, Context
from dpath import util as dpath_util
from dict2xml import dict2xml
def geopoint_xpaths(username, id_string):
d = DataDictionary.objects.get(
user__username__iexact=username, id_string__exact=id_string)
return [e.get_abbreviated_xpath()
for e in d.get_survey_elements()
if e.bind.get(u'type') == u'geopoint']
value = request.GET.get('coordinates')
xpaths = geopoint_xpaths(username, id_string)
xml_dict = {}
for path in xpaths:
dpath_util.new(xml_dict, path, value)
context = {'username': username,
'id_string': id_string,
'xml_content': dict2xml(xml_dict)}
instance_xml = loader.get_template("instance_add.xml")\
.render(Context(context))
url = settings.ENKETO_API_INSTANCE_IFRAME_URL
return_url = reverse('thank_you_submission',
kwargs={"username": username, "id_string": id_string})
if settings.DEBUG:
openrosa_url = "https://dev.formhub.org/{}".format(username)
else:
openrosa_url = request.build_absolute_uri("/{}".format(username))
payload = {'return_url': return_url,
'form_id': id_string,
from django.template import loader, Context
from dpath import util as dpath_util
from dict2xml import dict2xml
def geopoint_xpaths(username, id_string):
d = DataDictionary.objects.get(
user__username__iexact=username, id_string__iexact=id_string)
return [e.get_abbreviated_xpath()
for e in d.get_survey_elements()
if e.bind.get(u'type') == u'geopoint']
value = request.GET.get('coordinates')
xpaths = geopoint_xpaths(username, id_string)
xml_dict = {}
for path in xpaths:
dpath_util.new(xml_dict, path, value)
context = {'username': username,
'id_string': id_string,
'xml_content': dict2xml(xml_dict)}
instance_xml = loader.get_template("instance_add.xml")\
.render(Context(context))
url = settings.ENKETO_API_INSTANCE_IFRAME_URL
return_url = reverse('thank_you_submission',
kwargs={"username": username, "id_string": id_string})
if settings.DEBUG:
openrosa_url = "https://dev.formhub.org/{}".format(username)
else:
openrosa_url = request.build_absolute_uri(settings.APP_ROOT + "{}".format(username))
payload = {'return_url': return_url,
'form_id': id_string,
for x in range(len(manifest)):
kind = dpath.util.get(manifest, "/%d/kind" % x)
name = dpath.util.get(manifest, "/%d/metadata/name" % x)
if kind in kinds_to_delete:
# print("Skipping %s %s" % (kind, name))
continue
print("Adding namespace %s to %s %s" % (namespace, kind, name))
dpath.util.new(manifest, "/%d/metadata/namespace" % x, namespace)
if kind == 'Deployment':
print("Setting environment for %s %s" % (kind, name))
path = "/%d/spec/template/spec/containers/0/env" % x
dpath.util.new(manifest, path, [
{
'name': 'AMBASSADOR_NAMESPACE',
'value': namespace
},
{
'name': 'AMBASSADOR_ID',
'value': ambassador_id
}
])
keep.append(manifest[x])
yaml.safe_dump_all(keep, open(output_yaml_path, "w"), default_flow_style=False)
"ServiceAccount": True,
"ClusterRoleBinding": True,
}
keep = []
for x in range(len(manifest)):
kind = dpath.util.get(manifest, "/%d/kind" % x)
name = dpath.util.get(manifest, "/%d/metadata/name" % x)
if kind in kinds_to_delete:
# print("Skipping %s %s" % (kind, name))
continue
print("Adding namespace %s to %s %s" % (namespace, kind, name))
dpath.util.new(manifest, "/%d/metadata/namespace" % x, namespace)
if kind == 'Deployment':
print("Setting environment for %s %s" % (kind, name))
path = "/%d/spec/template/spec/containers/0/env" % x
dpath.util.new(manifest, path, [
{
'name': 'AMBASSADOR_NAMESPACE',
'value': namespace
},
{
'name': 'AMBASSADOR_ID',
'value': ambassador_id
}
])
keep.append(manifest[x])