Skip to content

pydantic_evals.otel

SpanQuery

Bases: TypedDict

A serializable query for filtering SpanNodes based on various conditions.

All fields are optional and combined with AND logic by default.

Attributes

stop_recursing_when

If present, stop recursing through ancestors or descendants at nodes that match this condition.

Type: SpanQuery

SpanNode

A node in the span tree; provides references to parents/children for easy traversal and queries.

Attributes

duration

Return the span’s duration as a timedelta, or None if start/end not set.

Type: timedelta

descendants

Return all descendants of this node in DFS order.

Type: list[SpanNode]

ancestors

Return all ancestors of this node.

Type: list[SpanNode]

Methods

add_child
def add_child(child: SpanNode) -> None

Attach a child node to this node’s list of children.

Returns

None

find_children
def find_children(predicate: SpanQuery | SpanPredicate) -> list[SpanNode]

Return all immediate children that satisfy the given predicate.

Returns

list[SpanNode]

first_child
def first_child(predicate: SpanQuery | SpanPredicate) -> SpanNode | None

Return the first immediate child that satisfies the given predicate, or None if none match.

Returns

SpanNode | None

any_child
def any_child(predicate: SpanQuery | SpanPredicate) -> bool

Returns True if there is at least one child that satisfies the predicate.

Returns

bool

find_descendants
def find_descendants(
    predicate: SpanQuery | SpanPredicate,
    stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> list[SpanNode]

Return all descendant nodes that satisfy the given predicate in DFS order.

Returns

list[SpanNode]

first_descendant
def first_descendant(
    predicate: SpanQuery | SpanPredicate,
    stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> SpanNode | None

DFS: Return the first descendant (in DFS order) that satisfies the given predicate, or None if none match.

Returns

SpanNode | None

any_descendant
def any_descendant(
    predicate: SpanQuery | SpanPredicate,
    stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> bool

Returns True if there is at least one descendant that satisfies the predicate.

Returns

bool

find_ancestors
def find_ancestors(
    predicate: SpanQuery | SpanPredicate,
    stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> list[SpanNode]

Return all ancestors that satisfy the given predicate.

Returns

list[SpanNode]

first_ancestor
def first_ancestor(
    predicate: SpanQuery | SpanPredicate,
    stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> SpanNode | None

Return the closest ancestor that satisfies the given predicate, or None if none match.

Returns

SpanNode | None

any_ancestor
def any_ancestor(
    predicate: SpanQuery | SpanPredicate,
    stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> bool

Returns True if any ancestor satisfies the predicate.

Returns

bool

matches
def matches(query: SpanQuery | SpanPredicate) -> bool

Check if the span node matches the query conditions or predicate.

Returns

bool

repr_xml
def repr_xml(
    include_children: bool = True,
    include_trace_id: bool = False,
    include_span_id: bool = False,
    include_start_timestamp: bool = False,
    include_duration: bool = False,
) -> str

Return an XML-like string representation of the node.

Optionally includes children, trace_id, span_id, start_timestamp, and duration.

Returns

str

SpanTree

A container that builds a hierarchy of SpanNode objects from a list of finished spans.

You can then search or iterate the tree to make your assertions (using DFS for traversal).

Methods

add_spans
def add_spans(spans: list[SpanNode]) -> None

Add a list of spans to the tree, rebuilding the tree structure.

Returns

None

find
def find(predicate: SpanQuery | SpanPredicate) -> list[SpanNode]

Find all nodes in the entire tree that match the predicate, scanning from each root in DFS order.

Returns

list[SpanNode]

first
def first(predicate: SpanQuery | SpanPredicate) -> SpanNode | None

Find the first node that matches a predicate, scanning from each root in DFS order. Returns None if not found.

Returns

SpanNode | None

any
def any(predicate: SpanQuery | SpanPredicate) -> bool

Returns True if any node in the tree matches the predicate.

Returns

bool

__iter__
def __iter__() -> Iterator[SpanNode]

Return an iterator over all nodes in the tree.

Returns

Iterator[SpanNode]

repr_xml
def repr_xml(
    include_children: bool = True,
    include_trace_id: bool = False,
    include_span_id: bool = False,
    include_start_timestamp: bool = False,
    include_duration: bool = False,
) -> str

Return an XML-like string representation of the tree, optionally including children, trace_id, span_id, duration, and timestamps.

Returns

str