_jsonnet.py

"""Actions to support jsonnet."""

from typing import Any, Dict, Optional, Union

from ._actions import _is_action_value_list
from ._common import Action, parser_context
from ._jsonschema import ActionJsonSchema
from ._loaders_dumpers import get_loader_exceptions, load_value
from ._optionals import (
    _get_config_read_mode,
    get_jsonschema_exceptions,
    import_jsonnet,
    import_jsonschema,
    pyyaml_available,
)
from ._subcommands import find_action
from ._typehints import ActionTypeHint
from ._util import NoneType, Path, argument_error

__all__ = ["ActionJsonnet"]


class ActionJsonnet(Action):
    """Action to parse a Jsonnet, optionally validating against a JSON Schema."""

    def __init__(
        self,
        ext_vars: Optional[str] = None,
        schema: Optional[Union[str, dict]] = None,
        **kwargs,
    ):
        """Initializer for ActionJsonnet instance.

        Args:
            ext_vars: Key where to find the external variables required to parse the Jsonnet.
            schema: Schema to validate values against.

        Raises:
            ValueError: If a parameter is invalid.
            jsonschema.exceptions.SchemaError: If the schema is invalid.
        """
        if "_validator" not in kwargs:
            import_jsonnet("ActionJsonnet")
            if not isinstance(ext_vars, (str, NoneType)):
                raise ValueError("ext_vars has to be either None or a string.")
            self._ext_vars = ext_vars
            if schema is not None:
                jsonvalidator = import_jsonschema("ActionJsonnet")[1]
                if isinstance(schema, str):
                    mode = "yaml" if pyyaml_available else "json"
                    with parser_context(load_value_mode=mode):
                        try:
                            schema = load_value(schema)
                        except get_loader_exceptions(mode) as ex:
                            raise ValueError(f"Problems parsing schema: {ex}") from ex
                jsonvalidator.check_schema(schema)
                self._validator = ActionJsonSchema._extend_jsonvalidator_with_default(jsonvalidator)(schema)
            else:
                self._validator = None
        else:
            self._ext_vars = kwargs.pop("_ext_vars")
            self._validator = kwargs.pop("_validator")
            super().__init__(**kwargs)

    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:
                import json

                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]))
        return None

    def _check_type(self, value, cfg):
        islist = _is_action_value_list(self)
        ext_vars = {}
        if cfg:
            ext_vars = cfg.get(self._ext_vars, {})
        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) + get_jsonschema_exceptions() + get_loader_exceptions() as ex:
                elem = "" if not islist else " element " + str(num + 1)
                raise TypeError(f'Parser key "{self.dest}"{elem}: {ex}') from ex
        return value if islist else value[0]

    @staticmethod
    def _check_ext_vars_action(parser, action):
        if isinstance(action, ActionJsonnet) and action._ext_vars:
            ext_vars_action = find_action(parser, action._ext_vars)
            if not ext_vars_action:
                raise ValueError(f"No argument found for ext_vars='{action._ext_vars}'")
            ext_vars_type = isinstance(ext_vars_action, ActionTypeHint) and ext_vars_action._typehint
            if ext_vars_type not in {dict, Dict}:
                raise ValueError(
                    f"Type for ext_vars='{action._ext_vars}' argument must be dict, given: {ext_vars_type}"
                )
            if ext_vars_action.default is None:
                ext_vars_action.default = {}
            if not isinstance(ext_vars_action.default, dict):
                raise ValueError(
                    f"Default value for the ext_vars='{action._ext_vars}' argument "
                    f"must be dict or None, given: {ext_vars_action.default}"
                )
            ext_vars_action.jsonnet_ext_vars = True

    @staticmethod
    def split_ext_vars(ext_vars: Optional[dict[str, Any]]) -> 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 = {}
        import json

        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

    def parse(
        self,
        jsonnet: Union[str, Path],
        ext_vars: Optional[dict[str, Any]] = None,
        with_meta: bool = False,
    ) -> dict:
        """Method that can be used to parse Jsonnet independent from an :class:`.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.read_text()
        try:
            with parser_context(load_value_mode="yaml" if pyyaml_available else "json"):
                values = load_value(_jsonnet.evaluate_snippet(fname, snippet, ext_vars=ext_vars, ext_codes=ext_codes))
        except RuntimeError as ex:
            raise argument_error(f"Problems evaluating Jsonnet '{fname}': {ex}") from ex
        if self._validator is not None:
            self._validator.validate(values)
        if with_meta:
            if fpath is not None:
                values["__path__"] = fpath
            values["__orig__"] = snippet
        return values