Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pydantic_evals/pydantic_evals/evaluators/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from dataclasses import dataclass, field
from typing import Any, Generic

from pydantic import TypeAdapter
from typing_extensions import TypeVar

from ..otel._errors import SpanTreeRecordingError
Expand Down Expand Up @@ -100,3 +101,7 @@ def span_tree(self) -> SpanTree:
# In this case, there was a reason we couldn't record the SpanTree. We raise that now
raise self._span_tree
return self._span_tree


EVALUATOR_CONTEXT_ADAPTER = TypeAdapter(EvaluatorContext)
"""This adapter can be used to serialize and deserialize `EvaluatorContext` objects to and from JSON."""
29 changes: 28 additions & 1 deletion pydantic_evals/pydantic_evals/otel/_errors.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,34 @@
from typing import Any

from pydantic_core import core_schema


class SpanTreeRecordingError(Exception):
"""An exception that is used to provide the reason why a SpanTree was not recorded by `context_subtree`.

This will either be due to missing dependencies or because a tracer provider had not been set.
"""

pass
message: str

def __init__(self, message: str):
self.message = message
super().__init__(message)

@classmethod
def __get_pydantic_core_schema__(cls, _: Any, __: Any) -> core_schema.CoreSchema:
"""Pydantic core schema to allow `SpanTreeRecordingError` to be (de)serialized."""
schema = core_schema.typed_dict_schema(
{
'message': core_schema.typed_dict_field(core_schema.str_schema()),
'kind': core_schema.typed_dict_field(core_schema.literal_schema(['span-tree-recording-error'])),
}
)
return core_schema.no_info_after_validator_function(
lambda dct: SpanTreeRecordingError(dct['message']),
schema,
serialization=core_schema.plain_serializer_function_ser_schema(
lambda x: {'message': x.message, 'kind': 'span-tree-recording-error'},
return_schema=schema,
),
)