Class _NodePathBuilder

src/google/adk/events/_node_path_builder.py:20–107

Represents a path to a node in a hierarchical workflow. A node path is a sequence of segments, each identifying a node instance, typically in the form 'node_name@run_id' or just 'node_name'.

class _NodePathBuilder:
  """Represents a path to a node in a hierarchical workflow.

  A node path is a sequence of segments, each identifying a node instance,
  typically in the form 'node_name@run_id' or just 'node_name'.
  """

  def __init__(self, segments: list[str]):
    """Initializes a _NodePathBuilder with a list of segments."""
    self._segments = segments

  @classmethod
  def from_string(cls, path_str: str) -> _NodePathBuilder:
    """Parses a _NodePathBuilder from a string representation.

    Example: 'wf@1/node@2'.
    """
    if not path_str:
      return cls([])
    return cls(path_str.split('/'))

  def __str__(self) -> str:
    """Returns the string representation of the path."""
    return '/'.join(self._segments)

  def __eq__(self, other: object) -> bool:
    """Returns True if segments are equal."""
    if not isinstance(other, _NodePathBuilder):
      return NotImplemented
    return self._segments == other._segments

  @property
  def node_name(self) -> str:
    """Returns the node name of the leaf segment."""
    if not self._segments:
      return ''
    return self._segments[-1].rsplit('@', 1)[0]

  @property
  def leaf_segment(self) -> str:
    """Returns the full leaf segment."""
    if not self._segments:
      return ''
    return self._segments[-1]

  @property
  def run_id(self) -> str | None:
    """Returns the run ID of the leaf segment, if any."""
    if not self._segments:
      return None
    parts = self._segments[-1].rsplit('@', 1)
    return parts[1] if len(parts) > 1 else None

  @property
  def parent(self) -> _NodePathBuilder | None:
    """Returns the parent _NodePathBuilder, or None if this is a root path."""
    if len(self._segments) <= 1:
      return None