Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def save_paths(cfg, base=None):
replace_keys = {}
for key, val in cfg.items():
kbase = key if base is None else base+'.'+key
if isinstance(val, dict):
if '__path__' in val:
val_path = Path(os.path.join(dirname, os.path.basename(val['__path__']())), mode='fc')
if not overwrite and os.path.isfile(val_path()):
raise ValueError('Refusing to overwrite existing file: '+val_path)
action = _find_action(self, kbase)
if isinstance(action, ActionParser):
replace_keys[key] = val_path
action._parser.save(val, val_path(), branch=action.dest, **save_kwargs)
elif isinstance(action, (ActionJsonSchema, ActionJsonnet)):
replace_keys[key] = val_path
val_out = strip_meta(val)
if format == 'json_indented' or isinstance(action, ActionJsonnet):
val_str = json.dumps(val_out, indent=2, sort_keys=True)
elif format == 'yaml':
val_str = yaml.dump(val_out, default_flow_style=False, allow_unicode=True)
elif format == 'json':
val_str = json.dumps(val_out, sort_keys=True)
with open(val_path(), 'w') as f:
f.write(val_str)
else:
save_paths(val, kbase)
else:
save_paths(val, kbase)
for key, val in replace_keys.items():
cfg[key] = os.path.basename(val())
def __call__(self, *args, **kwargs):
"""Parses an argument as jsonnet using ext_vars if defined.
Raises:
TypeError: If the argument is not valid.
"""
if len(args) == 0:
kwargs['_ext_vars'] = self._ext_vars
kwargs['_validator'] = self._validator
if 'help' in kwargs and '%s' in kwargs['help'] and self._validator is not None:
kwargs['help'] = kwargs['help'] % json.dumps(self._validator.schema, indent=2, sort_keys=True)
return ActionJsonnet(**kwargs)
setattr(args[1], self.dest, self._check_type(args[2], cfg=args[1]))
replace_keys = {}
for key, val in cfg.items():
kbase = key if base is None else base+'.'+key
if isinstance(val, dict):
if '__path__' in val:
val_path = Path(os.path.join(dirname, os.path.basename(val['__path__']())), mode='fc')
if not overwrite and os.path.isfile(val_path()):
raise ValueError('Refusing to overwrite existing file: '+val_path)
action = _find_action(self, kbase)
if isinstance(action, ActionParser):
replace_keys[key] = val_path
action._parser.save(val, val_path(), branch=action.dest, **save_kwargs)
elif isinstance(action, (ActionJsonSchema, ActionJsonnet)):
replace_keys[key] = val_path
val_out = strip_meta(val)
if format == 'json_indented' or isinstance(action, ActionJsonnet):
val_str = json.dumps(val_out, indent=2, sort_keys=True)
elif format == 'yaml':
val_str = yaml.dump(val_out, default_flow_style=False, allow_unicode=True)
elif format == 'json':
val_str = json.dumps(val_out, sort_keys=True)
with open(val_path(), 'w') as f:
f.write(val_str)
else:
save_paths(val, kbase)
else:
save_paths(val, kbase)
for key, val in replace_keys.items():
cfg[key] = os.path.basename(val())