Add typ option to load_yaml_files (#49)

Author: oboehmer

CHANGELOG.md

@@ -3,6 +3,7 @@
 - BREAKING CHANGE: Rework list merge logic — within-file duplicates are now preserved; if any file contains duplicate dict items in a list, merging is disabled for that entire list and items are concatenated instead
 - BREAKING CHANGE: Two dict items in a list now merge when all shared primitive fields match, even if both sides have additional unique primitive fields (previously this was blocked)
 - Add support for Python 3.14
+- Add `typ` parameter to `load_yaml_files()` to select ruamel.yaml loader mode (e.g. `"safe"` for native dict/list)
 
 # 1.1.1
 

README.md

@@ -5,6 +5,29 @@
 
 A Python library with common YAML utility functions supporting `Network as Code`.
 
+## Usage
+
+### Load and merge YAML files
+
+```python
+from pathlib import Path
+from nac_yaml.yaml import load_yaml_files
+
+# Default is ruamel's round-trip loader (preserves formatting internally).
+data = load_yaml_files([
+    Path("path/to/file1.yaml"),
+    Path("path/to/file2.yaml"),
+])
+
+# Use the safe loader to get native dict/list containers.
+data_safe = load_yaml_files([
+    Path("path/to/file1.yaml"),
+    Path("path/to/file2.yaml"),
+], typ="safe")
+```
+
+Note: when `typ` is not round-trip (e.g. `"safe"`), formatting features (quotes/comments/style) are not preserved.
+
 ## Installation
 
 ### Using uv (recommended)

nac_yaml/yaml.py

@@ -142,7 +142,11 @@ def from_yaml(cls, loader: Any, node: Any) -> str:
         return str(cls(node.value))
 
 
-def load_yaml_files(paths: list[Path], deduplicate: bool = True) -> dict[str, Any]:
+def load_yaml_files(
+    paths: list[Path],
+    deduplicate: bool = True,
+    typ: str | None = None,
+) -> dict[str, Any]:
     """Load and merge YAML files from provided paths.
 
     Args:
@@ -151,6 +155,13 @@ def load_yaml_files(paths: list[Path], deduplicate: bool = True) -> dict[str, An
                     - If ANY file has duplicates in a list, that list is concatenated (no merging)
                     - If NO duplicates exist, matching dict items are merged across files
                     When False, simply concatenates all list items.
+        typ: Optional ruamel.yaml parser type passed to ``ruamel.yaml.YAML(typ=...)``.
+             Use ``"safe"`` to load native Python types (dict/list) instead of round-trip
+             containers.
+
+             Caveat: when ``typ`` is not round-trip (e.g. ``"safe"``), formatting-related
+             options like ``preserve_quotes`` do not apply and comments/quoting/style are not
+             preserved.
 
     Returns:
         Merged dictionary structure
@@ -188,7 +199,7 @@ def load_yaml_files(paths: list[Path], deduplicate: bool = True) -> dict[str, An
         Result: devices: [{name: switch1}, {name: switch1}, {name: switch1, port: 1/0/1}]  # all preserved
     """
     # Create YAML parser once and reuse for all files
-    y = yaml.YAML()
+    y = yaml.YAML(typ=typ) if typ is not None else yaml.YAML()
     y.preserve_quotes = True
     y.register_class(VaultTag)
     y.register_class(EnvTag)

scripts/compare_merge.py

@@ -80,7 +80,7 @@ def make_serializable(obj):
         return str(obj)
 
     paths = [Path(p) for p in sys.argv[1:]]
-    result = load_yaml_files(paths, deduplicate=True)
+    result = load_yaml_files(paths, deduplicate=True{typ_arg})
     json.dump(make_serializable(result), sys.stdout, indent=2, sort_keys=True)
 """)
 
@@ -123,12 +123,15 @@ def resolve_paths(raw_paths: list[str]) -> list[Path]:
     return resolved
 
 
-def run_version(version: str, paths: list[Path]) -> dict[str, Any]:
+def run_version(
+    version: str, paths: list[Path], typ: str | None = None
+) -> dict[str, Any]:
     """Run load_yaml_files() with a specific nac-yaml version in an isolated uv env.
 
     Args:
         version: nac-yaml PyPI version string (e.g. "1.1.1").
         paths: Absolute paths to YAML files/directories.
+        typ: Optional ruamel.yaml parser type to pass to load_yaml_files() for the NEW version.
 
     Returns:
         Parsed JSON dict of the merged YAML output.
@@ -137,6 +140,12 @@ def run_version(version: str, paths: list[Path]) -> dict[str, Any]:
         SystemExit: On subprocess failure.
     """
     str_paths = [str(p) for p in paths]
+
+    if version == NEW_VERSION and typ is not None:
+        typ_arg = f", typ={typ!r}"
+    else:
+        typ_arg = ""
+
     cmd = [
         "uv",
         "run",
@@ -145,7 +154,7 @@ def run_version(version: str, paths: list[Path]) -> dict[str, Any]:
         f"nac-yaml=={version}",
         "python",
         "-c",
-        _LOADER_SCRIPT_TEMPLATE,
+        _LOADER_SCRIPT_TEMPLATE.format(typ_arg=typ_arg),
         *str_paths,
     ]
     try:
@@ -1017,6 +1026,14 @@ def main() -> int:
         action="store_true",
         help="Render both merge results as YAML and show unified diff (like diff -u)",
     )
+    parser.add_argument(
+        "--typ",
+        metavar="TYP",
+        help=(
+            "Pass ruamel.yaml typ to load_yaml_files() for the NEW version only "
+            '(e.g. "safe" for native dict/list)'
+        ),
+    )
     args = parser.parse_args()
 
     if args.diff and (args.json_output or args.raw):
@@ -1035,7 +1052,7 @@ def main() -> int:
     )
     with ThreadPoolExecutor(max_workers=2) as pool:
         future_old = pool.submit(run_version, OLD_VERSION, resolved)
-        future_new = pool.submit(run_version, NEW_VERSION, resolved)
+        future_new = pool.submit(run_version, NEW_VERSION, resolved, args.typ)
         old_data = future_old.result()
         new_data = future_new.result()
 

tests/unit/test_yaml.py

@@ -13,7 +13,16 @@
 pytestmark = pytest.mark.unit
 
 
-def test_load_yaml_files(tmpdir: Path) -> None:
+def _is_only_dict_list_tree(value: Any) -> bool:
+    if isinstance(value, dict):
+        return all(_is_only_dict_list_tree(v) for v in value.values())
+    if isinstance(value, list):
+        return all(_is_only_dict_list_tree(v) for v in value)
+    return True
+
+
+@pytest.mark.parametrize("typ", [None, "rt", "safe"])
+def test_load_yaml_files(tmpdir: Path, typ: str | None) -> None:
     input_path_1 = Path("tests/unit/fixtures/data_merge/file1.yaml")
     input_path_2 = Path("tests/unit/fixtures/data_merge/file2.yaml")
     output_path = Path(tmpdir, "output.yaml")
@@ -22,22 +31,33 @@ def test_load_yaml_files(tmpdir: Path) -> None:
         "tests/unit/fixtures/data_merge/result_no_deduplicate.yaml"
     )
 
-    data = yaml.load_yaml_files([input_path_1, input_path_2])
+    data = yaml.load_yaml_files([input_path_1, input_path_2], typ=typ)
+    if typ == "safe":
+        assert _is_only_dict_list_tree(data)
     yaml.write_yaml_file(data, output_path)
     assert filecmp.cmp(output_path, result_path, shallow=False)
 
-    data = yaml.load_yaml_files([input_path_1, input_path_2], deduplicate=False)
+    data = yaml.load_yaml_files(
+        [input_path_1, input_path_2], deduplicate=False, typ=typ
+    )
+    if typ == "safe":
+        assert _is_only_dict_list_tree(data)
     yaml.write_yaml_file(data, output_path)
     assert filecmp.cmp(output_path, result_no_deduplicate_path, shallow=False)
 
     input_path = Path("tests/unit/fixtures/data_vault/")
     os.environ["ANSIBLE_VAULT_ID"] = "dev"
     os.environ["ANSIBLE_VAULT_PASSWORD"] = "Password123"
-    data = yaml.load_yaml_files([input_path])
+    data = yaml.load_yaml_files([input_path], typ=typ)
+    if typ == "safe":
+        assert _is_only_dict_list_tree(data)
+    assert data["root"]["children"][0]["name"] == "ABC\n"
 
     input_path = Path("tests/unit/fixtures/data_env/")
     os.environ["ABC"] = "DEF"
-    data = yaml.load_yaml_files([input_path])
+    data = yaml.load_yaml_files([input_path], typ=typ)
+    if typ == "safe":
+        assert _is_only_dict_list_tree(data)
     assert data["root"]["children"][0]["name"] == "DEF"