pygx.topology

Tree topology and JSON serialization primitives shared across PyGX.

topology

Tree topology and JSON serialization primitives shared across PyGX.

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

• KeyPath / KeyPathSet address nodes inside a nested structure, 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.
• 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.

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) implements this interface, as their symbolic attributes can be partially filled.

Example::

d = pg.Dict(x=pg.MISSING_VALUE, y=1) assert d.sym_partial assert 'x' in d.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. The subclass can override this method 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.

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.sym_path) within its symbolic tree. For example::

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

class B(pg.Object): z: A

a = A(x=1, y='foo') b = B(z=a) assert a.sym_path == 'z' # The path to object a is 'z'. assert b.sym_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.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::

:= | {[.]} := ['[('|)']'] := := := '[' | ']' | '.'

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 that contains three dict keys: 'a', 'x.y', 'b'. Since 'x.y' has delimiter characters, it needs to be enclosed in brackets.

TODO(daiyip): Support paring KeyPath from keys of complex types. Now this method only supports parsing KeyPath of string and int keys. That being said, format/parse are not symmetric, while format can convert a KeyPath that includes complex keys into a string, parse is not able to convert them 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::

      <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 that contains three dict keys: 'a', 'x.y', 'b'.
                         Since 'x.y' has delimiter characters, it needs to be
                         enclosed in brackets.

    TODO(daiyip): Support paring ``KeyPath`` from keys of complex types.
    Now this method only supports parsing KeyPath of string and int keys.
    That being said, ``format``/``parse`` are not symmetric, while ``format``
    can convert a ``KeyPath`` that includes complex keys into a string,
    ``parse`` is not able to convert them 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.
RuntimeError	Called on a KeyPath that is considered as removed.

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::

      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.
      RuntimeError: Called on a KeyPath that is considered as removed.
    """
    return self._query(0, src, use_inferred)

get

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

Gets the value of current path from an object with a default value.

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 of current path from an object with a default value."""
    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 KeyPath set based on trie-like data structure.

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

Returns a KeyPathSet with the given prefix path added.

Source code in pygx/topology/_value_location.py

def rebase(
    self,
    root_path: int | str | KeyPath,
) -> None:
    """Returns a KeyPathSet with the given prefix path added."""
    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 the subset KeyPathSet based on a prefix path.

Source code in pygx/topology/_value_location.py

def difference(
    self,
    other: 'KeyPathSet',
) -> Self:
    """Returns the subset KeyPathSet based on a prefix path."""
    x = self.copy()
    x.difference_update(other)
    return x

intersection_update

intersection_update(other: KeyPathSet) -> None

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

Source code in pygx/topology/_value_location.py

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

    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 KeyPathSet.

Source code in pygx/topology/_value_location.py

def union(self, other: 'KeyPathSet', copy: bool = False) -> Self:
    """Returns the union KeyPathSet."""
    x = self.copy()
    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 path.
Optional[KeyPathSet]	Please note that the returned value share the same trie as the current
Optional[KeyPathSet]	value. So addition/removal of paths in the returned value will also
Optional[KeyPathSet]	affect the current value. If there is no child path under the given root
Optional[KeyPathSet]	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 share 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 can be 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.utils.StrKey):

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

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

path = pg.KeyPath(['a', MyKey('b')]) print(str(path)) # Should print "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 ('.' or '[]') into multi-level dicts.

For example::

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

will result in::

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

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 (by default), dict above will be converted to::

{ 'a': [123, 234] }

Otherwise sparse indexer will be kept so the container type will remain as a dict::

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

(Please note that sparse indexer as key is not JSON serializable.)

This is the reverse operation of method flatten. If input value is a simple type, the value itself will be returned.

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	Whether to convert sparse list to dict. When this is set to True, indices specified in the key path will be kept. Otherwise, a list will be returned with elements ordered by indices in the path.	True

Returns:

Type	Description
Any	A nested structure of ordered dict that has only canonicalized keys or
Any	src itself if it's not a nested structure of dict. For dict of int keys
Any	whose values form a perfect range(0, N) will be returned as a list.

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 ('.' or '[]') into multi-level dicts.

      For example::

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

      will result in::

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

     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 (by default), dict above will be
     converted to::

       {
         'a': [123, 234]
       }

     Otherwise sparse indexer will be kept so the container type will
     remain as a dict::

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

     (Please note that sparse indexer as key is not JSON serializable.)

     This is the reverse operation of method flatten.
     If input value is a simple type, the value itself will be returned.

    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: Whether to convert sparse list to dict.
          When this is set to True, indices specified in the key path will be
          kept. Otherwise, a list will be returned with elements ordered by
          indices in the path.

    Returns:
       A nested structure of ordered dict that has only canonicalized keys or
       src itself if it's not a nested structure of dict. For dict of int keys
       whose values form a perfect range(0, N) will be returned as a list.

    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.utils.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:: { "a": { "b": 4, "c": {"d": 10}, "e": [1, 2] 'f': [], 'g.h': {}, } }
Any	then the output dict is:: { "a.b": 4, "a.c.d": 10, "a.e[0]": 1, "a.e[1]": 2, "a.f": [], "a.g.h": {}, }
Any	when flatten_complex_keys is True, and:: { "a.b": 4, "a.c.d": 10, "a.e[0]": 1, "a.e[1]": 2, "a.f": [], "a.[g.h]": {}, }
Any	when flatten_complex_keys is False.

Raises:

Type	Description
ValueError	If any key from the nested dictionary contains ".".

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::

      inputs = {
          'a': {
              'e': 1,
              'f': [{
                  'g': 2
              }, {
                  'g[0]': 3
              }],
              'h': [],
              'i.j': {},
          },
          'b': 'hi',
          'c': None
      }
      output = pg.utils.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::

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

      then the output dict is::

        {
          "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::

        {
          "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.

    Raises:
      ValueError: If any key from the nested dictionary contains ".".
    """
    # 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.

Source code in pygx/topology/_hierarchical.py

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

    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.utils.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 function is: (path, left_value, right_value) -> final_value If a key is only present in src dict, old_value is MISSING_VALUE; If a key is only present in dest dict, new_value is MISSING_VALUE; otherwise both new_value and old_value 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 new key is found while not allowed.

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::

      original = {
          'a': 1,
          'b': 2,
          'c': {
              'd': 'foo',
              'e': 'bar'
          }
      }
      patch =  {
          'b': 3,
          'd': [1, 2, 3],
          'c': {
              'e': 'bar2',
              'f': 10
          }
      }
      output = pg.utils.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 function is: `(path, left_value,
        right_value) -> final_value` If a key is only present in src dict,
        old_value is MISSING_VALUE; If a key is only present in dest dict,
        new_value is MISSING_VALUE; otherwise both new_value and old_value 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 new key is found while not allowed.
    """
    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 function is: (path, left_value, right_value) -> final_value If a key is only present in src dict, old_value is MISSING_VALUE. If a key is only present in dest dict, new_value is MISSING_VALUE. Otherwise both new_value and old_value are filled. If final value is MISSING_VALUE, it will be removed from its parent collection.	None

root_path: KeyPath of dest.

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 function is: (path, left_value, right_value) ->
          final_value
          If a key is only present in src dict, old_value is MISSING_VALUE.
          If a key is only present in dest dict, new_value is MISSING_VALUE.
          Otherwise both new_value and old_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.utils.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::

      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.utils.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)

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

# Should print: # '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::

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

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

      # Should print:
      # '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 consequentive integer keys to a list.

Parameters:

Name	Type	Description	Default
src	dict[Any, Any]	A dict whose keys may be int type and their range form a perfect range(0, N) list unless convert_when_sparse is set to True.	required
convert_when_sparse	bool	When src is a int-key dict, force convert it to a list ordered by key, even it's sparse.	False

Returns:

Type	Description
tuple[list[Any] | dict[Any, Any], bool]	converted list or src unchanged.

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 consequentive integer keys to a list.

    Args:
      src: A dict whose keys may be int type and their range form a perfect
        range(0, N) list unless convert_when_sparse is set to True.
      convert_when_sparse: When src is a int-key dict, force convert
        it to a list ordered by key, even it's sparse.

    Returns:
      converted list or src unchanged.
    """
    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.

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`."""
    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