pygx.topology

Tree topology primitives: paths, missing-value sentinels, and traversal.

topology

Tree topology primitives: paths, missing-value sentinels, and traversal.

pygx.topology defines the value-space vocabulary that other layers build on:

• KeyPath / KeyPathSet address nodes inside a nested structure (StrKey marks key types rendered dot-delimited), and message_on_path produces path-tagged error messages.
• MissingValue and the MISSING_VALUE sentinel mark unbound fields.
• MaybePartial is the partiality interface implemented by every symbolic value, and is_partial tests any value tree for missing parts.
• CopyIfDefault is the opt-in marker for types whose instances copy per construct when used as static defaults.
• traverse, transform, flatten, canonicalize, merge, and merge_tree walk and rewrite (maybe-)hierarchical Python values keyed by KeyPath.

JSON serialization for these structures lives in pygx.wire.

CopyIfDefault

Opt-in marker: copy instances of this type when used as a static default.

Static defaults copy selectively (aligned with Pydantic): builtin mutable containers, symbolic values, and dataclass instances deep-copy per construct, while arbitrary-type templates are shared by reference — and defining __deepcopy__ does not opt a type into copying (incidental deepcopyability is not intent).

CopyIfDefault is the declared intent: a class extending this interface has its instances deep-copied per construct when they appear as a static default — for types that carry per-placement mutable state and must never be shared across instances. (Symbolic values, including contextual/inferred markers like pg.contextual.Placeholder, are already copied via their own copy protocol and need no marker.) The value-side alternative remains default_factory, which declares per-instance construction at the field rather than the type.

MaybePartial

Interface for classes whose instances can be partially constructed.

A MaybePartial object is an object whose __init__ method can accept pg.MISSING_VALUE as its argument values. All symbolic types (see pygx.Symbolic) implement this interface, as their symbolic attributes can be partially filled.

Partiality is judged against a schema: a required field left as pg.MISSING_VALUE makes the value partial, while a schemaless pg.Dict simply drops missing entries and stays complete.

Example:

class A(pg.Object):
  x: int
  y: str = 'foo'

a = A.partial()  # `x` is left missing.
assert a.sym_partial
assert 'x' in a.sym_missing()

sym_partial property

sym_partial: bool

Returns True if this object is partial. Otherwise False.

An object is considered partial when any of its required fields is missing, or at least one member is partial. A subclass can override this property to provide a more efficient solution.

sym_missing abstractmethod

sym_missing(flatten: bool = True) -> dict[str | int, Any]

Returns missing values from this object.

Parameters:

Name	Type	Description	Default
flatten	bool	If True, convert nested structures into a flattened dict using key path (delimited by '.' and '[]') as key.	True

Returns:

Type	Description
dict[str | int, Any]	A dict of key to MISSING_VALUE.

Source code in pygx/topology/_maybe_partial.py

@abc.abstractmethod
def sym_missing(self, flatten: bool = True) -> dict[str | int, Any]:  # pylint: disable=redefined-outer-name
    """Returns missing values from this object.

    Args:
      flatten: If True, convert nested structures into a flattened dict
        using key path (delimited by '.' and '[]') as key.

    Returns:
      A dict of key to MISSING_VALUE.
    """

MissingValue

Bases: Formattable, WireConvertible

Value placeholder for an unassigned attribute.

pg.MISSING_VALUE is the canonical instance: field machinery uses it to mark a declared field whose value has not been provided (see MaybePartial). All MissingValue instances compare equal (and share one hash), it is wire-convertible, and the class cannot be subclassed -- the symbolic core detects the sentinel with an exact-type check on its hot paths.

KeyPath

KeyPath(
    key_or_key_list: Any | list[Any] | None = None,
    parent: Optional[KeyPath] = None,
)

Bases: Formattable, WireConvertible

Represents a path of keys from the root to a node in a tree.

KeyPath is an important concept in PyGX, which is used for representing a symbolic object's location (see pygx.Symbolic.topo_path) within its symbolic tree. Note that topo_path requires classes declared with topo=True (the default topo=False keeps no parent/path state); KeyPath itself works with values of either mode. For example:

class A(pg.Object, topo=True):
  x: int
  y: str

class B(pg.Object, topo=True):
  z: A

a = A(x=1, y='foo')
b = B(z=a)
assert a.topo_path == 'z' # The path to object `a` is 'z'.
assert b.topo_path == ''  # The root object's KeyPath is empty.

Since each node in a tree has a unique location, given the root we shall be able to use a KeyPath object to locate the node. With the example above, we can query the member x of object a via:

pg.KeyPath.parse('z.x').query(b)  # Should return 1.

Similarly, we can modify a symbolic object's sub-node based on a KeyPath object. See pygx.Symbolic.sym_rebind for modifying sub-nodes in a symbolic tree.

Constructor.

Parameters:

Name	Type	Description	Default
key_or_key_list	Any | list[Any] | None	A single object as key, or a list/tuple of objects as keys in the path. When string types or StrKey objects are used as key, dot ('.') is used as the delimiter, otherwise square brackets ('[]') is used as the delimiter when formatting a KeyPath. For object type key, str(object) will be used to represent the key in string form.	None
parent	Optional[KeyPath]	Parent KeyPath.	None

Source code in pygx/topology/_value_location.py

def __init__(
    self,
    key_or_key_list: Any | list[Any] | None = None,
    parent: Optional['KeyPath'] = None,
):
    """Constructor.

    Args:
      key_or_key_list: A single object as key, or a list/tuple of objects
        as keys in the path.
        When string types or StrKey objects are used as key, dot ('.') is
        used as the delimiter, otherwise square brackets ('[]') is used as
        the delimiter when formatting a KeyPath.
        For object type key, str(object) will be used to represent the key
        in string form.
      parent: Parent KeyPath.
    """
    # `KeyPath(key, parent)` is on the per-element apply hot path. The branch
    # order below is tuned for that: `is None` (cheap `is` check) first, then
    # the two container types. Each branch builds a fresh list — relying on
    # `list +` to produce a new list — so we never alias an input.
    if parent is None:
        if key_or_key_list is None:
            self._keys = []
        elif isinstance(key_or_key_list, list):
            self._keys = list(key_or_key_list)
        elif isinstance(key_or_key_list, tuple):
            self._keys = list(key_or_key_list)
        else:
            self._keys = [key_or_key_list]
    else:
        parent_keys = parent._keys  # pylint: disable=protected-access
        if key_or_key_list is None:
            self._keys = list(parent_keys)
        elif isinstance(key_or_key_list, list):
            self._keys = parent_keys + key_or_key_list
        elif isinstance(key_or_key_list, tuple):
            self._keys = parent_keys + list(key_or_key_list)
        else:
            self._keys = parent_keys + [key_or_key_list]
    # Lazy cache for `path_str()`: many KeyPath instances never need the
    # string form, and computing it allocates O(depth) substrings.
    self._path_str = None

keys property

keys: list[Any]

A list of keys in this path.

key property

key: Any

The rightmost key of this path.

is_root property

is_root: bool

Returns True if this path is the root of a tree.

depth property

depth: int

The depth of this path.

parent property

parent: KeyPath

