Raise if duplicate keys exist (#22)

Author: stevebachmeier

.github/CODEOWNERS

@@ -1,2 +1,2 @@
 # default owners
-* @albrja @collijk @hussain-jafari @patricktnast @rmudambi @stevebachmeier
+* @albrja @hussain-jafari @patricktnast @rmudambi @stevebachmeier

CHANGELOG.rst

@@ -1,3 +1,7 @@
+**3.1.0 - 03/18/2025**
+
+ - Raise an error if YAML contains duplicate keys within the same level
+
 **3.0.0 - 02/18/2025**
 
  - Better handle dunder-style keys

src/layered_config_tree/__init__.py

@@ -14,4 +14,4 @@
     DuplicatedConfigurationError,
     ImproperAccessError,
 )
-from layered_config_tree.main import ConfigNode, LayeredConfigTree
+from layered_config_tree.main import ConfigNode, LayeredConfigTree, load_yaml

src/layered_config_tree/main.py

@@ -33,15 +33,14 @@
 from pathlib import Path
 from typing import Any
 
-import yaml
-
 from layered_config_tree import (
     ConfigurationError,
     ConfigurationKeyError,
     DuplicatedConfigurationError,
     ImproperAccessError,
 )
 from layered_config_tree.types import InputData
+from layered_config_tree.utilities import load_yaml
 
 
 class ConfigNode:
@@ -461,30 +460,14 @@ def _coerce(
         """Coerces data into dictionary format."""
         if isinstance(data, dict):
             return data, source
-        elif (isinstance(data, str) and data.endswith((".yaml", ".yml"))) or isinstance(
-            data, Path
-        ):
-            source = source if source else str(data)
-            with open(data) as f:
-                data_file = f.read()
-            coerced_data = yaml.full_load(data_file)
-            if not isinstance(coerced_data, dict):
-                raise ValueError(
-                    f"Loaded yaml file {coerced_data} should be a dictionary but is type {type(coerced_data)}"
-                )
-            return coerced_data, source
-        elif isinstance(data, str):
-            data = yaml.full_load(data)
-            if not isinstance(data, dict):
-                raise ValueError(
-                    f"Loaded yaml file {data} should be a dictionary but is type {type(data)}"
-                )
-            return data, source
         elif isinstance(data, LayeredConfigTree):
             return data.to_dict(), source
+        elif isinstance(data, (str, Path)):
+            source = source if source else str(data)
+            return load_yaml(data), source
         else:
             raise ConfigurationError(
-                f"LayeredConfigTree can only update from dictionaries, strings, paths and LayeredConfigTrees. "
+                f"LayeredConfigTree can only update from dictionaries, strings, paths, and LayeredConfigTrees. "
                 f"You passed in {type(data)}",
                 value_name=None,
             )

src/layered_config_tree/utilities.py

@@ -0,0 +1,95 @@
+"""
+=========
+Utilities
+=========
+
+This module contains utility functions and classes for the layered_config_tree
+package.
+
+"""
+
+from collections.abc import Hashable
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from layered_config_tree import DuplicatedConfigurationError
+
+
+def load_yaml(data: str | Path) -> dict[str, Any]:
+    """Loads a YAML filepath or string into a dictionary.
+
+    Parameters
+    ----------
+    data
+        The YAML content to load. This can be a file path to a YAML file or a string
+        containing YAML-formatted text.
+
+    Raises
+    ------
+    ValueError
+        If the loaded YAML content is not a dictionary.
+
+    Returns
+    -------
+        A dictionary representation of the loaded YAML content.
+
+    Notes
+    -----
+    If `data` is a Path object or a string that ends with ".yaml" or ".yml", it is
+    treated as a filepath and this function loads the file. Otherwise, `data` is a
+    string that does _not_ end in ".yaml" or ".yml" and it is treated as YAML-formatted
+    text which is loaded directly into a dictionary.
+    """
+
+    if (isinstance(data, str) and data.endswith((".yaml", ".yml"))) or isinstance(data, Path):
+        # 'data' is a filepath to a yaml file (rather than a yaml string)
+        with open(data) as f:
+            data = f.read()
+    data_dict: dict[str, Any] = yaml.load(data, Loader=SafeLoader)
+
+    if not isinstance(data_dict, dict):
+        raise ValueError(
+            f"Loaded yaml file {data_dict} should be a dictionary but is type {type(data_dict)}"
+        )
+
+    return data_dict
+
+
+class SafeLoader(yaml.SafeLoader):
+    """A yaml.SafeLoader that restricts duplicate keys."""
+
+    def construct_mapping(
+        self, node: yaml.MappingNode, deep: bool = False
+    ) -> dict[Hashable, Any]:
+        """Constructs the standard mapping after checking for duplicates.
+
+        Raises
+        ------
+        DuplicateKeysInYAMLError
+            If duplicate keys within the same level are detected in the YAML file
+            being loaded.
+
+        Notes
+        -----
+        A key is considered a duplicate only if it is the same as another key
+        _at the same level in the YAML_.
+
+        This raises upon the _first_ duplicate key found; other duplicates may exist
+        (in which case a new error will be raised upon re-loading of the YAML file
+        once the duplicate is resolved).
+        """
+        mapping = []
+        for key_node, _value_node in node.value:
+            key = self.construct_object(key_node, deep=deep)  # type: ignore[no-untyped-call]
+            if key in mapping:
+                raise DuplicatedConfigurationError(
+                    f"Duplicate key detected at same level of YAML: {key}. Resolve duplicates and try again.",
+                    name=f"duplicated_{key}",
+                    layer=None,
+                    source=None,
+                    value=None,
+                )
+            mapping.append(key)
+        return super().construct_mapping(node, deep)

tests/test_basic_functionality.py

@@ -14,6 +14,7 @@
     DuplicatedConfigurationError,
     ImproperAccessError,
     LayeredConfigTree,
+    load_yaml,
 )
 
 
