Source code for semi_cr.core.lab.utils.yaml_utils

from collections.abc import Mapping
from pathlib import Path
from typing import Any

import yaml




def deep_merge(base: dict[str, Any], incoming: Mapping[str, Any]) -> dict[str, Any]:
    """
    Recursively merge `incoming` into `base`.

    - dict + dict => merge recursively
    - otherwise   => incoming overwrites base

    Returns `base` (mutated) for convenience.
    """

    # We're looping over each key-value pair in the incoming dict. If we know the
    # key already, we may recurse to merge the subtree. If not, we just add the sub-
    # tree
    for k, v in incoming.items():
        if k in base and isinstance(base[k], dict) and isinstance(v, Mapping):
            deep_merge(base[k], v)
        else:
            base[k] = v
    return base



def _read_yaml(path: str | Path) -> dict[str, Any]:
    p = Path(path)
    data = yaml.safe_load(p.read_text(encoding="utf-8"))
    if data is None:
        return {}
    if not isinstance(data, dict):
        raise ValueError(f"Expected a YAML mapping at top-level in {p}, got {type(data).__name__}")
    return data