commons/src/commons/paths.py

"""nested dict/list access by dotted path

- deep_get/deep_set: single path, scalar in/out. numeric segment indexes a list;
  setting into a tuple raises (immutable).
- deep_gets/deep_sets: `*` segments iterate every element at a level (bulk get -> list,
  bulk set -> value or fn(current)->new); multiple `*` fan out cartesian.

a `*` in the scalar verbs raises ValueError - use the plural verbs so the return type
is never ambiguous.
"""
from typing import Any, Callable, List, Union

_MISSING = object()
_WILDCARD = "*"


def deep_get(data: Any, path: str, default: Any = None, *, sep: str = ".") -> Any:
    """get a nested value by dotted path, returning default if absent

    dict keys are matched by name; a numeric segment indexes a list/tuple
    (e.g. "items.0.id"). any missing key, out-of-range index, or non-container
    along the way yields `default` rather than raising. a `*` segment raises
    ValueError - use deep_gets for wildcard paths.
    """
    segments = path.split(sep)
    if _WILDCARD in segments:
        raise ValueError(f"deep_get: wildcard '*' in path {path!r}; use deep_gets for wildcard paths")
    cur = data
    for seg in segments:
        if isinstance(cur, dict):
            cur = cur.get(seg, _MISSING)
            if cur is _MISSING:
                return default
        elif isinstance(cur, (list, tuple)):
            try:
                cur = cur[int(seg)]
            except (ValueError, IndexError):
                return default
        else:
            return default
    return cur


def deep_set(data: dict, path: str, value: Any, *, sep: str = ".") -> dict:
    """set a nested value by dotted path, creating intermediate dicts; returns data for chaining

    a numeric segment updates a LIST element in place (out-of-range raises IndexError);
    setting through or into a tuple raises TypeError (immutable). a `*` raises ValueError -
    use deep_sets for wildcard paths.
    """
    segments = path.split(sep)
    if _WILDCARD in segments:
        raise ValueError(f"deep_set: wildcard '*' in path {path!r}; use deep_sets for wildcard paths")
    cur = data
    for seg in segments[:-1]:
        if isinstance(cur, tuple):
            raise TypeError(f"deep_set: cannot set into an immutable tuple at {path!r}")
        if isinstance(cur, list):
            idx = int(seg)
            if not -len(cur) <= idx < len(cur):
                raise IndexError(f"deep_set: index {seg!r} out of range for list of length {len(cur)}")
            nxt = cur[idx]
            if not isinstance(nxt, (dict, list, tuple)):
                nxt = {}
                cur[idx] = nxt
            cur = nxt
        else:
            nxt = cur.get(seg)
            if not isinstance(nxt, (dict, list, tuple)):
                nxt = {}
                cur[seg] = nxt
            cur = nxt
    last = segments[-1]
    if isinstance(cur, tuple):
        raise TypeError(f"deep_set: cannot set into an immutable tuple at {path!r}")
    if isinstance(cur, list):
        idx = int(last)
        if not -len(cur) <= idx < len(cur):
            raise IndexError(f"deep_set: index {last!r} out of range for list of length {len(cur)}")
        cur[idx] = value
    else:
        cur[last] = value
    return data


def _iter_children(node: Any):
    """yield (key, child) pairs for a `*` step: dict items, or list/tuple index/item pairs"""
    if isinstance(node, dict):
        yield from node.items()
    elif isinstance(node, (list, tuple)):
        yield from enumerate(node)


def deep_gets(data: Any, path: str, *, sep: str = ".") -> List[Any]:
    """get every value matching a dotted path with `*` wildcards, as a list (may be empty)

    a `*` iterates every element at that level; multiple `*` fan out cartesian. a branch
    that dead-ends (missing key, out-of-range index, non-container) is skipped, not raised.
    a path with no `*` returns a one- or zero-element list.
    """
    segments = path.split(sep)
    frontier = [data]
    for seg in segments:
        nxt: List[Any] = []
        if seg == _WILDCARD:
            for node in frontier:
                for _, child in _iter_children(node):
                    nxt.append(child)
        else:
            for node in frontier:
                if isinstance(node, dict):
                    got = node.get(seg, _MISSING)
                    if got is not _MISSING:
                        nxt.append(got)
                elif isinstance(node, (list, tuple)):
                    try:
                        nxt.append(node[int(seg)])
                    except (ValueError, IndexError):
                        pass
        frontier = nxt
    return frontier


def deep_sets(data: Any, path: str, value: "Union[Any, Callable[[Any], Any]]", *, sep: str = ".") -> Any:
    """set every value matching a dotted path with `*` wildcards; returns data for chaining

    `value` is a plain value written to every match, or a callable `fn(current) -> new`
    (a bare callable is always invoked - to store one, wrap it `lambda _c, f=fn: f`). a `*`
    iterates every element at that level; multiple `*` fan out cartesian. best-effort: any
    dead-end (missing key, out-of-range index) is skipped, not raised. setting through or
    into a tuple raises TypeError (immutable).
    """
    compute: Callable[[Any], Any] = value if callable(value) else (lambda _cur, _v=value: _v)
    segments = path.split(sep)
    parents = segments[:-1]
    last = segments[-1]

    frontier = [data]
    for seg in parents:
        nxt: List[Any] = []
        if seg == _WILDCARD:
            for node in frontier:
                if isinstance(node, tuple):
                    raise TypeError(f"deep_sets: cannot set through an immutable tuple at {path!r}")
                for _, child in _iter_children(node):
                    nxt.append(child)
        else:
            for node in frontier:
                if isinstance(node, tuple):
                    raise TypeError(f"deep_sets: cannot set through an immutable tuple at {path!r}")
                if isinstance(node, dict):
                    got = node.get(seg, _MISSING)
                    if got is not _MISSING:
                        nxt.append(got)
                elif isinstance(node, list):
                    try:
                        nxt.append(node[int(seg)])
                    except (ValueError, IndexError):
                        pass
        frontier = nxt

    for node in frontier:
        if isinstance(node, tuple):
            raise TypeError(f"deep_sets: cannot set into an immutable tuple at {path!r}")
        if last == _WILDCARD:
            if isinstance(node, dict):
                for key in list(node.keys()):
                    node[key] = compute(node[key])
            elif isinstance(node, list):
                for idx in range(len(node)):
                    node[idx] = compute(node[idx])
        elif isinstance(node, dict):
            if node.get(last, _MISSING) is not _MISSING:
                node[last] = compute(node[last])
        elif isinstance(node, list):
            try:
                idx = int(last)
            except ValueError:
                continue
            if -len(node) <= idx < len(node):
                node[idx] = compute(node[idx])
    return data