The KeyPath object for current node's parent.

Example:

path = pg.KeyPath.parse('a.b.c')
assert path.parent == 'a.b'

Returns:

Type	Description
KeyPath	A KeyPath object for the parent of current node.

Raises:

Type	Description
KeyError	If current path is the root.

path property

path: str

JSONPath representation of current path.

from_value classmethod

from_value(value: Union[KeyPath, str, int]) -> KeyPath

Returns a KeyPath object from a KeyPath equivalence.

Source code in pygx/topology/_value_location.py

@classmethod
def from_value(cls, value: Union['KeyPath', str, int]) -> 'KeyPath':
    """Returns a KeyPath object from a KeyPath equivalence."""
    # Fast path for already-built KeyPaths and for simple keys: a non-empty
    # string without path-syntax chars ('.', '[', ']') is always a single
    # element path, and `parse` would just scan it character-by-character.
    if isinstance(value, KeyPath):
        return value
    if isinstance(value, str):
        if not value or '.' in value or '[' in value or ']' in value:
            return cls.parse(value)
        return cls(value)
    if isinstance(value, int):
        return cls(value)
    raise ValueError(f'{value!r} is not a valid KeyPath equivalence.')

parse classmethod

parse(path_str: str, parent: Optional[KeyPath] = None) -> KeyPath

Creates a KeyPath object from parsing a JSONPath-like string.

The JSONPath (https://restfulapi.net/json-jsonpath/) like string is defined as following:

<path>             := <empty> | {<dict-key>[.<dict-key>]*}
<dict-key>         := <identifier>
                      ['[('<list-key>|<special-dict-key>)']']*
<list-key>         := <number>
<special-dict-key> := <string-with-delimiter-chars>
<delimiter-chars>  := '[' | ']' | '.'

For example, following keys are valid path strings:

''               : An empty path representing the root of a path.
'a'              : A path that contains a dict key 'a'.
'a.b'            : A path that contains two dict keys 'a' and 'b'.
'a[0]'           : A path that contains a dict key 'a' and a list key 0.
'a.0.'           : A path that contains two dict keys 'a' and '0'.
'a[0][1]'        : A path that contains a dict key 'a' and two list keys
                   0 and 1 for a multi-dimension list.
'a[x.y].b'       : A path with three dict keys: 'a', 'x.y' and 'b'.
                   Since 'x.y' has delimiter characters, it needs to be
                   enclosed in brackets.

Note: only string and int keys can be parsed, so format and parse are not symmetric — format can render a KeyPath that includes complex (non-string, non-int) keys into a string, but parse cannot convert such a string back.

Parameters:

Name	Type	Description	Default
path_str	str	A JSON-path-like string.	required
parent	Optional[KeyPath]	Parent KeyPath object.	None

Returns:

Type	Description
KeyPath	A KeyPath object.

Raises:

Type	Description
ValueError	Path string is in bad format.

Source code in pygx/topology/_value_location.py

@classmethod
def parse(
    cls, path_str: str, parent: Optional['KeyPath'] = None
) -> 'KeyPath':
    """Creates a ``KeyPath`` object from parsing a JSONPath-like string.

    The JSONPath (https://restfulapi.net/json-jsonpath/) like string is
    defined as following:

    ```text
    <path>             := <empty> | {<dict-key>[.<dict-key>]*}
    <dict-key>         := <identifier>
                          ['[('<list-key>|<special-dict-key>)']']*
    <list-key>         := <number>
    <special-dict-key> := <string-with-delimiter-chars>
    <delimiter-chars>  := '[' | ']' | '.'
    ```

    For example, following keys are valid path strings:

    ```text
    ''               : An empty path representing the root of a path.
    'a'              : A path that contains a dict key 'a'.
    'a.b'            : A path that contains two dict keys 'a' and 'b'.
    'a[0]'           : A path that contains a dict key 'a' and a list key 0.
    'a.0.'           : A path that contains two dict keys 'a' and '0'.
    'a[0][1]'        : A path that contains a dict key 'a' and two list keys
                       0 and 1 for a multi-dimension list.
    'a[x.y].b'       : A path with three dict keys: 'a', 'x.y' and 'b'.
                       Since 'x.y' has delimiter characters, it needs to be
                       enclosed in brackets.
    ```

    Note: only string and int keys can be parsed, so ``format`` and
    ``parse`` are not symmetric — ``format`` can render a ``KeyPath``
    that includes complex (non-string, non-int) keys into a string, but
    ``parse`` cannot convert such a string back.

    Args:
      path_str: A JSON-path-like string.
      parent: Parent KeyPath object.

    Returns:
      A KeyPath object.

    Raises:
      ValueError: Path string is in bad format.
    """
    if not isinstance(path_str, str):
        raise ValueError(
            f"'path_str' must be a string type. Encountered: {path_str!r}"
        )

    keys = []

    def _append_key(key, preserve_empty=False, maybe_numeric=False):
        """Helper method to append key."""
        if not (preserve_empty or key):
            return
        if maybe_numeric and key.lstrip('-').isdigit():
            key = int(key)
        keys.append(key)

    pos, key_start, unmatched_brackets = 0, 0, 0
    while pos != len(path_str):
        ch = path_str[pos]
        if ch == ']':
            unmatched_brackets -= 1
            if unmatched_brackets == 0:
                key = path_str[key_start:pos]
                _append_key(key, True, True)
                key_start = pos + 1
            elif unmatched_brackets < 0:
                raise ValueError(
                    f'KeyPath parse failed: unmatched close bracket at position '
                    f'{pos}:{path_str!r}'
                )
        elif ch == '[':
            if unmatched_brackets == 0:
                key = path_str[key_start:pos]
                _append_key(key)
                key_start = pos + 1
            unmatched_brackets += 1
        elif ch == '.' and unmatched_brackets == 0:
            key = path_str[key_start:pos]
            _append_key(key)
            key_start = pos + 1
        pos += 1
    if key_start != len(path_str):
        _append_key(path_str[key_start:])
    if unmatched_brackets != 0:
        raise ValueError(
            f'KeyPath parse failed: unmatched open bracket at position '
            f'{key_start - 1}: {path_str!r}'
        )
    return KeyPath(keys, parent)

query

query(src: Any, use_inferred: bool = False) -> Any

Query the value from the source object based on current path.

Example:

class A(pg.Object):
  x: int
  y: str

class B(pg.Object):
  z: A

b = B(z=A(x=1, y='foo'))
assert pg.KeyPath.parse('z.x').query(b) == 1

Parameters:

Name	Type	Description	Default
src	Any	Source value to query.	required
use_inferred	bool	If True, infer pg.Inferential values. Otherwise returns their symbolic form. Applicable only for symbolic values.	False

Returns:

Type	Description
Any	Value from src if path exists.

Raises:

Type	Description
KeyError	Path doesn't exist in src.

Source code in pygx/topology/_value_location.py

