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.
Attribute values are compared with equality; a dict or list value also matches an attribute stored as its JSON serialization, since OTel attributes cannot hold nested objects and instrumentation libraries like Logfire store them as JSON strings. A list value also matches an attribute stored as a tuple.
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.
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']