pydantic_evals.otel
Bases: Exception
An exception that is used to provide the reason why a SpanTree was not recorded by context_subtree.
This may be due to missing dependencies, a tracer provider not having been set, or a custom TracerProvider
that does not support add_span_processor.
@classmethod
def __get_pydantic_core_schema__(cls, _: Any, __: Any) -> core_schema.CoreSchema
Pydantic core schema to allow SpanTreeRecordingError to be (de)serialized.
Only the human-readable message is preserved by design: the exception’s __context__,
__cause__, and traceback (e.g. the underlying ImportError chained in context_subtree) are
dropped on serialization and not reconstructed on the way back.
core_schema.CoreSchema
Bases: TypedDict
A serializable query for filtering SpanNodes based on various conditions.
All fields are optional and combined with AND logic by default.
If present, stop recursing through ancestors or descendants at nodes that match this condition.
Type: SpanQuery
A node in the span tree; provides references to parents/children for easy traversal and queries.
The span’s status; 'error' if the operation the span represents raised an exception.
Type: SpanStatus Default: 'unset'
Return the span’s duration as a timedelta, or None if start/end not set.
Type: timedelta
Return all descendants of this node in DFS order.
Type: list[SpanNode]
Return all ancestors of this node.
Type: list[SpanNode]
def add_child(child: SpanNode) -> None
Attach a child node to this node’s list of children.
def find_children(predicate: SpanQuery | SpanPredicate) -> list[SpanNode]
Return all immediate children that satisfy the given predicate.
list[SpanNode]
def first_child(predicate: SpanQuery | SpanPredicate) -> SpanNode | None
Return the first immediate child that satisfies the given predicate, or None if none match.
SpanNode | None
def any_child(predicate: SpanQuery | SpanPredicate) -> bool
Returns True if there is at least one child that satisfies the predicate.
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.
list[SpanNode]
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.
SpanNode | None
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.
def find_ancestors(
predicate: SpanQuery | SpanPredicate,
stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> list[SpanNode]
Return all ancestors that satisfy the given predicate.
list[SpanNode]
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.
SpanNode | None
def any_ancestor(
predicate: SpanQuery | SpanPredicate,
stop_recursing_when: SpanQuery | SpanPredicate | None = None,
) -> bool
Returns True if any ancestor satisfies the predicate.
def matches(query: SpanQuery | SpanPredicate) -> bool
Check if the span node matches the query conditions or predicate.
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.
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).
def add_spans(spans: list[SpanNode]) -> None
Add a list of spans to the tree, rebuilding the tree structure.
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.
list[SpanNode]
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.
SpanNode | None
def any(predicate: SpanQuery | SpanPredicate) -> bool
Returns True if any node in the tree matches the predicate.
def __iter__() -> Iterator[SpanNode]
Return an iterator over all nodes in the tree.
Iterator[SpanNode]
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.
The status of a span, mirroring opentelemetry.trace.StatusCode.
Default: Literal['unset', 'ok', 'error']