def query(self, src: Any, use_inferred: bool = False) -> Any:
    """Query the value from the source object based on current path.

    **Example:**

    ```python
    class A(pg.Object):
      x: int
      y: str

    class B(pg.Object):
      z: A

    b = B(z=A(x=1, y='foo'))
    assert pg.KeyPath.parse('z.x').query(b) == 1
    ```

    Args:
      src: Source value to query.
      use_inferred: If True, infer `pg.Inferential` values. Otherwise
        returns their symbolic form. Applicable only for symbolic values.

    Returns:
      Value from src if path exists.

    Raises:
      KeyError: Path doesn't exist in src.
    """
    return self._query(0, src, use_inferred)

get

get(
    src: Any, default_value: Any | None = None, use_inferred: bool = False
) -> Any

Gets the value at the current path, or a default if it's absent.

Source code in pygx/topology/_value_location.py

def get(
    self,
    src: Any,
    default_value: Any | None = None,
    use_inferred: bool = False,
) -> Any:
    """Gets the value at the current path, or a default if it's absent."""
    try:
        return self.query(src, use_inferred)
    except KeyError:
        return default_value

exists

exists(src: Any) -> bool

Returns whether current path exists in source object.

Source code in pygx/topology/_value_location.py

def exists(self, src: Any) -> bool:
    """Returns whether current path exists in source object."""
    try:
        self.query(src)
        return True
    except KeyError:
        return False

is_relative_to

is_relative_to(other: Union[int, str, KeyPath]) -> bool

Returns whether current path is relative to another path.

Source code in pygx/topology/_value_location.py

def is_relative_to(self, other: Union[int, str, 'KeyPath']) -> bool:
    """Returns whether current path is relative to another path."""
    other = KeyPath.from_value(other)
    self_keys = self._keys
    other_keys = other._keys  # pylint: disable=protected-access
    n = len(other_keys)
    if len(self_keys) < n:
        return False
    for i in range(n):
        if self_keys[i] != other_keys[i]:
            return False
    return True

to_json

to_json(**kwargs: Any) -> dict[str, Any]

Serializes to its JSONPath string (wire-convertible, round-trips).

Source code in pygx/topology/_value_location.py

def to_json(self, **kwargs: Any) -> dict[str, Any]:
    """Serializes to its JSONPath string (wire-convertible, round-trips)."""
    return self.to_json_dict(fields=dict(value=self.path), **kwargs)

from_json classmethod

from_json(json_value: Any, **kwargs: Any) -> KeyPath

Rebuilds a KeyPath from its serialized JSONPath string.

Source code in pygx/topology/_value_location.py

@classmethod
def from_json(cls, json_value: Any, **kwargs: Any) -> 'KeyPath':
    """Rebuilds a `KeyPath` from its serialized JSONPath string."""
    del kwargs
    return cls.parse(json_value['value'])

path_str

path_str(preserve_complex_keys: bool = True) -> str

Returns JSONPath representation of current path.

Parameters:

Name	Type	Description	Default
preserve_complex_keys	bool	If True, complex keys such as 'x.y' are preserved by quoting them in brackets. For example, KeyPath(['a', 'x.y', 'b']) returns 'a[x.y].b' when preserve_complex_keys is True, and 'a.x.y.b' when preserve_complex_keys is False.	True

Returns:

Type	Description
str	Path string.

Source code in pygx/topology/_value_location.py

def path_str(self, preserve_complex_keys: bool = True) -> str:
    """Returns JSONPath representation of current path.

    Args:
      preserve_complex_keys: If True, complex keys such as ``'x.y'`` are
        preserved by quoting them in brackets. For example,
        ``KeyPath(['a', 'x.y', 'b'])`` returns ``'a[x.y].b'`` when
        `preserve_complex_keys` is True, and ``'a.x.y.b'`` when
        `preserve_complex_keys` is False.

    Returns:
      Path string.
    """
    keys = self._keys
    if not keys:
        return ''
    # Three rendering cases:
    #  - String key without delimiter chars (the symbolic-apply hot path):
    #    dot-delimited form, e.g. ".name".
    #  - StrKey instance: always dot-delimited (its `__str__` is the form).
    #  - Anything else (int, custom): bracket form, e.g. "[0]" or "[obj]".
    parts: list[str] = []
    for key in keys:
        if isinstance(key, str):
            if preserve_complex_keys and (
                '.' in key or '[' in key or ']' in key
            ):
                parts.append(f'[{key}]')
                continue
            if parts:
                parts.append('.')
            parts.append(key)
        elif isinstance(key, StrKey):
            if parts:
                parts.append('.')
            parts.append(str(key))
        else:
            parts.append(f'[{key}]')
    return ''.join(parts)

format

format(*args, **kwargs)

Format current path.

Source code in pygx/topology/_value_location.py

def format(self, *args, **kwargs):
    """Format current path."""
    return self.path

KeyPathSet

KeyPathSet(
    paths: Iterable[str | int | KeyPath] | None = None,
    *,
    include_intermediate: bool = False
)

Bases: Formattable

A set of KeyPath values based on a trie-like data structure.

Membership is exact: only added paths are members, while ancestors of added paths are queryable via has_prefix.

Example:

s = pg.KeyPathSet(['a.b', 'a.c[0]'])
assert 'a.b' in s
assert 'a' not in s       # Only added paths are members ...
assert s.has_prefix('a')  # ... but prefixes are queryable.
assert sorted(str(p) for p in s) == ['a.b', 'a.c[0]']

Source code in pygx/topology/_value_location.py

def __init__(
    self,
    paths: Iterable[str | int | KeyPath] | None = None,
    *,
    include_intermediate: bool = False,
):
    self._trie: dict[Any, Any] = {}
    if paths:
        for path in paths:
            self.add(path, include_intermediate=include_intermediate)

add

add(path: str | int | KeyPath, include_intermediate: bool = False) -> bool

Adds a path to the set.

Source code in pygx/topology/_value_location.py

def add(
    self,
    path: str | int | KeyPath,
    include_intermediate: bool = False,
) -> bool:
    """Adds a path to the set."""
    path = KeyPath.from_value(path)
    root = self._trie
    updated = False
    for key in path.keys:
        if key not in root:
            root[key] = {}
            if include_intermediate:
                root['$'] = True
                updated = True
        root = root[key]

    assert isinstance(root, dict), root
    if '$' not in root:
        root['$'] = True
        updated = True
    return updated

remove

remove(path: str | int | KeyPath) -> bool

Removes a path from the set.

Source code in pygx/topology/_value_location.py

def remove(self, path: str | int | KeyPath) -> bool:
    """Removes a path from the set."""
    path = KeyPath.from_value(path)
    stack = [self._trie]
    for key in path.keys:
        if key not in stack[-1]:
            return False
        value = stack[-1][key]
        assert isinstance(value, dict), value
        stack.append(value)

    if '$' in stack[-1]:
        stack[-1].pop('$')
        stack.pop(-1)
        assert len(stack) == len(path.keys), (path.keys, stack)
        for key, parent_node in zip(reversed(path.keys), reversed(stack)):
            if not parent_node[key]:
                del parent_node[key]
        return True
    return False

has_prefix

has_prefix(root_path: int | str | KeyPath) -> bool

Returns True if the set has a path with the given prefix.

Source code in pygx/topology/_value_location.py

