fix: deep_set fails loud on a tuple instead of silently replacing it; add deep_gets/deep_sets wildcard verbs
deep_set stepping into or through a tuple used to fall through to the dict branch and
replace the tuple with {}, silently corrupting data on a deep_get/deep_set round-trip
(deep_get traverses tuples read-only). it now raises TypeError - tuples are immutable, so
an in-place update is impossible. also adds deep_gets/deep_sets: '*' path segments for
bulk get (always a list) and bulk set (plain value or fn(current)->new); a '*' passed to
the scalar deep_get/deep_set raises ValueError so the return type is never ambiguous.
Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
@@ -89,13 +89,17 @@ timing.FAST_MODE = True # in test setup
|
||||
|
||||
## paths
|
||||
|
||||
Two pairs of verbs: **single-path** (scalar in/out) and **wildcard** (bulk, always a list).
|
||||
|
||||
### single-path — `deep_get` / `deep_set`
|
||||
|
||||
```python
|
||||
from commons import deep_get, deep_set
|
||||
|
||||
data = {"in": {"this": {"old": {"notation": 42}}}, "items": [{"id": "a"}, {"id": "b"}]}
|
||||
|
||||
deep_get(data, "in.this.old.notation") # 42
|
||||
deep_get(data, "items.1.id") # "b" (numeric segment indexes a list)
|
||||
deep_get(data, "items.1.id") # "b" (numeric segment indexes a list/tuple)
|
||||
deep_get(data, "in.nope.here", "DEF") # "DEF" (missing -> default, no raise)
|
||||
|
||||
deep_set({}, "a.b.c", 9) # {"a": {"b": {"c": 9}}}
|
||||
@@ -105,10 +109,34 @@ deep_set(data, "items.9.id", "X") # raises IndexError (out of range, no
|
||||
# silent mis-store)
|
||||
```
|
||||
|
||||
`deep_set` mirrors `deep_get`'s list indexing: a numeric segment over an existing
|
||||
list/tuple updates that element in place rather than replacing the list with a dict.
|
||||
An out-of-range numeric segment raises `IndexError` instead of silently corrupting
|
||||
the structure.
|
||||
`deep_get` steps into both lists and tuples on a numeric segment. `deep_set` updates a
|
||||
**list** element in place; setting into (or through) a **tuple** raises `TypeError` —
|
||||
tuples are immutable, so "update in place" is impossible; flatten to a list first. An
|
||||
out-of-range numeric segment raises `IndexError` instead of silently corrupting the
|
||||
structure.
|
||||
|
||||
### wildcard — `deep_gets` / `deep_sets`
|
||||
|
||||
A `*` segment iterates every element at that level (dict values or list/tuple items);
|
||||
multiple `*` fan out cartesian. These are **separate verbs** with a fixed list/bulk
|
||||
return type — a `*` passed to `deep_get`/`deep_set` raises `ValueError` (use the plural
|
||||
verb).
|
||||
|
||||
```python
|
||||
from commons import deep_gets, deep_sets
|
||||
|
||||
data = {"users": [{"username": "al"}, {"username": "bo"}, {"username": "cy"}]}
|
||||
|
||||
deep_gets(data, "users.*.username") # ["al", "bo", "cy"] (ALWAYS a list, in order)
|
||||
deep_gets(data, "users.*.nope") # [] (no match -> empty list, no raise)
|
||||
|
||||
deep_sets(data, "users.*.username", "X") # set every match to "X"
|
||||
deep_sets(data, "users.*.username", str.upper) # callable fn(current)->new per match
|
||||
```
|
||||
|
||||
`deep_sets`'s second arg is either a plain value (write it to every match) or a callable
|
||||
`fn(current) -> new` (compute each). It writes only matches that already exist (it does
|
||||
not create missing keys). Setting through/into a tuple raises `TypeError`.
|
||||
|
||||
## masking
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ from importlib.metadata import PackageNotFoundError, version
|
||||
|
||||
from . import addr, masking, paths, timing
|
||||
from .masking import credit, cvv, mask_proxy, mask_url, phantom, provider
|
||||
from .paths import deep_get, deep_set
|
||||
from .paths import deep_get, deep_gets, deep_set, deep_sets
|
||||
from .retry import aretry, retry
|
||||
from .timing import (
|
||||
UTC,
|
||||
@@ -45,6 +45,8 @@ __all__ = [
|
||||
"Clock",
|
||||
"deep_get",
|
||||
"deep_set",
|
||||
"deep_gets",
|
||||
"deep_sets",
|
||||
"credit",
|
||||
"cvv",
|
||||
"phantom",
|
||||
|
||||
+128
-9
@@ -1,7 +1,17 @@
|
||||
"""nested dict/list access by dotted path (deep_get / deep_set)."""
|
||||
from typing import Any
|
||||
"""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:
|
||||
@@ -9,10 +19,14 @@ def deep_get(data: Any, path: str, default: Any = None, *, sep: str = ".") -> An
|
||||
|
||||
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.
|
||||
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 path.split(sep):
|
||||
for seg in segments:
|
||||
if isinstance(cur, dict):
|
||||
cur = cur.get(seg, _MISSING)
|
||||
if cur is _MISSING:
|
||||
@@ -30,29 +44,35 @@ def deep_get(data: Any, path: str, default: Any = None, *, sep: str = ".") -> An
|
||||
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
|
||||
|
||||
mirrors deep_get's list indexing (a numeric segment updates an existing list/tuple
|
||||
element rather than corrupting it); unlike deep_get, an out-of-range index raises
|
||||
IndexError instead of silently mis-storing, since there's no safe default to fall back to.
|
||||
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)):
|
||||
if not isinstance(nxt, (dict, list, tuple)):
|
||||
nxt = {}
|
||||
cur[idx] = nxt
|
||||
cur = nxt
|
||||
else:
|
||||
nxt = cur.get(seg)
|
||||
if not isinstance(nxt, (dict, list)):
|
||||
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):
|
||||
@@ -61,3 +81,102 @@ def deep_set(data: dict, path: str, value: Any, *, sep: str = ".") -> dict:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user