Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
empty dictionaries, or None
* Dumps dictionary as JSON
Args:
metadata: Dictionary that needs to be formatted into an event.
Returns:
JSON-formatted file event.
"""
def visit(path, key, value):
if isinstance(value, (bytes, bytearray)):
value = str(value, encoding='UTF-8', errors='replace')
return key, value
remap1 = iterutils.remap(metadata, visit=visit)
remap2 = iterutils.remap(
remap1,
lambda p, k, v: v != '' and v != [] and v != {} and v is not None,
)
return json.dumps(remap2)
def pop(self, key: str) -> Any:
"""Delete and return value at key.
Args:
key (str): Key to pop.
Returns:
Any: Popped value.
"""
path, target = self.parse_key(key)
value = self.get(key)
remapped = iterutils.remap(self._config, lambda p, k,
v: False if p == path and k == target else True)
self._config = remapped
self.log.debug(f"popped config value {value} <- [{key}]")
return self.sync()
new_parent, new_items)
if paths_to_keep_tagged and isinstance(resp, dict):
updated_values = {}
for sub_key, value in resp.items():
if not isinstance(value, list):
continue
new_key = '{}@'.format(sub_key)
updated_values[new_key] = value
resp.update(updated_values)
try:
paths_to_keep_tagged.remove((path, key))
except KeyError:
pass
return resp
return iterutils.remap(data, visit=_visit, exit=_remap_exit)
def chain_argspec(func_list, provides, inner_name):
provided_sofar = set([inner_name]) # the inner function name is an extremely special case
optional_sofar = set()
required_sofar = set()
for f, p in zip(func_list, provides):
# middlewares can default the same parameter to different values;
# can't properly keep track of default values
fb = get_fb(f)
arg_names = fb.get_arg_names()
defaults_dict = fb.get_defaults_dict()
defaulted, undefaulted = iterutils.partition(arg_names, key=defaults_dict.__contains__)
optional_sofar.update(defaulted)
# keep track of defaults so that e.g. endpoint default param
# can pick up request injected/provided param
required_sofar |= set(undefaulted) - provided_sofar
provided_sofar.update(p)
return required_sofar, optional_sofar
new_parent, new_items)
if paths_to_keep_tagged and isinstance(resp, dict):
updated_values = {}
for sub_key, value in resp.items():
if not isinstance(value, list):
continue
new_key = '{}@'.format(sub_key)
updated_values[new_key] = value
resp.update(updated_values)
try:
paths_to_keep_tagged.remove((path, key))
except KeyError:
pass
return resp
return iterutils.remap(data, visit=visit, exit=remap_exit)
locale_parent = locale_parent[path_key]
else:
locale_parent = locale_parent[path_key]
if base_key in parent:
locale_parent[base_key] = value
else:
locale_parent['{}@'.format(base_key)] = value
if key in locale_parent:
locale_parent.pop(key, None)
return False
return key, value
parsed = iterutils.remap(parsed, visit=visit)
return (yaml.dump(
parsed, Dumper=PlainTextYamlDumper,
allow_unicode=True, default_flow_style=False).strip() if parsed else '',
locale_extra)
if not match:
return key, value
base_key = match.group(1)
locale_from_key = match.group(2)
if locale_from_key == locale:
# If there is a key without the trailing @ then override it.
parent = parsed
for path_key in path:
parent = parent[path_key]
if base_key in parent:
return base_key, value
return '{}@'.format(base_key), value
return False
parsed = iterutils.remap(parsed, visit=visit)
# If there are pre-existing fields, use them as a base for the locale
# specific values.
result = base or {}
_update_deep(result, parsed)
return result