python: modeling: added JSON/YAML parsing

diff --git a/python/knot_resolver/utils/modeling/__init__.py b/python/knot_resolver/utils/modeling/__init__.py
new file mode 100644 (file)
index 0000000..e69de29

diff --git a/python/knot_resolver/utils/modeling/parsing.py b/python/knot_resolver/utils/modeling/parsing.py
new file mode 100644 (file)
index 0000000..d57000c
--- /dev/null
+++ b/python/knot_resolver/utils/modeling/parsing.py
@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import json
+from enum import Enum, auto
+from typing import TYPE_CHECKING, Any
+
+import yaml
+from yaml.constructor import ConstructorError
+
+from knot_resolver.utils.modeling.errors import DataParsingError
+
+if TYPE_CHECKING:
+    from yaml.nodes import MappingNode
+
+
+def _json_raise_duplicates(pairs: list[tuple[Any, Any]]) -> dict[Any, Any]:
+    """
+    JSON hook used in 'json.loads()' that detects duplicate keys in the parsed data.
+
+    The code for this hook was highly inspired by: https://stackoverflow.com/q/14902299/12858520
+    """
+    mapping: dict[Any, Any] = {}
+    for key, value in pairs:
+        if key in mapping:
+            msg = f"duplicate key detected: {key}"
+            raise DataParsingError(msg)
+        mapping[key] = value
+    return mapping
+
+
+class _YAMLRaiseDuplicatesLoader(yaml.SafeLoader):
+    """
+    YAML loader used in 'yaml.loads()' that detects duplicate keys in the parsed data.
+
+    The code for this loader was highly inspired by: https://gist.github.com/pypt/94d747fe5180851196eb
+    The loader extends yaml.SafeLoader, so it should be safe, even though the linter reports unsafe-yaml-load (S506).
+    More about safe loader: https://python.land/data-processing/python-yaml#PyYAML_safe_load_vs_load
+    """
+
+    def construct_mapping(self, node: MappingNode, deep: bool = False) -> dict[Any, Any]:
+        mapping: dict[Any, Any] = {}
+        for key_node, value_node in node.value:
+            key = self.construct_object(key_node, deep=deep)
+            # we need to check, that the key object can be used in a hash table
+            try:
+                _ = hash(key)
+            except TypeError as exc:
+                msg = f"while constructing a mapping {node.start_mark}"
+                f"found unacceptable key ({exc}) {key_node.start_mark}"
+                raise ConstructorError(msg) from exc
+
+            # check for duplicate keys
+            if key in mapping:
+                msg = f"duplicate key detected: {key_node.start_mark}"
+                raise DataParsingError(msg)
+            value = self.construct_object(value_node, deep=deep)
+            mapping[key] = value
+        return mapping
+
+
+class DataFormat(Enum):
+    YAML = auto()
+    JSON = auto()
+
+    def loads(self, text: str) -> dict[Any, Any]:
+        """Load data from string in data format and return the data in dictionary."""
+        if self is DataFormat.YAML:
+            return yaml.load(text, Loader=_YAMLRaiseDuplicatesLoader)  # noqa: S506
+        if self is DataFormat.JSON:
+            return json.loads(text, object_pairs_hook=_json_raise_duplicates)
+        msg = f"parsing data from '{self}' format is not implemented"
+        raise NotImplementedError(msg)
+
+    def dumps(self, data: dict[Any, Any], indent: int | None = None) -> str:
+        """Dump dictionary data to string in required data format."""
+        if self is DataFormat.YAML:
+            return yaml.safe_dump(data, indent=indent)
+        if self is DataFormat.JSON:
+            return json.dumps(data, indent=indent)
+        msg = f"exporting data to '{self}' format is not implemented"
+        raise NotImplementedError(msg)
+
+
+def parse_yaml(data: str) -> dict[Any, Any]:
+    """Parse YAML string and return the data in dictionary."""
+    return DataFormat.YAML.loads(data)
+
+
+def parse_json(data: str) -> dict[Any, Any]:
+    """Parse JSON string and return the data in dictionary."""
+    return DataFormat.JSON.loads(data)
+
+
+def try_to_parse(data: str) -> dict[Any, Any]:
+    """Attempt to parse data string as a JSON or YAML and return it's dictionary."""
+    try:
+        return parse_json(data)
+    except json.JSONDecodeError:
+        try:
+            return parse_yaml(data)
+        except yaml.YAMLError as e:
+            # YAML parsing error should be sufficient because the JSON can be parsed by the YAML parser.
+            # We should receive a helpful error message for JSON as well.
+            raise DataParsingError(e) from e

diff --git a/tests/python/knot_resolver/utils/modeling/test_parsing.py b/tests/python/knot_resolver/utils/modeling/test_parsing.py
new file mode 100644 (file)
index 0000000..63fe122
--- /dev/null
+++ b/tests/python/knot_resolver/utils/modeling/test_parsing.py
@@ -0,0 +1,122 @@
+import pytest
+
+from knot_resolver.utils.modeling.errors import DataParsingError
+from knot_resolver.utils.modeling.parsing import parse_json, parse_yaml, try_to_parse
+
+json_data = """
+{
+    "none": null,
+    "boolean": false,
+    "number": 2026,
+    "string": "this is string",
+    "object": {
+        "number": 5000,
+        "string": "this is object string"
+    },
+    "array": [
+        "item1",
+        "item2",
+        "item3"
+    ]
+}
+"""
+
+json_data_duplicates = """
+{
+    "duplicity-key": 1,
+    "duplicity-key": 2
+}
+"""
+
+json_data_duplicates_inner = """
+{
+    "object": {
+        "duplicity-key": 1,
+        "duplicity-key": 2
+    }
+}
+"""
+
+yaml_data = """
+none: null
+boolean: false
+number: 2026
+string: this is string
+object:
+  number: 5000
+  string: this is object string
+array:
+  - item1
+  - item2
+  - item3
+"""
+
+yaml_data_duplicates = """
+duplicity-key: 1
+duplicity-key: 2
+"""
+
+yaml_data_duplicates_inner = """
+object:
+    duplicity-key: 1
+    duplicity-key: 2
+"""
+
+data_dict = {
+    "none": None,
+    "boolean": False,
+    "number": 2026,
+    "string": "this is string",
+    "object": {
+        "number": 5000,
+        "string": "this is object string",
+    },
+    "array": [
+        "item1",
+        "item2",
+        "item3",
+    ],
+}
+
+
+def test_parse_json() -> None:
+    data = parse_json(json_data)
+    assert data == data_dict
+
+
+@pytest.mark.parametrize("data", [json_data, yaml_data])
+def test_parse_yaml(data: str) -> None:
+    data = parse_yaml(data)
+    assert data == data_dict
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        json_data_duplicates,
+        json_data_duplicates_inner,
+    ],
+)
+def test_parse_json_duplicates(data: str) -> None:
+    with pytest.raises(DataParsingError):
+        parse_json(data)
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        json_data_duplicates,
+        json_data_duplicates_inner,
+        yaml_data_duplicates,
+        yaml_data_duplicates_inner,
+    ],
+)
+def test_parse_yaml_duplicates(data: str) -> None:
+    with pytest.raises(DataParsingError):
+        parse_yaml(data)
+
+
+@pytest.mark.parametrize("data", [json_data, yaml_data])
+def test_try_to_parse(data: str) -> None:
+    data = try_to_parse(data)
+    assert data == data_dict