def has_prefix(self, root_path: int | str | KeyPath) -> bool:
    """Returns True if the set has a path with the given prefix."""
    root_path = KeyPath.from_value(root_path)
    root = self._trie
    for key in root_path.keys:
        if key not in root:
            return False
        root = root[key]
    return True

rebase

rebase(root_path: int | str | KeyPath) -> None

Prepends the given prefix path to all paths in the set, in place.

Source code in pygx/topology/_value_location.py

def rebase(
    self,
    root_path: int | str | KeyPath,
) -> None:
    """Prepends the given prefix path to all paths in the set, in place."""
    root_path = KeyPath.from_value(root_path)
    root = self._trie
    for key in reversed(root_path.keys):
        root = {key: root}
    self._trie = root

clear

clear() -> None

Clears the set.

Source code in pygx/topology/_value_location.py

def clear(self) -> None:
    """Clears the set."""
    self._trie.clear()

copy

copy() -> Self

Returns a deep copy of the set.

Source code in pygx/topology/_value_location.py

def copy(self) -> Self:
    """Returns a deep copy of the set."""
    return copy_lib.deepcopy(self)

difference_update

difference_update(other: KeyPathSet) -> None

Removes the paths in the other set from the current set.

Source code in pygx/topology/_value_location.py

def difference_update(self, other: 'KeyPathSet') -> None:
    """Removes the paths in the other set from the current set."""

    def _remove_same(target_dict, src_dict):
        keys_to_remove = []
        for key, value in target_dict.items():
            if key in src_dict:
                if key == '$' or _remove_same(value, src_dict[key]):
                    keys_to_remove.append(key)
        for key in keys_to_remove:
            del target_dict[key]
        if not target_dict:
            return True
        return False

    _remove_same(self._trie, other._trie)  # pylint: disable=protected-access

difference

difference(other: KeyPathSet) -> Self

Returns a new set with the other set's paths removed.

Source code in pygx/topology/_value_location.py

def difference(
    self,
    other: 'KeyPathSet',
) -> Self:
    """Returns a new set with the other set's paths removed."""
    x = self.copy()
    x.difference_update(other)
    return x

intersection_update

intersection_update(other: KeyPathSet) -> None

Keeps only the paths that are also in the other set, in place.

Source code in pygx/topology/_value_location.py

def intersection_update(self, other: 'KeyPathSet') -> None:
    """Keeps only the paths that are also in the other set, in place."""

    def _remove_diff(target_dict, src_dict):
        keys_to_remove = []
        for key, value in target_dict.items():
            if key not in src_dict:
                keys_to_remove.append(key)
            elif key != '$':
                _remove_diff(value, src_dict[key])
                if not value:
                    keys_to_remove.append(key)
        for key in keys_to_remove:
            del target_dict[key]

    _remove_diff(self._trie, other._trie)  # pylint: disable=protected-access

intersection

intersection(other: KeyPathSet) -> Self

Returns the intersection KeyPathSet.

Source code in pygx/topology/_value_location.py

def intersection(self, other: 'KeyPathSet') -> Self:
    """Returns the intersection KeyPathSet."""
    copy = self.copy()
    copy.intersection_update(other)
    return copy

update

update(other: KeyPathSet) -> None

Updates the current set with the other set.

Source code in pygx/topology/_value_location.py

def update(self, other: 'KeyPathSet') -> None:
    """Updates the current set with the other set."""

    def _merge(target_dict, src_dict):
        for key, value in src_dict.items():
            if key != '$' and key in target_dict:
                _merge(target_dict[key], value)
            else:
                target_dict[key] = copy_lib.deepcopy(value)

    _merge(self._trie, other._trie)  # pylint: disable=protected-access

union

union(other: KeyPathSet, copy: bool = False) -> Self

Returns the union of this set and other.

Parameters:

Name	Type	Description	Default
other	KeyPathSet	The set whose paths are merged in.	required
copy	bool	If True, the union is computed on a copy and this set is left unchanged. If False (the default), other's paths are merged into THIS set, which is also the return value.	False

Returns:

Type	Description
Self	The union set — a new set when copy is True, otherwise self.

Source code in pygx/topology/_value_location.py

def union(self, other: 'KeyPathSet', copy: bool = False) -> Self:
    """Returns the union of this set and `other`.

    Args:
      other: The set whose paths are merged in.
      copy: If True, the union is computed on a copy and this set is
        left unchanged. If False (the default), `other`'s paths are
        merged into THIS set, which is also the return value.

    Returns:
      The union set — a new set when `copy` is True, otherwise `self`.
    """
    x = self.copy() if copy else self
    x.update(other)
    return x

subtree

subtree(root_path: int | str | KeyPath) -> Optional[KeyPathSet]

Returns the relative paths of the sub-tree rooted at the given path.

Parameters:

Name	Type	Description	Default
root_path	int | str | KeyPath	A KeyPath for the root of the sub-tree.	required

Returns:

Type	Description
Optional[KeyPathSet]	A KeyPathSet that contains all the child paths of the given root
Optional[KeyPathSet]	path. Please note that the returned value shares the same trie as
Optional[KeyPathSet]	the current value, so addition/removal of paths in the returned
Optional[KeyPathSet]	value will also affect the current value. If there is no child path
Optional[KeyPathSet]	under the given root path, None will be returned.

Source code in pygx/topology/_value_location.py

def subtree(
    self,
    root_path: int | str | KeyPath,
) -> Optional['KeyPathSet']:
    """Returns the relative paths of the sub-tree rooted at the given path.

    Args:
      root_path: A KeyPath for the root of the sub-tree.

    Returns:
      A KeyPathSet that contains all the child paths of the given root
      path. Please note that the returned value shares the same trie as
      the current value, so addition/removal of paths in the returned
      value will also affect the current value. If there is no child path
      under the given root path, None will be returned.
    """
    root_path = KeyPath.from_value(root_path)
    if not root_path:
        return self
    root = self._trie
    for key in root_path.keys:
        if key not in root:
            return None
        root = root[key]
    ret = KeyPathSet()
    ret._trie = root  # pylint: disable=protected-access
    return ret

format

format(*args, **kwargs) -> str

Formats the set.

Source code in pygx/topology/_value_location.py

def format(self, *args, **kwargs) -> str:
    """Formats the set."""
    return formatting.kvlist_str(
        [('', list(self), [])], label=self.__class__.__name__, **kwargs
    )

from_value classmethod

from_value(
    value: Union[Iterable[int | str | KeyPath], KeyPathSet],
    include_intermediate: bool = False,
)

Returns a KeyPathSet from a compatible value.

Source code in pygx/topology/_value_location.py

@classmethod
def from_value(
    cls,
    value: Union[Iterable[int | str | KeyPath], 'KeyPathSet'],
    include_intermediate: bool = False,
):
    """Returns a KeyPathSet from a compatible value."""
    if isinstance(value, KeyPathSet):
        return value
    if isinstance(value, (list, set, tuple)):
        return cls(value, include_intermediate=include_intermediate)
    raise ValueError(
        f'Cannot convert {value!r} to KeyPathSet. '
        f'Expected a list, set, tuple, or KeyPathSet.'
    )

StrKey

Interface for classes whose instances are treated as str in KeyPath.