@@ -500,8 +501,7 @@ def test_to_dict_dict() -> None:
 
 def test_to_dict_yaml(test_spec: Path) -> None:
     lct = LayeredConfigTree(str(test_spec))
-    with test_spec.open() as f:
-        yaml_config = yaml.full_load(f)
+    yaml_config = load_yaml(test_spec)
     assert yaml_config == lct.to_dict()
 
 

tests/test_ingestion.py

@@ -1,6 +1,9 @@
+import re
 from pathlib import Path
 
-from layered_config_tree import LayeredConfigTree
+import pytest
+
+from layered_config_tree import DuplicatedConfigurationError, LayeredConfigTree
 
 TEST_YAML_ONE = """
 test_section:
@@ -11,6 +14,26 @@
     test_key2: test_value4
 """
 
+TEST_YAML_DUPLICATE_KEYS = """
+cats:
+    simba:
+        size: tiny
+        features:
+            - cute
+            - playful
+        color: brown
+    garfield:
+        size: chonky
+        traits:
+            - lazy
+            - grumpy
+            - loves lasagna
+        color: orange
+        size: thick  # first duplicate; we raise here
+        color: brown  # second duplicate; no raise
+
+"""
+
 
 def test_load_yaml_string() -> None:
     lct = LayeredConfigTree()
@@ -21,13 +44,41 @@ def test_load_yaml_string() -> None:
     assert lct.test_section2.test_key == "test_value3"
 
 
-def test_load_yaml_file(tmp_path: Path) -> None:
+@pytest.mark.parametrize("path_type", [str, Path])
+def test_load_yaml_file(tmp_path: Path, path_type: type[str | Path]) -> None:
     tmp_file = tmp_path / "test_file.yaml"
     tmp_file.write_text(TEST_YAML_ONE)
 
     lct = LayeredConfigTree()
-    lct.update(str(tmp_file))
+    filepath_to_test = str(tmp_file) if path_type is str else tmp_file
+    lct.update(filepath_to_test)
 
     assert lct.test_section.test_key == "test_value"
     assert lct.test_section.test_key2 == "test_value2"
     assert lct.test_section2.test_key == "test_value3"
+
+
+@pytest.mark.parametrize("duplicates", [True, False])
+@pytest.mark.parametrize("load_from_file", [True, False])
+def test_load_yaml_duplicates_raise(
+    duplicates: bool, load_from_file: bool, tmp_path: Path
+) -> None:
+    test_str: str = TEST_YAML_DUPLICATE_KEYS if duplicates else TEST_YAML_ONE
+    if load_from_file:
+        tmp_file = tmp_path / "test_duplicate_keys.yaml"
+        tmp_file.write_text(test_str)
+    test_yaml = tmp_file if load_from_file else test_str
+
+    lct = LayeredConfigTree()
+
+    if duplicates:
+        with pytest.raises(
+            DuplicatedConfigurationError,
+            match=re.escape(
+                "Duplicate key detected at same level of YAML: size. Resolve duplicates and try again."
+            ),
+        ):
+            lct.update(test_yaml)
+    else:
+        # Only duplicate keys on the same level are problematic!
+        lct.update(test_yaml)