# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from ..types.comment import Comment from ..types.error_info import ErrorInfo from ..types.error_info_write import ErrorInfoWrite from ..types.feedback_score_batch_item import FeedbackScoreBatchItem from ..types.feedback_score_names_public import FeedbackScoreNamesPublic from ..types.feedback_score_source import FeedbackScoreSource from ..types.json_list_string import JsonListString from ..types.json_list_string_write import JsonListStringWrite from ..types.project_stats_public import ProjectStatsPublic from ..types.span_filter_public import SpanFilterPublic from ..types.span_page_public import SpanPagePublic from ..types.span_public import SpanPublic from ..types.span_update import SpanUpdate from ..types.span_update_type import SpanUpdateType from ..types.span_write import SpanWrite from ..types.span_write_type import SpanWriteType from ..types.value_entry import ValueEntry from .raw_client import AsyncRawSpansClient, RawSpansClient from .types.find_feedback_score_names1request_type import FindFeedbackScoreNames1RequestType from .types.get_span_stats_request_type import GetSpanStatsRequestType from .types.get_spans_by_project_request_type import GetSpansByProjectRequestType from .types.span_search_stream_request_public_type import SpanSearchStreamRequestPublicType # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class SpansClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawSpansClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawSpansClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawSpansClient """ return self._raw_client def add_span_comment( self, id_: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add span comment Parameters ---------- id_ : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.add_span_comment(id_='id', text='text', ) """ _response = self._raw_client.add_span_comment( id_, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data def add_span_feedback_score( self, id: str, *, name: str, value: float, source: FeedbackScoreSource, category_name: typing.Optional[str] = OMIT, reason: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, value_by_author: typing.Optional[typing.Dict[str, ValueEntry]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add span feedback score Parameters ---------- id : str name : str value : float source : FeedbackScoreSource category_name : typing.Optional[str] reason : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] value_by_author : typing.Optional[typing.Dict[str, ValueEntry]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.add_span_feedback_score(id='id', name='name', value=1.1, source="ui", ) """ _response = self._raw_client.add_span_feedback_score( id, name=name, value=value, source=source, category_name=category_name, reason=reason, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, value_by_author=value_by_author, request_options=request_options, ) return _response.data def create_spans( self, *, spans: typing.Sequence[SpanWrite], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Create spans Parameters ---------- spans : typing.Sequence[SpanWrite] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import SpanWrite import datetime client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.create_spans(spans=[SpanWrite(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), )], ) """ _response = self._raw_client.create_spans(spans=spans, request_options=request_options) return _response.data def batch_update_spans( self, *, ids: typing.Sequence[str], update: SpanUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple spans Parameters ---------- ids : typing.Sequence[str] List of span IDs to update (max 1000) update : SpanUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import SpanUpdate client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.batch_update_spans(ids=['ids'], update=SpanUpdate(trace_id='trace_id', ), ) """ _response = self._raw_client.batch_update_spans( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data def get_spans_by_project( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, project_name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, trace_id: typing.Optional[str] = None, type: typing.Optional[GetSpansByProjectRequestType] = None, filters: typing.Optional[str] = None, truncate: typing.Optional[bool] = None, strip_attachments: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, exclude: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> SpanPagePublic: """ Get spans by project_name or project_id and optionally by trace_id and/or type Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] project_name : typing.Optional[str] project_id : typing.Optional[str] trace_id : typing.Optional[str] type : typing.Optional[GetSpansByProjectRequestType] filters : typing.Optional[str] truncate : typing.Optional[bool] strip_attachments : typing.Optional[bool] sorting : typing.Optional[str] exclude : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpanPagePublic Spans resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.get_spans_by_project() """ _response = self._raw_client.get_spans_by_project( page=page, size=size, project_name=project_name, project_id=project_id, trace_id=trace_id, type=type, filters=filters, truncate=truncate, strip_attachments=strip_attachments, sorting=sorting, exclude=exclude, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data def create_span( self, *, start_time: dt.datetime, id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, trace_id: typing.Optional[str] = OMIT, parent_span_id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, type: typing.Optional[SpanWriteType] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListStringWrite] = OMIT, output: typing.Optional[JsonListStringWrite] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, model: typing.Optional[str] = OMIT, provider: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, usage: typing.Optional[typing.Dict[str, int]] = OMIT, error_info: typing.Optional[ErrorInfoWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, total_estimated_cost: typing.Optional[float] = OMIT, total_estimated_cost_version: typing.Optional[str] = OMIT, ttft: typing.Optional[float] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create span Parameters ---------- start_time : dt.datetime id : typing.Optional[str] project_name : typing.Optional[str] If null, the default project is used trace_id : typing.Optional[str] parent_span_id : typing.Optional[str] name : typing.Optional[str] type : typing.Optional[SpanWriteType] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListStringWrite] output : typing.Optional[JsonListStringWrite] metadata : typing.Optional[JsonListStringWrite] model : typing.Optional[str] provider : typing.Optional[str] tags : typing.Optional[typing.Sequence[str]] usage : typing.Optional[typing.Dict[str, int]] error_info : typing.Optional[ErrorInfoWrite] last_updated_at : typing.Optional[dt.datetime] total_estimated_cost : typing.Optional[float] total_estimated_cost_version : typing.Optional[str] ttft : typing.Optional[float] Time to first token in milliseconds request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi import datetime client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.create_span(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), ) """ _response = self._raw_client.create_span( start_time=start_time, id=id, project_name=project_name, trace_id=trace_id, parent_span_id=parent_span_id, name=name, type=type, end_time=end_time, input=input, output=output, metadata=metadata, model=model, provider=provider, tags=tags, usage=usage, error_info=error_info, last_updated_at=last_updated_at, total_estimated_cost=total_estimated_cost, total_estimated_cost_version=total_estimated_cost_version, ttft=ttft, request_options=request_options, ) return _response.data def get_span_by_id( self, id: str, *, strip_attachments: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> SpanPublic: """ Get span by id Parameters ---------- id : str strip_attachments : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpanPublic Span resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.get_span_by_id(id='id', ) """ _response = self._raw_client.get_span_by_id( id, strip_attachments=strip_attachments, request_options=request_options ) return _response.data def delete_span_by_id(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete span by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.delete_span_by_id(id='id', ) """ _response = self._raw_client.delete_span_by_id(id, request_options=request_options) return _response.data def update_span( self, id: str, *, trace_id: str, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, parent_span_id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, type: typing.Optional[SpanUpdateType] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListString] = OMIT, output: typing.Optional[JsonListString] = OMIT, metadata: typing.Optional[JsonListString] = OMIT, model: typing.Optional[str] = OMIT, provider: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, usage: typing.Optional[typing.Dict[str, int]] = OMIT, total_estimated_cost: typing.Optional[float] = OMIT, error_info: typing.Optional[ErrorInfo] = OMIT, ttft: typing.Optional[float] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update span by id Parameters ---------- id : str trace_id : str project_name : typing.Optional[str] If null and project_id not specified, Default Project is assumed project_id : typing.Optional[str] If null and project_name not specified, Default Project is assumed parent_span_id : typing.Optional[str] name : typing.Optional[str] type : typing.Optional[SpanUpdateType] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListString] output : typing.Optional[JsonListString] metadata : typing.Optional[JsonListString] model : typing.Optional[str] provider : typing.Optional[str] tags : typing.Optional[typing.Sequence[str]] Tags tags_to_add : typing.Optional[typing.Sequence[str]] Tags to add tags_to_remove : typing.Optional[typing.Sequence[str]] Tags to remove usage : typing.Optional[typing.Dict[str, int]] total_estimated_cost : typing.Optional[float] error_info : typing.Optional[ErrorInfo] ttft : typing.Optional[float] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.update_span(id='id', trace_id='trace_id', ) """ _response = self._raw_client.update_span( id, trace_id=trace_id, project_name=project_name, project_id=project_id, parent_span_id=parent_span_id, name=name, type=type, end_time=end_time, input=input, output=output, metadata=metadata, model=model, provider=provider, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, usage=usage, total_estimated_cost=total_estimated_cost, error_info=error_info, ttft=ttft, request_options=request_options, ) return _response.data def delete_span_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete span comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.delete_span_comments(ids=['ids'], ) """ _response = self._raw_client.delete_span_comments(ids=ids, request_options=request_options) return _response.data def delete_span_feedback_score( self, id: str, *, name: str, author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete span feedback score Parameters ---------- id : str name : str author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.delete_span_feedback_score(id='id', name='name', ) """ _response = self._raw_client.delete_span_feedback_score( id, name=name, author=author, request_options=request_options ) return _response.data def find_feedback_score_names1( self, *, project_id: typing.Optional[str] = None, type: typing.Optional[FindFeedbackScoreNames1RequestType] = None, request_options: typing.Optional[RequestOptions] = None, ) -> FeedbackScoreNamesPublic: """ Find Feedback Score names Parameters ---------- project_id : typing.Optional[str] type : typing.Optional[FindFeedbackScoreNames1RequestType] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Feedback Scores resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.find_feedback_score_names1() """ _response = self._raw_client.find_feedback_score_names1( project_id=project_id, type=type, request_options=request_options ) return _response.data def get_span_comment( self, span_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Comment: """ Get span comment Parameters ---------- span_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Comment Comment resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.get_span_comment(span_id='spanId', comment_id='commentId', ) """ _response = self._raw_client.get_span_comment(span_id, comment_id, request_options=request_options) return _response.data def get_span_stats( self, *, project_id: typing.Optional[str] = None, project_name: typing.Optional[str] = None, trace_id: typing.Optional[str] = None, type: typing.Optional[GetSpanStatsRequestType] = None, filters: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ProjectStatsPublic: """ Get span stats Parameters ---------- project_id : typing.Optional[str] project_name : typing.Optional[str] trace_id : typing.Optional[str] type : typing.Optional[GetSpanStatsRequestType] filters : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectStatsPublic Span stats resource Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.get_span_stats() """ _response = self._raw_client.get_span_stats( project_id=project_id, project_name=project_name, trace_id=trace_id, type=type, filters=filters, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data def score_batch_of_spans( self, *, scores: typing.Sequence[FeedbackScoreBatchItem], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Batch feedback scoring for spans Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import FeedbackScoreBatchItem client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.score_batch_of_spans(scores=[FeedbackScoreBatchItem(name='name', value=1.1, source="ui", id='id', )], ) """ _response = self._raw_client.score_batch_of_spans(scores=scores, request_options=request_options) return _response.data def search_spans( self, *, trace_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, type: typing.Optional[SpanSearchStreamRequestPublicType] = OMIT, filters: typing.Optional[typing.Sequence[SpanFilterPublic]] = OMIT, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, from_time: typing.Optional[dt.datetime] = OMIT, to_time: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[bytes]: """ Search spans Parameters ---------- trace_id : typing.Optional[str] project_name : typing.Optional[str] project_id : typing.Optional[str] type : typing.Optional[SpanSearchStreamRequestPublicType] filters : typing.Optional[typing.Sequence[SpanFilterPublic]] limit : typing.Optional[int] Max number of spans to be streamed last_retrieved_id : typing.Optional[str] truncate : typing.Optional[bool] Truncate image included in either input, output or metadata from_time : typing.Optional[dt.datetime] Filter spans created from this time (ISO-8601 format). to_time : typing.Optional[dt.datetime] Filter spans created up to this time (ISO-8601 format). If not provided, defaults to current time. Must be after 'from_time'. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[bytes] Spans stream or error during process Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.search_spans() """ with self._raw_client.search_spans( trace_id=trace_id, project_name=project_name, project_id=project_id, type=type, filters=filters, limit=limit, last_retrieved_id=last_retrieved_id, truncate=truncate, from_time=from_time, to_time=to_time, request_options=request_options, ) as r: yield from r.data def update_span_comment( self, comment_id: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update span comment by id Parameters ---------- comment_id : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.spans.update_span_comment(comment_id='commentId', text='text', ) """ _response = self._raw_client.update_span_comment( comment_id, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data class AsyncSpansClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawSpansClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawSpansClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawSpansClient """ return self._raw_client async def add_span_comment( self, id_: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add span comment Parameters ---------- id_ : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.add_span_comment(id_='id', text='text', ) asyncio.run(main()) """ _response = await self._raw_client.add_span_comment( id_, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data async def add_span_feedback_score( self, id: str, *, name: str, value: float, source: FeedbackScoreSource, category_name: typing.Optional[str] = OMIT, reason: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, value_by_author: typing.Optional[typing.Dict[str, ValueEntry]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Add span feedback score Parameters ---------- id : str name : str value : float source : FeedbackScoreSource category_name : typing.Optional[str] reason : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] value_by_author : typing.Optional[typing.Dict[str, ValueEntry]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.add_span_feedback_score(id='id', name='name', value=1.1, source="ui", ) asyncio.run(main()) """ _response = await self._raw_client.add_span_feedback_score( id, name=name, value=value, source=source, category_name=category_name, reason=reason, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, value_by_author=value_by_author, request_options=request_options, ) return _response.data async def create_spans( self, *, spans: typing.Sequence[SpanWrite], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Create spans Parameters ---------- spans : typing.Sequence[SpanWrite] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import SpanWrite import datetime import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.create_spans(spans=[SpanWrite(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), )], ) asyncio.run(main()) """ _response = await self._raw_client.create_spans(spans=spans, request_options=request_options) return _response.data async def batch_update_spans( self, *, ids: typing.Sequence[str], update: SpanUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update multiple spans Parameters ---------- ids : typing.Sequence[str] List of span IDs to update (max 1000) update : SpanUpdate merge_tags : typing.Optional[bool] If true, merge tags with existing tags instead of replacing them. Default: false request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import SpanUpdate import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.batch_update_spans(ids=['ids'], update=SpanUpdate(trace_id='trace_id', ), ) asyncio.run(main()) """ _response = await self._raw_client.batch_update_spans( ids=ids, update=update, merge_tags=merge_tags, request_options=request_options ) return _response.data async def get_spans_by_project( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, project_name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, trace_id: typing.Optional[str] = None, type: typing.Optional[GetSpansByProjectRequestType] = None, filters: typing.Optional[str] = None, truncate: typing.Optional[bool] = None, strip_attachments: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, exclude: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> SpanPagePublic: """ Get spans by project_name or project_id and optionally by trace_id and/or type Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] project_name : typing.Optional[str] project_id : typing.Optional[str] trace_id : typing.Optional[str] type : typing.Optional[GetSpansByProjectRequestType] filters : typing.Optional[str] truncate : typing.Optional[bool] strip_attachments : typing.Optional[bool] sorting : typing.Optional[str] exclude : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpanPagePublic Spans resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.get_spans_by_project() asyncio.run(main()) """ _response = await self._raw_client.get_spans_by_project( page=page, size=size, project_name=project_name, project_id=project_id, trace_id=trace_id, type=type, filters=filters, truncate=truncate, strip_attachments=strip_attachments, sorting=sorting, exclude=exclude, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data async def create_span( self, *, start_time: dt.datetime, id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, trace_id: typing.Optional[str] = OMIT, parent_span_id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, type: typing.Optional[SpanWriteType] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListStringWrite] = OMIT, output: typing.Optional[JsonListStringWrite] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, model: typing.Optional[str] = OMIT, provider: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, usage: typing.Optional[typing.Dict[str, int]] = OMIT, error_info: typing.Optional[ErrorInfoWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, total_estimated_cost: typing.Optional[float] = OMIT, total_estimated_cost_version: typing.Optional[str] = OMIT, ttft: typing.Optional[float] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create span Parameters ---------- start_time : dt.datetime id : typing.Optional[str] project_name : typing.Optional[str] If null, the default project is used trace_id : typing.Optional[str] parent_span_id : typing.Optional[str] name : typing.Optional[str] type : typing.Optional[SpanWriteType] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListStringWrite] output : typing.Optional[JsonListStringWrite] metadata : typing.Optional[JsonListStringWrite] model : typing.Optional[str] provider : typing.Optional[str] tags : typing.Optional[typing.Sequence[str]] usage : typing.Optional[typing.Dict[str, int]] error_info : typing.Optional[ErrorInfoWrite] last_updated_at : typing.Optional[dt.datetime] total_estimated_cost : typing.Optional[float] total_estimated_cost_version : typing.Optional[str] ttft : typing.Optional[float] Time to first token in milliseconds request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import datetime import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.create_span(start_time=datetime.datetime.fromisoformat("2024-01-15 09:30:00+00:00", ), ) asyncio.run(main()) """ _response = await self._raw_client.create_span( start_time=start_time, id=id, project_name=project_name, trace_id=trace_id, parent_span_id=parent_span_id, name=name, type=type, end_time=end_time, input=input, output=output, metadata=metadata, model=model, provider=provider, tags=tags, usage=usage, error_info=error_info, last_updated_at=last_updated_at, total_estimated_cost=total_estimated_cost, total_estimated_cost_version=total_estimated_cost_version, ttft=ttft, request_options=request_options, ) return _response.data async def get_span_by_id( self, id: str, *, strip_attachments: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> SpanPublic: """ Get span by id Parameters ---------- id : str strip_attachments : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpanPublic Span resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.get_span_by_id(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.get_span_by_id( id, strip_attachments=strip_attachments, request_options=request_options ) return _response.data async def delete_span_by_id(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete span by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.delete_span_by_id(id='id', ) asyncio.run(main()) """ _response = await self._raw_client.delete_span_by_id(id, request_options=request_options) return _response.data async def update_span( self, id: str, *, trace_id: str, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, parent_span_id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, type: typing.Optional[SpanUpdateType] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, input: typing.Optional[JsonListString] = OMIT, output: typing.Optional[JsonListString] = OMIT, metadata: typing.Optional[JsonListString] = OMIT, model: typing.Optional[str] = OMIT, provider: typing.Optional[str] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_add: typing.Optional[typing.Sequence[str]] = OMIT, tags_to_remove: typing.Optional[typing.Sequence[str]] = OMIT, usage: typing.Optional[typing.Dict[str, int]] = OMIT, total_estimated_cost: typing.Optional[float] = OMIT, error_info: typing.Optional[ErrorInfo] = OMIT, ttft: typing.Optional[float] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update span by id Parameters ---------- id : str trace_id : str project_name : typing.Optional[str] If null and project_id not specified, Default Project is assumed project_id : typing.Optional[str] If null and project_name not specified, Default Project is assumed parent_span_id : typing.Optional[str] name : typing.Optional[str] type : typing.Optional[SpanUpdateType] end_time : typing.Optional[dt.datetime] input : typing.Optional[JsonListString] output : typing.Optional[JsonListString] metadata : typing.Optional[JsonListString] model : typing.Optional[str] provider : typing.Optional[str] tags : typing.Optional[typing.Sequence[str]] Tags tags_to_add : typing.Optional[typing.Sequence[str]] Tags to add tags_to_remove : typing.Optional[typing.Sequence[str]] Tags to remove usage : typing.Optional[typing.Dict[str, int]] total_estimated_cost : typing.Optional[float] error_info : typing.Optional[ErrorInfo] ttft : typing.Optional[float] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.update_span(id='id', trace_id='trace_id', ) asyncio.run(main()) """ _response = await self._raw_client.update_span( id, trace_id=trace_id, project_name=project_name, project_id=project_id, parent_span_id=parent_span_id, name=name, type=type, end_time=end_time, input=input, output=output, metadata=metadata, model=model, provider=provider, tags=tags, tags_to_add=tags_to_add, tags_to_remove=tags_to_remove, usage=usage, total_estimated_cost=total_estimated_cost, error_info=error_info, ttft=ttft, request_options=request_options, ) return _response.data async def delete_span_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> None: """ Delete span comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.delete_span_comments(ids=['ids'], ) asyncio.run(main()) """ _response = await self._raw_client.delete_span_comments(ids=ids, request_options=request_options) return _response.data async def delete_span_feedback_score( self, id: str, *, name: str, author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Delete span feedback score Parameters ---------- id : str name : str author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.delete_span_feedback_score(id='id', name='name', ) asyncio.run(main()) """ _response = await self._raw_client.delete_span_feedback_score( id, name=name, author=author, request_options=request_options ) return _response.data async def find_feedback_score_names1( self, *, project_id: typing.Optional[str] = None, type: typing.Optional[FindFeedbackScoreNames1RequestType] = None, request_options: typing.Optional[RequestOptions] = None, ) -> FeedbackScoreNamesPublic: """ Find Feedback Score names Parameters ---------- project_id : typing.Optional[str] type : typing.Optional[FindFeedbackScoreNames1RequestType] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- FeedbackScoreNamesPublic Feedback Scores resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.find_feedback_score_names1() asyncio.run(main()) """ _response = await self._raw_client.find_feedback_score_names1( project_id=project_id, type=type, request_options=request_options ) return _response.data async def get_span_comment( self, span_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> Comment: """ Get span comment Parameters ---------- span_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Comment Comment resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.get_span_comment(span_id='spanId', comment_id='commentId', ) asyncio.run(main()) """ _response = await self._raw_client.get_span_comment(span_id, comment_id, request_options=request_options) return _response.data async def get_span_stats( self, *, project_id: typing.Optional[str] = None, project_name: typing.Optional[str] = None, trace_id: typing.Optional[str] = None, type: typing.Optional[GetSpanStatsRequestType] = None, filters: typing.Optional[str] = None, search: typing.Optional[str] = None, from_time: typing.Optional[dt.datetime] = None, to_time: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> ProjectStatsPublic: """ Get span stats Parameters ---------- project_id : typing.Optional[str] project_name : typing.Optional[str] trace_id : typing.Optional[str] type : typing.Optional[GetSpanStatsRequestType] filters : typing.Optional[str] search : typing.Optional[str] from_time : typing.Optional[dt.datetime] to_time : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectStatsPublic Span stats resource Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.get_span_stats() asyncio.run(main()) """ _response = await self._raw_client.get_span_stats( project_id=project_id, project_name=project_name, trace_id=trace_id, type=type, filters=filters, search=search, from_time=from_time, to_time=to_time, request_options=request_options, ) return _response.data async def score_batch_of_spans( self, *, scores: typing.Sequence[FeedbackScoreBatchItem], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Batch feedback scoring for spans Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import FeedbackScoreBatchItem import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.score_batch_of_spans(scores=[FeedbackScoreBatchItem(name='name', value=1.1, source="ui", id='id', )], ) asyncio.run(main()) """ _response = await self._raw_client.score_batch_of_spans(scores=scores, request_options=request_options) return _response.data async def search_spans( self, *, trace_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, type: typing.Optional[SpanSearchStreamRequestPublicType] = OMIT, filters: typing.Optional[typing.Sequence[SpanFilterPublic]] = OMIT, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, from_time: typing.Optional[dt.datetime] = OMIT, to_time: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[bytes]: """ Search spans Parameters ---------- trace_id : typing.Optional[str] project_name : typing.Optional[str] project_id : typing.Optional[str] type : typing.Optional[SpanSearchStreamRequestPublicType] filters : typing.Optional[typing.Sequence[SpanFilterPublic]] limit : typing.Optional[int] Max number of spans to be streamed last_retrieved_id : typing.Optional[str] truncate : typing.Optional[bool] Truncate image included in either input, output or metadata from_time : typing.Optional[dt.datetime] Filter spans created from this time (ISO-8601 format). to_time : typing.Optional[dt.datetime] Filter spans created up to this time (ISO-8601 format). If not provided, defaults to current time. Must be after 'from_time'. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[bytes] Spans stream or error during process Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.search_spans() asyncio.run(main()) """ async with self._raw_client.search_spans( trace_id=trace_id, project_name=project_name, project_id=project_id, type=type, filters=filters, limit=limit, last_retrieved_id=last_retrieved_id, truncate=truncate, from_time=from_time, to_time=to_time, request_options=request_options, ) as r: async for data in r.data: yield data async def update_span_comment( self, comment_id: str, *, text: str, id: typing.Optional[str] = OMIT, created_at: typing.Optional[dt.datetime] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, created_by: typing.Optional[str] = OMIT, last_updated_by: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Update span comment by id Parameters ---------- comment_id : str text : str id : typing.Optional[str] created_at : typing.Optional[dt.datetime] last_updated_at : typing.Optional[dt.datetime] created_by : typing.Optional[str] last_updated_by : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.spans.update_span_comment(comment_id='commentId', text='text', ) asyncio.run(main()) """ _response = await self._raw_client.update_span_comment( comment_id, text=text, id=id, created_at=created_at, last_updated_at=last_updated_at, created_by=created_by, last_updated_by=last_updated_by, request_options=request_options, ) return _response.data