A pygx.KeyPath will format the path string using . (dot) as the delimiter for a key represented by this object. Otherwise [] (square brackets) will be used as the delimiters.

Example:

class MyKey(pg.topology.StrKey):

  def __init__(self, name):
    self.name = name

  def __str__(self):
    return f'__{self.name}__'

path = pg.KeyPath(['a', MyKey('b')])
assert str(path) == 'a.__b__'

canonicalize

canonicalize(src: Any, sparse_list_as_dict: bool = True) -> Any

Canonicalize (maybe) non-canonical hierarchical value.

Non-canonical hierarchical values are dicts or nested structures of dicts that contain keys with '.' or '[]'. Canonicalization is to unfold '.' and '[]' in their keys into multi-level dicts. It is the reverse operation of flatten. If the input value is a simple type, the value itself is returned.

For example:

[1, {
  'a.b[0]': {
    'e.f': 1
  },
  'a.b[0].c[x.y].d': 10
}]

will result in:

[1, {
  'a': {
    'b': [{
      'e': {
        'f': 1
      },
      'c': {
        'x.y': {
          'd': 10
        }
      }
    }]
  }
}]

A sparse array indexer can be used in a non-canonical form, e.g:

{
  'a[1]': 123,
  'a[5]': 234
}

This is to accommodate scenarios of list element update/append. When sparse_list_as_dict is set to True (the default), the sparse indices are kept as keys of an int-keyed dict:

{
  'a': {
    1: 123,
    5: 234
  }
}

(Please note that a sparse indexer as key is not JSON serializable.) Otherwise the values are packed into a list ordered by index:

{
  'a': [123, 234]
}

Parameters:

Name	Type	Description	Default
src	Any	A simple type or a nested structure of dict that may contain keys with JSON paths like 'a.b.c'.	required
sparse_list_as_dict	bool	If True (the default), a group of entries with sparse list indices (e.g. 'a[1]', 'a[5]') canonicalizes to an int-keyed dict that keeps those indices; if False, it canonicalizes to a list with the values ordered by index. An index group that forms a perfect range(0, N) becomes a list either way.	True

Returns:

Type	Description
Any	A nested structure of dict that has only canonicalized keys, or src
Any	itself if it's not a nested structure of dict.

Raises:

Type	Description
KeyError	If key is empty or the same key yields conflicting values after resolving non-canonical paths. E.g: {'': 1} or {'a.b': 1, 'a.b.c': True}.

Source code in pygx/topology/_hierarchical.py

def canonicalize(src: Any, sparse_list_as_dict: bool = True) -> Any:
    """Canonicalize (maybe) non-canonical hierarchical value.

    Non-canonical hierarchical values are dicts or nested structures of dicts
    that contain keys with '.' or '[<number>]'. Canonicalization is to unfold
    '.' and '[]' in their keys into multi-level dicts. It is the reverse
    operation of `flatten`. If the input value is a simple type, the value
    itself is returned.

    For example:

    ```python
    [1, {
      'a.b[0]': {
        'e.f': 1
      },
      'a.b[0].c[x.y].d': 10
    }]
    ```

    will result in:

    ```python
    [1, {
      'a': {
        'b': [{
          'e': {
            'f': 1
          },
          'c': {
            'x.y': {
              'd': 10
            }
          }
        }]
      }
    }]
    ```

    A sparse array indexer can be used in a non-canonical form, e.g:

    ```python
    {
      'a[1]': 123,
      'a[5]': 234
    }
    ```

    This is to accommodate scenarios of list element update/append. When
    `sparse_list_as_dict` is set to True (the default), the sparse indices
    are kept as keys of an int-keyed dict:

    ```python
    {
      'a': {
        1: 123,
        5: 234
      }
    }
    ```

    (Please note that a sparse indexer as key is not JSON serializable.)
    Otherwise the values are packed into a list ordered by index:

    ```python
    {
      'a': [123, 234]
    }
    ```

    Args:
      src: A simple type or a nested structure of dict that may contain keys
        with JSON paths like 'a.b.c'.
      sparse_list_as_dict: If True (the default), a group of entries with
        sparse list indices (e.g. ``'a[1]'``, ``'a[5]'``) canonicalizes to an
        int-keyed dict that keeps those indices; if False, it canonicalizes
        to a list with the values ordered by index. An index group that forms
        a perfect ``range(0, N)`` becomes a list either way.

    Returns:
      A nested structure of dict that has only canonicalized keys, or `src`
      itself if it's not a nested structure of dict.

    Raises:
      KeyError: If key is empty or the same key yields conflicting values
        after resolving non-canonical paths. E.g: `{'': 1}` or
        `{'a.b': 1, 'a.b.c': True}`.
    """

    def _merge_fn(path, old_value, new_value):
        if old_value is not MISSING_VALUE and new_value is not MISSING_VALUE:
            raise KeyError(
                f"Path '{path}' is assigned with conflicting values. "
                f'Left value: {old_value}, Right value: {new_value}'
            )
        # Always merge.
        return new_value if new_value is not MISSING_VALUE else old_value

    if isinstance(src, dict):
        # We keep order of keys.
        canonical_dict = dict()

        # Make deterministic traversal of dict.
        for key, value in src.items():
            if isinstance(key, str):
                path = KeyPath.parse(key)
            else:
                path = KeyPath(key)

            if len(path) == 1:
                # Key is already canonical.
                # NOTE(daiyip): pass through sparse_list_as_dict to canonicalize
                # value to keep consistency with the container.
                new_value = canonicalize(value, sparse_list_as_dict)
                if path.key not in canonical_dict:
                    canonical_dict[path.key] = new_value
                else:
                    old_value = canonical_dict[path.key]
                    # merge dict is in-place.
                    merge_tree(old_value, new_value, _merge_fn)
            else:
                # Key is a path.
                if path.is_root:
                    raise KeyError(
                        f'Key must not be empty. Encountered: {src}.'
                    )
                sub_root = dict()
                cur_dict = sub_root
                for token in path.keys[:-1]:
                    cur_dict[token] = dict()
                    cur_dict = cur_dict[token]
                cur_dict[path.key] = canonicalize(value, sparse_list_as_dict)
                # merge dict is in-place.
                merge_tree(canonical_dict, sub_root, _merge_fn)

        # NOTE(daiyip): We restore the list form of integer-keyed dict
        # if its keys form a perfect range(0, N), unless sparse_list_as_dict is set.
        def _listify_dict_equivalent(p, v):
            del p
            if isinstance(v, dict):
                v = try_listify_dict_with_int_keys(v, not sparse_list_as_dict)[
                    0
                ]
            return v

        return transform(canonical_dict, _listify_dict_equivalent)
    elif isinstance(src, list):
        return [canonicalize(item, sparse_list_as_dict) for item in src]
    else:
        return src

flatten

flatten(src: Any, flatten_complex_keys: bool = True) -> Any

Flattens a (maybe) hierarchical value into a depth-1 dict.

Example:

inputs = {
    'a': {
        'e': 1,
        'f': [{
            'g': 2
        }, {
            'g[0]': 3
        }],
        'h': [],
        'i.j': {},
    },
    'b': 'hi',
    'c': None
}
output = pg.topology.flatten(inputs)
assert output == {
    'a.e': 1,
    'a.f[0].g': 2,
    'a.f[1].g[0]': 3,
    'a.h': [],
    'a.i.j': {},
    'b': 'hi',
    'c': None
}

Parameters:

Name	Type	Description	Default
src	Any	source value to flatten.	required
flatten_complex_keys	bool	if True, complex keys such as 'x.y' will be flattened as 'x'.'y'. For example: {'a': {'b.c': 1}} will be flattened into {'a.b.c': 1} if this flag is on, otherwise it will be flattened as {'a[b.c]': 1}.	True

Returns:

Type	Description
Any	For primitive value types, src itself will be returned.
Any	For list and dict types, an 1-depth dict will be returned.
Any	For tuple, a tuple of the same length, with each element flattened will be
Any	returned. The order of keys in nested ordered dict will be preserved,
Any	Keys of different depth are joined into a string using "." for dict
Any	properties and "[]" for list elements.
Any	For example, if src is:
Any	```python
Any	{ "a": { "b": 4, "c": {"d": 10}, "e": [1, 2], "f": [], "g.h": {}, }
Any	}
Any	```
Any	then the output dict is:
Any	```python
Any	{ "a.b": 4, "a.c.d": 10, "a.e[0]": 1, "a.e[1]": 2, "a.f": [], "a.g.h": {},
Any	}
Any	```
Any	when flatten_complex_keys is True, and:
Any	```python
Any	{ "a.b": 4, "a.c.d": 10, "a.e[0]": 1, "a.e[1]": 2, "a.f": [], "a[g.h]": {},
Any	}
Any	```
Any	when flatten_complex_keys is False.

Source code in pygx/topology/_hierarchical.py

def flatten(src: Any, flatten_complex_keys: bool = True) -> Any:
    """Flattens a (maybe) hierarchical value into a depth-1 dict.

    **Example:**

    ```python
    inputs = {
        'a': {
            'e': 1,
            'f': [{
                'g': 2
            }, {
                'g[0]': 3
            }],
            'h': [],
            'i.j': {},
        },
        'b': 'hi',
        'c': None
    }
    output = pg.topology.flatten(inputs)
    assert output == {
        'a.e': 1,
        'a.f[0].g': 2,
        'a.f[1].g[0]': 3,
        'a.h': [],
        'a.i.j': {},
        'b': 'hi',
        'c': None
    }
    ```

    Args:
      src: source value to flatten.
      flatten_complex_keys: if True, complex keys such as 'x.y' will be
        flattened as 'x'.'y'. For example: {'a': {'b.c': 1}} will be flattened
        into {'a.b.c': 1} if this flag is on, otherwise it will be flattened
        as {'a[b.c]': 1}.

    Returns:
      For primitive value types, `src` itself will be returned.
      For list and dict types, an 1-depth dict will be returned.
      For tuple, a tuple of the same length, with each element flattened will be
      returned. The order of keys in nested ordered dict will be preserved,
      Keys of different depth are joined into a string using "." for dict
      properties and "[]" for list elements.
      For example, if src is:

      ```python
      {
        "a": {
               "b": 4,
               "c": {"d": 10},
               "e": [1, 2],
               "f": [],
               "g.h": {},
             }
      }
      ```

      then the output dict is:

      ```python
      {
        "a.b": 4,
        "a.c.d": 10,
        "a.e[0]": 1,
        "a.e[1]": 2,
        "a.f": [],
        "a.g.h": {},
      }
      ```

      when `flatten_complex_keys` is True, and:

      ```python
      {
        "a.b": 4,
        "a.c.d": 10,
        "a.e[0]": 1,
        "a.e[1]": 2,
        "a.f": [],
        "a[g.h]": {},
      }
      ```

      when `flatten_complex_keys` is False.
    """
    # NOTE(daiyip): Comparing to list, tuple is treated as a single value,
    # whose index is not treated as key. That being said, there is no partial
    # update semantics on elements of tuple in `merge` method too.
    # Thus we simply flatten its elements and keep the tuple form.
    if isinstance(src, tuple):
        return tuple([flatten(elem) for elem in src])

    if not isinstance(src, (dict, list)) or not src:
        return src

    dest = dict()

    def _output_leaf(path: KeyPath, value: Any):
        if path and (not isinstance(value, (dict, list)) or not value):
            dest[path.path_str(not flatten_complex_keys)] = value
        return True

    traverse(src, postorder_visitor_fn=_output_leaf)
    return dest

is_partial

is_partial(value: Any) -> bool

Returns True if a value is partially bound.

A value is partially bound when it is (or contains, at any depth) pg.MISSING_VALUE or a partial MaybePartial value.

Source code in pygx/topology/_hierarchical.py

def is_partial(value: Any) -> bool:
    """Returns True if a value is partially bound.

    A value is partially bound when it is (or contains, at any depth)
    ``pg.MISSING_VALUE`` or a partial `MaybePartial` value.
    """

    def _check_full_bound(path: KeyPath, value: Any) -> bool:
        del path
        if MISSING_VALUE == value:
            return False
        elif isinstance(value, maybe_partial.MaybePartial) and not isinstance(
            value, (dict, list)
        ):
            return not value.sym_partial
        return True

    return not traverse(value, _check_full_bound)

merge

merge(
    value_list: list[Any],
    merge_fn: Callable[[KeyPath, Any, Any], Any] | None = None,
) -> Any

Merge a list of hierarchical values.

Example:

original = {
    'a': 1,
    'b': 2,
    'c': {
        'd': 'foo',
        'e': 'bar'
    }
}
patch =  {
    'b': 3,
    'd': [1, 2, 3],
    'c': {
        'e': 'bar2',
        'f': 10
    }
}
output = pg.topology.merge([original, patch])
assert output == {
    'a': 1,
    # b is updated.
    'b': 3,
    'c': {
        'd': 'foo',
        # e is updated.
        'e': 'bar2',
        # f is added.
        'f': 10
    },
    # d is inserted.
    'd': [1, 2, 3]
}

Parameters:

Name	Type	Description	Default
value_list	list[Any]	A list of hierarchical values to merge. Later value will be treated as updates if it's a dict or otherwise a replacement of former value. The merge process will keep input values intact.	required
merge_fn	Callable[[KeyPath, Any, Any], Any] | None	A function to handle value merge that will be called for updated or added keys. If a branch is added/updated, the root of branch will be passed to merge_fn. The signature of the function is (path, left_value, right_value) -> final_value. If a key is only present in the later value, left_value is MISSING_VALUE; if a key is only present in the earlier value, right_value is MISSING_VALUE; otherwise both are filled. If final_value is MISSING_VALUE for a path, it will be removed from its parent collection.	None

Returns:

Type	Description
Any	A merged value.

Raises:

Type	Description
TypeError	If value_list is not a list.
KeyError	If a non-canonical key resolves to conflicting values, or a dict with non-integer keys is merged into a list.

Source code in pygx/topology/_hierarchical.py

