# This file was auto-generated by Fern from our API Definition. import contextlib import datetime as dt import typing from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.datetime_utils import serialize_datetime from ..core.http_response import AsyncHttpResponse, HttpResponse from ..core.jsonable_encoder import jsonable_encoder from ..core.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from ..core.serialization import convert_and_respect_annotation_metadata from ..errors.bad_request_error import BadRequestError from ..errors.conflict_error import ConflictError from ..errors.not_found_error import NotFoundError from ..errors.not_implemented_error import NotImplementedError 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 .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 RawSpansClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper 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, ) -> HttpResponse[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 ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id_)}/comments", method="POST", json={ "id": id, "text": text, "created_at": created_at, "last_updated_at": last_updated_at, "created_by": created_by, "last_updated_by": last_updated_by, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> HttpResponse[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 ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}/feedback-scores", method="PUT", json={ "name": name, "category_name": category_name, "value": value, "reason": reason, "source": source, "created_at": created_at, "last_updated_at": last_updated_at, "created_by": created_by, "last_updated_by": last_updated_by, "value_by_author": convert_and_respect_annotation_metadata( object_=value_by_author, annotation=typing.Dict[str, ValueEntry], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def create_spans( self, *, spans: typing.Sequence[SpanWrite], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Create spans Parameters ---------- spans : typing.Sequence[SpanWrite] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans/batch", method="POST", json={ "spans": convert_and_respect_annotation_metadata( object_=spans, annotation=typing.Sequence[SpanWrite], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def batch_update_spans( self, *, ids: typing.Sequence[str], update: SpanUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[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 ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans/batch", method="PATCH", json={ "ids": ids, "update": convert_and_respect_annotation_metadata( object_=update, annotation=SpanUpdate, direction="write" ), "merge_tags": merge_tags, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> HttpResponse[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 ------- HttpResponse[SpanPagePublic] Spans resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans", method="GET", params={ "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": serialize_datetime(from_time) if from_time is not None else None, "to_time": serialize_datetime(to_time) if to_time is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SpanPagePublic, parse_obj_as( type_=SpanPagePublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> HttpResponse[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 ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans", method="POST", json={ "id": id, "project_name": project_name, "trace_id": trace_id, "parent_span_id": parent_span_id, "name": name, "type": type, "start_time": start_time, "end_time": end_time, "input": convert_and_respect_annotation_metadata( object_=input, annotation=JsonListStringWrite, direction="write" ), "output": convert_and_respect_annotation_metadata( object_=output, annotation=JsonListStringWrite, direction="write" ), "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "model": model, "provider": provider, "tags": tags, "usage": usage, "error_info": convert_and_respect_annotation_metadata( object_=error_info, annotation=ErrorInfoWrite, direction="write" ), "last_updated_at": last_updated_at, "total_estimated_cost": total_estimated_cost, "total_estimated_cost_version": total_estimated_cost_version, "ttft": ttft, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 409: raise ConflictError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_span_by_id( self, id: str, *, strip_attachments: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[SpanPublic]: """ Get span by id Parameters ---------- id : str strip_attachments : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SpanPublic] Span resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}", method="GET", params={ "strip_attachments": strip_attachments, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SpanPublic, parse_obj_as( type_=SpanPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete_span_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete span by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 501: raise NotImplementedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> HttpResponse[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 ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}", method="PATCH", json={ "project_name": project_name, "project_id": project_id, "trace_id": trace_id, "parent_span_id": parent_span_id, "name": name, "type": type, "end_time": end_time, "input": convert_and_respect_annotation_metadata( object_=input, annotation=JsonListString, direction="write" ), "output": convert_and_respect_annotation_metadata( object_=output, annotation=JsonListString, direction="write" ), "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListString, direction="write" ), "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": convert_and_respect_annotation_metadata( object_=error_info, annotation=ErrorInfo, direction="write" ), "ttft": ttft, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete_span_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete span comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans/comments/delete", method="POST", json={ "ids": ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete_span_feedback_score( self, id: str, *, name: str, author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Delete span feedback score Parameters ---------- id : str name : str author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}/feedback-scores/delete", method="POST", json={ "name": name, "author": author, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def find_feedback_score_names1( self, *, project_id: typing.Optional[str] = None, type: typing.Optional[FindFeedbackScoreNames1RequestType] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[FeedbackScoreNamesPublic]: """ Find Feedback Score names Parameters ---------- project_id : typing.Optional[str] type : typing.Optional[FindFeedbackScoreNames1RequestType] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[FeedbackScoreNamesPublic] Feedback Scores resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans/feedback-scores/names", method="GET", params={ "project_id": project_id, "type": type, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FeedbackScoreNamesPublic, parse_obj_as( type_=FeedbackScoreNamesPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_span_comment( self, span_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[Comment]: """ Get span comment Parameters ---------- span_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Comment] Comment resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(span_id)}/comments/{jsonable_encoder(comment_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Comment, parse_obj_as( type_=Comment, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> HttpResponse[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 ------- HttpResponse[ProjectStatsPublic] Span stats resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans/stats", method="GET", params={ "project_id": project_id, "project_name": project_name, "trace_id": trace_id, "type": type, "filters": filters, "search": search, "from_time": serialize_datetime(from_time) if from_time is not None else None, "to_time": serialize_datetime(to_time) if to_time is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectStatsPublic, parse_obj_as( type_=ProjectStatsPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def score_batch_of_spans( self, *, scores: typing.Sequence[FeedbackScoreBatchItem], request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Batch feedback scoring for spans Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/spans/feedback-scores", method="PUT", json={ "scores": convert_and_respect_annotation_metadata( object_=scores, annotation=typing.Sequence[FeedbackScoreBatchItem], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) @contextlib.contextmanager 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[HttpResponse[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[HttpResponse[typing.Iterator[bytes]]] Spans stream or error during process """ with self._client_wrapper.httpx_client.stream( "v1/private/spans/search", method="POST", json={ "trace_id": trace_id, "project_name": project_name, "project_id": project_id, "type": type, "filters": convert_and_respect_annotation_metadata( object_=filters, annotation=typing.Sequence[SpanFilterPublic], direction="write" ), "limit": limit, "last_retrieved_id": last_retrieved_id, "truncate": truncate, "from_time": from_time, "to_time": to_time, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) as _response: def stream() -> HttpResponse[typing.Iterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", None) if request_options is not None else None return HttpResponse( response=_response, data=(_chunk for _chunk in _response.iter_bytes(chunk_size=_chunk_size)) ) _response.read() if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.text ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) yield stream() 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, ) -> HttpResponse[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 ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/spans/comments/{jsonable_encoder(comment_id)}", method="PATCH", json={ "id": id, "text": text, "created_at": created_at, "last_updated_at": last_updated_at, "created_by": created_by, "last_updated_by": last_updated_by, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawSpansClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id_)}/comments", method="POST", json={ "id": id, "text": text, "created_at": created_at, "last_updated_at": last_updated_at, "created_by": created_by, "last_updated_by": last_updated_by, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}/feedback-scores", method="PUT", json={ "name": name, "category_name": category_name, "value": value, "reason": reason, "source": source, "created_at": created_at, "last_updated_at": last_updated_at, "created_by": created_by, "last_updated_by": last_updated_by, "value_by_author": convert_and_respect_annotation_metadata( object_=value_by_author, annotation=typing.Dict[str, ValueEntry], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def create_spans( self, *, spans: typing.Sequence[SpanWrite], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Create spans Parameters ---------- spans : typing.Sequence[SpanWrite] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans/batch", method="POST", json={ "spans": convert_and_respect_annotation_metadata( object_=spans, annotation=typing.Sequence[SpanWrite], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def batch_update_spans( self, *, ids: typing.Sequence[str], update: SpanUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans/batch", method="PATCH", json={ "ids": ids, "update": convert_and_respect_annotation_metadata( object_=update, annotation=SpanUpdate, direction="write" ), "merge_tags": merge_tags, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[SpanPagePublic] Spans resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans", method="GET", params={ "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": serialize_datetime(from_time) if from_time is not None else None, "to_time": serialize_datetime(to_time) if to_time is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SpanPagePublic, parse_obj_as( type_=SpanPagePublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans", method="POST", json={ "id": id, "project_name": project_name, "trace_id": trace_id, "parent_span_id": parent_span_id, "name": name, "type": type, "start_time": start_time, "end_time": end_time, "input": convert_and_respect_annotation_metadata( object_=input, annotation=JsonListStringWrite, direction="write" ), "output": convert_and_respect_annotation_metadata( object_=output, annotation=JsonListStringWrite, direction="write" ), "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "model": model, "provider": provider, "tags": tags, "usage": usage, "error_info": convert_and_respect_annotation_metadata( object_=error_info, annotation=ErrorInfoWrite, direction="write" ), "last_updated_at": last_updated_at, "total_estimated_cost": total_estimated_cost, "total_estimated_cost_version": total_estimated_cost_version, "ttft": ttft, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 409: raise ConflictError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_span_by_id( self, id: str, *, strip_attachments: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[SpanPublic]: """ Get span by id Parameters ---------- id : str strip_attachments : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SpanPublic] Span resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}", method="GET", params={ "strip_attachments": strip_attachments, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SpanPublic, parse_obj_as( type_=SpanPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete_span_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete span by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 501: raise NotImplementedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}", method="PATCH", json={ "project_name": project_name, "project_id": project_id, "trace_id": trace_id, "parent_span_id": parent_span_id, "name": name, "type": type, "end_time": end_time, "input": convert_and_respect_annotation_metadata( object_=input, annotation=JsonListString, direction="write" ), "output": convert_and_respect_annotation_metadata( object_=output, annotation=JsonListString, direction="write" ), "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListString, direction="write" ), "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": convert_and_respect_annotation_metadata( object_=error_info, annotation=ErrorInfo, direction="write" ), "ttft": ttft, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete_span_comments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete span comments Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans/comments/delete", method="POST", json={ "ids": ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete_span_feedback_score( self, id: str, *, name: str, author: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Delete span feedback score Parameters ---------- id : str name : str author : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(id)}/feedback-scores/delete", method="POST", json={ "name": name, "author": author, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def find_feedback_score_names1( self, *, project_id: typing.Optional[str] = None, type: typing.Optional[FindFeedbackScoreNames1RequestType] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[FeedbackScoreNamesPublic]: """ Find Feedback Score names Parameters ---------- project_id : typing.Optional[str] type : typing.Optional[FindFeedbackScoreNames1RequestType] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[FeedbackScoreNamesPublic] Feedback Scores resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans/feedback-scores/names", method="GET", params={ "project_id": project_id, "type": type, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FeedbackScoreNamesPublic, parse_obj_as( type_=FeedbackScoreNamesPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_span_comment( self, span_id: str, comment_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[Comment]: """ Get span comment Parameters ---------- span_id : str comment_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Comment] Comment resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/{jsonable_encoder(span_id)}/comments/{jsonable_encoder(comment_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Comment, parse_obj_as( type_=Comment, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[ProjectStatsPublic] Span stats resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans/stats", method="GET", params={ "project_id": project_id, "project_name": project_name, "trace_id": trace_id, "type": type, "filters": filters, "search": search, "from_time": serialize_datetime(from_time) if from_time is not None else None, "to_time": serialize_datetime(to_time) if to_time is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectStatsPublic, parse_obj_as( type_=ProjectStatsPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def score_batch_of_spans( self, *, scores: typing.Sequence[FeedbackScoreBatchItem], request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Batch feedback scoring for spans Parameters ---------- scores : typing.Sequence[FeedbackScoreBatchItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/spans/feedback-scores", method="PUT", json={ "scores": convert_and_respect_annotation_metadata( object_=scores, annotation=typing.Sequence[FeedbackScoreBatchItem], direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) @contextlib.asynccontextmanager 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[AsyncHttpResponse[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[AsyncHttpResponse[typing.AsyncIterator[bytes]]] Spans stream or error during process """ async with self._client_wrapper.httpx_client.stream( "v1/private/spans/search", method="POST", json={ "trace_id": trace_id, "project_name": project_name, "project_id": project_id, "type": type, "filters": convert_and_respect_annotation_metadata( object_=filters, annotation=typing.Sequence[SpanFilterPublic], direction="write" ), "limit": limit, "last_retrieved_id": last_retrieved_id, "truncate": truncate, "from_time": from_time, "to_time": to_time, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) as _response: async def stream() -> AsyncHttpResponse[typing.AsyncIterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", None) if request_options is not None else None return AsyncHttpResponse( response=_response, data=(_chunk async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size)), ) await _response.aread() if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.text ) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) yield await stream() 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, ) -> AsyncHttpResponse[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 ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/spans/comments/{jsonable_encoder(comment_id)}", method="PATCH", json={ "id": id, "text": text, "created_at": created_at, "last_updated_at": last_updated_at, "created_by": created_by, "last_updated_by": last_updated_by, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)