"""Actions to support jsonnet."""
import json
import yaml
from typing import Optional, Union, Dict, Tuple, Any
from argparse import Action, Namespace
from .actions import _is_action_value_list
from .jsonschema import ActionJsonSchema
from .optionals import (
import_jsonschema,
import_jsonnet,
get_config_read_mode,
jsonschemaValidationError,
)
from .util import (
dict_to_namespace,
namespace_to_dict,
yamlParserError,
yamlScannerError,
ParserError,
Path,
_check_unknown_kwargs,
_get_key_value,
)
__all__ = [
'ActionJsonnetExtVars',
'ActionJsonnet',
]
[docs]class ActionJsonnetExtVars(ActionJsonSchema):
"""Action to add argument to provide ext_vars for jsonnet parsing."""
[docs] def __init__(self, **kwargs):
"""Initializer for ActionJsonnetExtVars instance."""
super().__init__(schema={'type': 'object'}, with_meta=False)
[docs] def __call__(self, *args, **kwargs):
if len(args) == 0 and 'default' not in kwargs:
kwargs['default'] = {}
return super().__call__(*args, **kwargs)
[docs]class ActionJsonnet(Action):
"""Action to parse a jsonnet, optionally validating against a jsonschema."""
[docs] def __init__(self, **kwargs):
"""Initializer for ActionJsonnet instance.
Args:
ext_vars (str or None): Key where to find the external variables required to parse the jsonnet.
schema (str or object or None): Schema to validate values against. Keyword argument required even if schema=None.
Raises:
ValueError: If a parameter is invalid.
jsonschema.exceptions.SchemaError: If the schema is invalid.
"""
if 'ext_vars' in kwargs or 'schema' in kwargs:
import_jsonnet('ActionJsonnet')
_check_unknown_kwargs(kwargs, {'schema', 'ext_vars'})
if 'ext_vars' in kwargs and not isinstance(kwargs['ext_vars'], (str, type(None))):
raise ValueError('ext_vars has to be either None or a string.')
self._ext_vars = kwargs.get('ext_vars', None)
schema = kwargs.get('schema', None)
if schema is not None:
jsonvalidator = import_jsonschema('ActionJsonnet')[1]
if isinstance(schema, str):
try:
schema = yaml.safe_load(schema)
except (yamlParserError, yamlScannerError) as ex:
raise ValueError('Problems parsing schema :: '+str(ex)) from ex
jsonvalidator.check_schema(schema)
self._validator = ActionJsonSchema._extend_jsonvalidator_with_default(jsonvalidator)(schema)
else:
self._validator = None
elif '_ext_vars' not in kwargs or '_validator' not in kwargs:
raise ValueError('Expected ext_vars and/or schema keyword arguments.')
else:
self._ext_vars = kwargs.pop('_ext_vars')
self._validator = kwargs.pop('_validator')
super().__init__(**kwargs)
[docs] 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, sort_keys=True)
return ActionJsonnet(**kwargs)
setattr(args[1], self.dest, self._check_type(args[2], cfg=args[1]))
def _check_type(self, value, cfg):
islist = _is_action_value_list(self)
ext_vars = {}
if self._ext_vars is not None:
try:
ext_vars = _get_key_value(cfg, self._ext_vars)
except (KeyError, AttributeError):
pass
if not islist:
value = [value]
for num, val in enumerate(value):
try:
if isinstance(val, str):
val = self.parse(val, ext_vars=ext_vars, with_meta=True)
elif self._validator is not None:
self._validator.validate(val)
value[num] = val
except (TypeError, RuntimeError, yamlParserError, yamlScannerError, jsonschemaValidationError) as ex:
elem = '' if not islist else ' element '+str(num+1)
raise TypeError('Parser key "'+self.dest+'"'+elem+': '+str(ex)) from ex
return value if islist else value[0]
[docs] @staticmethod
def split_ext_vars(
ext_vars: Optional[Union[Dict[str, Any], Namespace]],
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""Splits an ext_vars dict into the ext_codes and ext_vars required by jsonnet.
Args:
ext_vars: External variables. Values can be strings or any other basic type.
"""
if ext_vars is None:
ext_vars = {}
elif isinstance(ext_vars, Namespace):
ext_vars = namespace_to_dict(ext_vars)
ext_codes = {k: json.dumps(v) for k, v in ext_vars.items() if not isinstance(v, str)}
ext_vars = {k: v for k, v in ext_vars.items() if isinstance(v, str)}
return ext_vars, ext_codes
[docs] def parse(
self,
jsonnet: Union[str, Path],
ext_vars: Union[Dict[str, Any], Namespace] = None,
with_meta: bool = False,
) -> Namespace:
"""Method that can be used to parse jsonnet independent from an ArgumentParser.
Args:
jsonnet: Either a path to a jsonnet file or the jsonnet content.
ext_vars: External variables. Values can be strings or any other basic type.
with_meta: Whether to include metadata in config object.
Returns:
The parsed jsonnet object.
Raises:
TypeError: If the input is neither a path to an existent file nor a jsonnet.
"""
_jsonnet = import_jsonnet('ActionJsonnet')
ext_vars, ext_codes = self.split_ext_vars(ext_vars)
fpath = None
fname = 'snippet'
snippet = jsonnet
try:
fpath = Path(jsonnet, mode=get_config_read_mode())
except TypeError:
pass
else:
fname = jsonnet(absolute=False) if isinstance(jsonnet, Path) else jsonnet
snippet = fpath.get_content()
try:
values = yaml.safe_load(_jsonnet.evaluate_snippet(fname, snippet, ext_vars=ext_vars, ext_codes=ext_codes))
except RuntimeError as ex:
raise ParserError('Problems evaluating jsonnet "'+fname+'" :: '+str(ex)) from ex
if self._validator is not None:
self._validator.validate(values)
if with_meta and isinstance(values, dict) and fpath is not None:
values['__path__'] = fpath
return dict_to_namespace(values)