def merge(
    value_list: list[Any],
    merge_fn: Callable[[KeyPath, Any, Any], Any] | None = None,
) -> Any:
    """Merge a list of hierarchical values.

    **Example:**

    ```python
    original = {
        'a': 1,
        'b': 2,
        'c': {
            'd': 'foo',
            'e': 'bar'
        }
    }
    patch =  {
        'b': 3,
        'd': [1, 2, 3],
        'c': {
            'e': 'bar2',
            'f': 10
        }
    }
    output = pg.topology.merge([original, patch])
    assert output == {
        'a': 1,
        # b is updated.
        'b': 3,
        'c': {
            'd': 'foo',
            # e is updated.
            'e': 'bar2',
            # f is added.
            'f': 10
        },
        # d is inserted.
        'd': [1, 2, 3]
    }
    ```

    Args:
      value_list: A list of hierarchical values to merge. Later value will be
        treated as updates if it's a dict or otherwise a replacement of former
        value. The merge process will keep input values intact.
      merge_fn: A function to handle value merge that will be called for updated
        or added keys. If a branch is added/updated, the root of branch will be
        passed to merge_fn. The signature of the function is ``(path,
        left_value, right_value) -> final_value``. If a key is only present in
        the later value, left_value is MISSING_VALUE; if a key is only present
        in the earlier value, right_value is MISSING_VALUE; otherwise both are
        filled. If final_value is MISSING_VALUE for a path, it will be removed
        from its parent collection.

    Returns:
      A merged value.

    Raises:
      TypeError: If `value_list` is not a list.
      KeyError: If a non-canonical key resolves to conflicting values, or a
        dict with non-integer keys is merged into a list.
    """
    if not isinstance(value_list, list):
        raise TypeError('value_list should be a list')

    if not value_list:
        return None

    new_value = canonicalize(value_list[0], sparse_list_as_dict=True)
    for value in value_list[1:]:
        if value is None:
            continue
        new_value = merge_tree(
            new_value, canonicalize(value, sparse_list_as_dict=True), merge_fn
        )

    def _listify_dict_equivalent(p, v):
        del p
        if isinstance(v, dict):
            v = try_listify_dict_with_int_keys(v, True)[0]
        return v

    return transform(new_value, _listify_dict_equivalent)

merge_tree

merge_tree(
    dest: Any,
    src: Any,
    merge_fn: Callable[[KeyPath, Any, Any], Any] | None = None,
    root_path: KeyPath | None = None,
) -> Any

Deep merge two (maybe) hierarchical values.

Parameters:

Name	Type	Description	Default
dest	Any	Destination value.	required
src	Any	Source value. When source value is a dict, it's considered as a patch (delta) to the destination when destination is a dict or list. For other source types, it's considered as a new value that will replace dest completely.	required
merge_fn	Callable[[KeyPath, Any, Any], Any] | None	A function to handle value merge that will be called for updated or added keys. If a branch is added/updated, the root of branch will be passed to merge_fn. The signature of the function is (path, left_value, right_value) -> final_value. If a key is only present in src dict, left_value is MISSING_VALUE. If a key is only present in dest dict, right_value is MISSING_VALUE. Otherwise both left_value and right_value are filled. If final_value is MISSING_VALUE, it will be removed from its parent collection.	None
root_path	KeyPath | None	KeyPath of dest.	None

Returns:

Type	Description
Any	Merged value.

Raises:

Type	Description
KeyError	Dict keys are not integers when merging into a list.

Source code in pygx/topology/_hierarchical.py

def merge_tree(
    dest: Any,
    src: Any,
    merge_fn: Callable[[KeyPath, Any, Any], Any] | None = None,
    root_path: KeyPath | None = None,
) -> Any:
    """Deep merge two (maybe) hierarchical values.

    Args:
      dest: Destination value.
      src: Source value. When source value is a dict, it's considered as a
        patch (delta) to the destination when destination is a dict or list.
        For other source types, it's considered as a new value that will replace
        dest completely.
      merge_fn: A function to handle value merge that will be called for updated
        or added keys. If a branch is added/updated, the root of branch will be
        passed to merge_fn. The signature of the function is
        ``(path, left_value, right_value) -> final_value``.
        If a key is only present in src dict, left_value is MISSING_VALUE.
        If a key is only present in dest dict, right_value is MISSING_VALUE.
        Otherwise both left_value and right_value are filled.
        If final_value is MISSING_VALUE, it will be removed from its parent
        collection.
      root_path: KeyPath of dest.

    Returns:
      Merged value.

    Raises:
      KeyError: Dict keys are not integers when merging into a list.
    """
    if not root_path:
        root_path = KeyPath()

    if isinstance(dest, dict) and isinstance(src, dict):
        # Merge dict into dict.
        return _merge_dict_into_dict(dest, src, merge_fn, root_path)

    if isinstance(dest, list) and isinstance(src, dict):
        # Merge (possible) sparse indexed list into a list.
        return _merge_dict_into_list(dest, src, root_path)

    # Merge at root level.
    if merge_fn:
        return merge_fn(root_path, dest, src)
    return src

transform

transform(
    value: Any,
    transform_fn: Callable[[KeyPath, Any], Any],
    root_path: KeyPath | None = None,
    inplace: bool = True,
) -> Any

Bottom-up (post-order) transform a (maybe) hierarchical value.

Transform on value is in-place unless transform_fn returns a different instance.

Example:

def _remove_int(path, value):
  del path
  if isinstance(value, int):
    return pg.MISSING_VALUE
  return value

inputs = {
    'a': {
        'b': 1,
        'c': [1, 'bar', 2, 3],
        'd': 'foo'
    },
    'e': 'bar',
    'f': 4
}
output = pg.topology.transform(inputs, _remove_int)
assert output == {
    'a': {
        'c': ['bar'],
        'd': 'foo',
    },
    'e': 'bar'
}

Parameters:

Name	Type	Description	Default
value	Any	Any python value type. If value is a list of dict, transformation will occur recursively.	required
transform_fn	Callable[[KeyPath, Any], Any]	Transform function in signature (path, value) -> new value. If new value is MISSING_VALUE, key will be deleted.	required
root_path	KeyPath | None	KeyPath of the root.	None
inplace	bool	If True, perform transformation in place.	True

Returns:

Type	Description
Any	Transformed value.

Source code in pygx/topology/_hierarchical.py

