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_values_passes_through(searchfunc):
searchfunc.return_value = []
def y():
pass
dpath.util.values({}, '/a/b', ':', y, False)
searchfunc.assert_called_with({}, '/a/b', dirs=False, yielded=True, separator=':', afilter=y)
dpath.util.values({}, ['a', 'b'], ':', y, False)
searchfunc.assert_called_with({}, ['a', 'b'], dirs=False, yielded=True, separator=':', afilter=y)
"b": {
"c": {
"d": 0,
"e": 1,
"f": 2
}
}
}
}
ret = dpath.util.values(ehash, '/a/b/c/*')
assert(isinstance(ret, list))
assert(0 in ret)
assert(1 in ret)
assert(2 in ret)
ret = dpath.util.values(ehash, ['a', 'b', 'c', '*'])
assert(isinstance(ret, list))
assert(0 in ret)
assert(1 in ret)
assert(2 in ret)
def fetch_field(self, doc, expr):
chunks = expr.split("|", 2)
path = chunks[0]
operation = chunks[1] if len(chunks) > 1 else ""
if not path:
return ""
try:
res = dpath.util.values(doc, path, separator='.')
except KeyError:
res = []
if operation:
res = self.apply_operation(res, operation)
if isinstance(res, (list, tuple)):
return ", ".join(map(str, res))
else:
return res
"b": {
"3": 2,
"43": 30,
"c": [],
"d": ['red', 'buggy', 'bumpers'],
}
}
}
print(dpath.util.get(x, 'a/b/d')) # ['red', 'buggy', 'bumpers']
print(dpath.util.get(x, 'a/b/d/0')) # red
print(dpath.util.get(x, 'a/b/d/1')) # buggy
print(dpath.util.get(x, 'a/b/43')) # 30
print()
print(dpath.util.values(x, "**/43")) # [30]
print(dpath.util.values(x, "**/d/1")) # [buggy]
print()
print(dpath.util.search(x, "**/43")) # {'a': {'b': {'43': 30}}}
print(list(dpath.util.search(x, "**/43", yielded=True))) # [('a/b/43', 30)]
print()
def afilter(x):
return str(x).isdecimal()
result = dpath.util.search(x, '**', afilter=afilter)
print(result) # {'a': {'b': {'3': 2, '43': 30}}}
# Фильтрация через лябмды:
def paths(dictionary, dictionary_path):
return dpath.util.values(dictionary, dictionary_path)
The lambda function auditor can define a list of `policy_keys`. Each
item in this list is the dpath to one of the resource policies.
The `policy_keys` defaults to ['Policy'] unless overriden by a subclass.
Returns:
list of Policy objects
"""
import dpath.util
from dpath.exceptions import PathNotFound
from policyuniverse.policy import Policy
policies = list()
for key in policy_keys:
try:
policy = dpath.util.values(item.config, key, separator='$')
if isinstance(policy, list):
for p in policy:
if not p:
continue
if isinstance(p, list):
policies.extend([Policy(pp) for pp in p])
else:
policies.append(Policy(p))
else:
policies.append(Policy(policy))
except PathNotFound:
continue
return policies
@Utilities.cache_in_disk
def prepareCommandLineArguments(parameters):
# remember the available choices for outputs
Parameters.command_line_flags['globals:output']['choices'] = parameters['manifesto']['Outputs'].keys()
Parameters.command_line_flags['globals:input']['choices'] = parameters['manifesto']['Inputs'].keys()
Parameters.command_line_flags['globals:setEnvironment']['choices'] = sum([ x.keys() for x in dpath.util.values(parameters,'manifesto/*/*/environments')],[])
# create a subset of all the parameters
subset = dict((x, parameters[x])
for x in ['Information', 'Transformers', 'Inputs', 'Outputs', 'globals', 'developer'])
# create argparse list parameters
flags, arguments = generateArgparseArguments(subset, parameters['command_line_flags'])
# get all file formats
file_formats = []
file_package_name = []
for key, value in parameters['manifesto']['Inputs'].iteritems():
# @TODO Add support to multiple file extensions per format
# the file extension
def get_current_values(self, service_info):
if '*' in self.path:
return dpath.util.values(service_info, self.path)
try:
return dpath.util.get(service_info, self.path)
except KeyError:
return []
def postCommandLineParser(parameters):
# Version
if parameters['globals']['version']:
import pkg_resources
print('The Robotics Language version: ' + pkg_resources.get_distribution('RoboticsLanguage').version)
# Deploy path
if parameters['developer']['showDeployPath']:
print(parameters['globals']['deploy'])
# environments
if parameters['globals']['setEnvironment'] != '':
for environment in dpath.util.values(parameters,'manifesto/*/*/environments'):
if parameters['globals']['setEnvironment'] in environment.keys():
parameters_filename = os.path.expanduser('~') + '/.rol/parameters.yaml'
try:
if os.path.exists(parameters_filename):
with open(parameters_filename, 'r') as file:
# read current environment
current_environment = yaml.safe_load(file)
# update environment with new definitions
new_environment = Utilities.mergeDictionaries(environment[parameters['globals']['setEnvironment']], current_environment)
# backup old parameters
old_parameters_filename = Utilities.getNonexistantPath(parameters_filename)
shutil.move(parameters_filename, old_parameters_filename)