From: Aleš Mrázek Date: Mon, 16 Feb 2026 18:28:49 +0000 (+0100) Subject: python: modeling: added JSON/YAML parsing X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=0aa8e21f84698eb32efc3aac3b366bcbe10df8bf;p=thirdparty%2Fknot-resolver.git 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 index 000000000..e69de29bb diff --git a/python/knot_resolver/utils/modeling/parsing.py b/python/knot_resolver/utils/modeling/parsing.py new file mode 100644 index 000000000..d57000c69 --- /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 index 000000000..63fe122da --- /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