def transform(
    value: Any,
    transform_fn: Callable[[KeyPath, Any], Any],
    root_path: KeyPath | None = None,
    inplace: bool = True,
) -> Any:
    """Bottom-up (post-order) transform a (maybe) hierarchical value.

    Transform on value is in-place unless `transform_fn` returns a different
    instance.

    **Example:**

    ```python
    def _remove_int(path, value):
      del path
      if isinstance(value, int):
        return pg.MISSING_VALUE
      return value

    inputs = {
        'a': {
            'b': 1,
            'c': [1, 'bar', 2, 3],
            'd': 'foo'
        },
        'e': 'bar',
        'f': 4
    }
    output = pg.topology.transform(inputs, _remove_int)
    assert output == {
        'a': {
            'c': ['bar'],
            'd': 'foo',
        },
        'e': 'bar'
    }
    ```

    Args:
      value: Any python value type. If value is a list of dict, transformation
        will occur recursively.
      transform_fn: Transform function in signature (path, value) -> new value.
        If new value is MISSING_VALUE, key will be deleted.
      root_path: KeyPath of the root.
      inplace: If True, perform transformation in place.

    Returns:
      Transformed value.
    """

    def _transform(value: Any, current_path: KeyPath) -> Any:
        """Implementation of transform function."""
        new_value = value
        if isinstance(value, dict):
            if not inplace:
                new_value = value.__class__()
            deleted_keys = []
            for k, v in value.items():
                nv = _transform(v, KeyPath(k, current_path))
                if MISSING_VALUE == nv and inplace:
                    deleted_keys.append(k)
                elif MISSING_VALUE != nv and (
                    not inplace or value[k] is not nv
                ):
                    new_value[k] = nv

            for k in deleted_keys:
                del value[k]
        elif isinstance(value, list):
            deleted_indices = []
            if not inplace:
                new_value = value.__class__()
            for i, v in enumerate(value):
                nv = _transform(v, KeyPath(i, current_path))
                if MISSING_VALUE == nv and inplace:
                    deleted_indices.append(i)
                elif MISSING_VALUE != nv and not inplace:
                    new_value.append(nv)
                elif MISSING_VALUE != nv and value[i] is not nv:
                    value[i] = nv
            for i in reversed(deleted_indices):
                del value[i]
        return transform_fn(current_path, new_value)

    return _transform(value, root_path or KeyPath())

traverse

traverse(
    value: Any,
    preorder_visitor_fn: Callable[[KeyPath, Any], bool] | None = None,
    postorder_visitor_fn: Callable[[KeyPath, Any], bool] | None = None,
    root_path: KeyPath | None = None,
) -> bool

Traverse a (maybe) hierarchical value.

Example:

def preorder_visit(path, value):
  print(path)
  return True

tree = {'a': [{'c': [1, 2]}, {'d': {'g': (3, 4)}}], 'b': 'foo'}
pg.topology.traverse(tree, preorder_visit)

# Should print (the first line is the empty root path):
#
# a
# a[0]
# a[0].c
# a[0].c[0]
# a[0].c[1]
# a[1]
# a[1].d
# a[1].d.g
# b

Parameters:

Name	Type	Description	Default
value	Any	A maybe hierarchical value to traverse.	required
preorder_visitor_fn	Callable[[KeyPath, Any], bool] | None	Preorder visitor function. Function signature is (path, value) -> should_continue.	None
postorder_visitor_fn	Callable[[KeyPath, Any], bool] | None	Postorder visitor function. Function signature is (path, value) -> should_continue.	None
root_path	KeyPath | None	The key path of the root value.	None

Returns:

Type	Description
bool	Whether visitor function returns True on all nodes.

Source code in pygx/topology/_hierarchical.py

def traverse(
    value: Any,
    preorder_visitor_fn: Callable[[KeyPath, Any], bool] | None = None,
    postorder_visitor_fn: Callable[[KeyPath, Any], bool] | None = None,
    root_path: KeyPath | None = None,
) -> bool:
    """Traverse a (maybe) hierarchical value.

    **Example:**

    ```python
    def preorder_visit(path, value):
      print(path)
      return True

    tree = {'a': [{'c': [1, 2]}, {'d': {'g': (3, 4)}}], 'b': 'foo'}
    pg.topology.traverse(tree, preorder_visit)

    # Should print (the first line is the empty root path):
    #
    # a
    # a[0]
    # a[0].c
    # a[0].c[0]
    # a[0].c[1]
    # a[1]
    # a[1].d
    # a[1].d.g
    # b
    ```

    Args:
      value: A maybe hierarchical value to traverse.
      preorder_visitor_fn: Preorder visitor function. Function signature is
        (path, value) -> should_continue.
      postorder_visitor_fn: Postorder visitor function. Function signature is
        (path, value) -> should_continue.
      root_path: The key path of the root value.

    Returns:
      Whether visitor function returns True on all nodes.
    """
    root_path = root_path or KeyPath()

    def no_op_visitor(key, value):
        del key, value
        return True

    if preorder_visitor_fn is None:
        preorder_visitor_fn = no_op_visitor
    if postorder_visitor_fn is None:
        postorder_visitor_fn = no_op_visitor

    if not preorder_visitor_fn(root_path, value):
        return False
    if isinstance(value, dict):
        for k in value.keys():
            if not traverse(
                value[k],
                preorder_visitor_fn,
                postorder_visitor_fn,
                KeyPath(k, root_path),
            ):
                return False
    elif isinstance(value, list):
        for i, v in enumerate(value):
            if not traverse(
                v,
                preorder_visitor_fn,
                postorder_visitor_fn,
                KeyPath(i, root_path),
            ):
                return False
    if not postorder_visitor_fn(root_path, value):
        return False
    return True

try_listify_dict_with_int_keys

try_listify_dict_with_int_keys(
    src: dict[Any, Any], convert_when_sparse: bool = False
) -> tuple[list[Any] | dict[Any, Any], bool]

Try to convert a dictionary with consecutive integer keys to a list.

Parameters:

Name	Type	Description	Default
src	dict[Any, Any]	A dict to maybe convert. Conversion happens only when all keys are ints and they form a perfect range(0, N), unless convert_when_sparse is set to True.	required
convert_when_sparse	bool	When src is an int-keyed dict, force converting it to a list ordered by key, even if it's sparse.	False

Returns:

Type	Description
list[Any] | dict[Any, Any]	A (value, converted) tuple: the converted list (or src unchanged)
bool	and whether conversion happened.

Source code in pygx/topology/_hierarchical.py

def try_listify_dict_with_int_keys(
    src: dict[Any, Any], convert_when_sparse: bool = False
) -> tuple[list[Any] | dict[Any, Any], bool]:
    """Try to convert a dictionary with consecutive integer keys to a list.

    Args:
      src: A dict to maybe convert. Conversion happens only when all keys are
        ints and they form a perfect ``range(0, N)``, unless
        `convert_when_sparse` is set to True.
      convert_when_sparse: When `src` is an int-keyed dict, force converting
        it to a list ordered by key, even if it's sparse.

    Returns:
      A ``(value, converted)`` tuple: the converted list (or `src` unchanged)
      and whether conversion happened.
    """
    if not src:
        return (src, False)

    min_key = None
    max_key = None
    for key in src.keys():
        if not isinstance(key, int):
            return (src, False)
        if min_key is None or min_key > key:
            min_key = key
        if max_key is None or max_key < key:
            max_key = key
    if convert_when_sparse or (min_key == 0 and max_key == len(src) - 1):
        return ([src[key] for key in sorted(src.keys())], True)
    return (src, False)

message_on_path

message_on_path(message: str, path: KeyPath | None) -> str

Formats a message that is associated with a KeyPath.

Appends ' (path=<path>)' to the message; returns the message unchanged when path is None.

Source code in pygx/topology/_value_location.py

def message_on_path(message: str, path: KeyPath | None) -> str:
    """Formats a message that is associated with a `KeyPath`.

    Appends ``' (path=<path>)'`` to the message; returns the message
    unchanged when `path` is None.
    """
    if path is None:
        return message
    return f'{message} (path={path})'

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2