# This file was auto-generated by Fern from our API Definition. 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.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.unprocessable_entity_error import UnprocessableEntityError from ..types.breakdown_config_public import BreakdownConfigPublic from ..types.feedback_score_names import FeedbackScoreNames from ..types.project_detailed import ProjectDetailed from ..types.project_metric_response_public import ProjectMetricResponsePublic from ..types.project_page_public import ProjectPagePublic from ..types.project_public import ProjectPublic from ..types.project_stats_summary import ProjectStatsSummary from ..types.span_filter_public import SpanFilterPublic from ..types.token_usage_names import TokenUsageNames from ..types.trace_filter_public import TraceFilterPublic from ..types.trace_thread_filter_public import TraceThreadFilterPublic from .types.project_metric_request_public_interval import ProjectMetricRequestPublicInterval from .types.project_metric_request_public_metric_type import ProjectMetricRequestPublicMetricType from .types.project_update_visibility import ProjectUpdateVisibility from .types.project_write_visibility import ProjectWriteVisibility # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawProjectsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def find_projects( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, name: typing.Optional[str] = None, sorting: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ProjectPagePublic]: """ Find projects Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] name : typing.Optional[str] sorting : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProjectPagePublic] Project resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/projects", method="GET", params={ "page": page, "size": size, "name": name, "sorting": sorting, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectPagePublic, parse_obj_as( type_=ProjectPagePublic, # 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_project( self, *, name: str, visibility: typing.Optional[ProjectWriteVisibility] = OMIT, description: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Create project Parameters ---------- name : str visibility : typing.Optional[ProjectWriteVisibility] description : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/projects", method="POST", json={ "name": name, "visibility": visibility, "description": description, }, 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(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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_project_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ProjectPublic]: """ Get project by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProjectPublic] Project resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectPublic, parse_obj_as( type_=ProjectPublic, # 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 delete_project_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete project 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/projects/{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 == 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 update_project( self, id: str, *, name: typing.Optional[str] = OMIT, description: typing.Optional[str] = OMIT, visibility: typing.Optional[ProjectUpdateVisibility] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Update project by id Parameters ---------- id : str name : typing.Optional[str] description : typing.Optional[str] visibility : typing.Optional[ProjectUpdateVisibility] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "description": description, "visibility": visibility, }, 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(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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_projects_batch( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete projects batch 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/projects/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 find_feedback_score_names_by_project_ids( self, *, project_ids: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[FeedbackScoreNames]: """ Find Feedback Score names By Project Ids Parameters ---------- project_ids : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[FeedbackScoreNames] Feedback Scores resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/projects/feedback-scores/names", method="GET", params={ "project_ids": project_ids, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FeedbackScoreNames, parse_obj_as( type_=FeedbackScoreNames, # 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 find_token_usage_names( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[TokenUsageNames]: """ Find Token Usage names Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[TokenUsageNames] Token Usage names resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}/token-usage/names", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( TokenUsageNames, parse_obj_as( type_=TokenUsageNames, # 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_project_metrics( self, id: str, *, metric_type: typing.Optional[ProjectMetricRequestPublicMetricType] = OMIT, interval: typing.Optional[ProjectMetricRequestPublicInterval] = OMIT, interval_start: typing.Optional[dt.datetime] = OMIT, interval_end: typing.Optional[dt.datetime] = OMIT, span_filters: typing.Optional[typing.Sequence[SpanFilterPublic]] = OMIT, trace_filters: typing.Optional[typing.Sequence[TraceFilterPublic]] = OMIT, thread_filters: typing.Optional[typing.Sequence[TraceThreadFilterPublic]] = OMIT, breakdown: typing.Optional[BreakdownConfigPublic] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ProjectMetricResponsePublic]: """ Gets specified metrics for a project Parameters ---------- id : str metric_type : typing.Optional[ProjectMetricRequestPublicMetricType] interval : typing.Optional[ProjectMetricRequestPublicInterval] interval_start : typing.Optional[dt.datetime] interval_end : typing.Optional[dt.datetime] span_filters : typing.Optional[typing.Sequence[SpanFilterPublic]] trace_filters : typing.Optional[typing.Sequence[TraceFilterPublic]] thread_filters : typing.Optional[typing.Sequence[TraceThreadFilterPublic]] breakdown : typing.Optional[BreakdownConfigPublic] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProjectMetricResponsePublic] Project Metrics """ _response = self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}/metrics", method="POST", json={ "metric_type": metric_type, "interval": interval, "interval_start": interval_start, "interval_end": interval_end, "span_filters": convert_and_respect_annotation_metadata( object_=span_filters, annotation=typing.Sequence[SpanFilterPublic], direction="write" ), "trace_filters": convert_and_respect_annotation_metadata( object_=trace_filters, annotation=typing.Sequence[TraceFilterPublic], direction="write" ), "thread_filters": convert_and_respect_annotation_metadata( object_=thread_filters, annotation=typing.Sequence[TraceThreadFilterPublic], direction="write" ), "breakdown": convert_and_respect_annotation_metadata( object_=breakdown, annotation=BreakdownConfigPublic, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectMetricResponsePublic, parse_obj_as( type_=ProjectMetricResponsePublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) 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(), ), ), ) 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_project_stats( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, name: typing.Optional[str] = None, sorting: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ProjectStatsSummary]: """ Get Project Stats Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] name : typing.Optional[str] sorting : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProjectStatsSummary] Project Stats """ _response = self._client_wrapper.httpx_client.request( "v1/private/projects/stats", method="GET", params={ "page": page, "size": size, "name": name, "sorting": sorting, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectStatsSummary, parse_obj_as( type_=ProjectStatsSummary, # 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 retrieve_project( self, *, name: str, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ProjectDetailed]: """ Retrieve project Parameters ---------- name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProjectDetailed] Project resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/projects/retrieve", method="POST", json={ "name": name, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectDetailed, parse_obj_as( type_=ProjectDetailed, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) 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(), ), ), ) 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(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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 AsyncRawProjectsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def find_projects( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, name: typing.Optional[str] = None, sorting: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ProjectPagePublic]: """ Find projects Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] name : typing.Optional[str] sorting : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProjectPagePublic] Project resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/projects", method="GET", params={ "page": page, "size": size, "name": name, "sorting": sorting, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectPagePublic, parse_obj_as( type_=ProjectPagePublic, # 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_project( self, *, name: str, visibility: typing.Optional[ProjectWriteVisibility] = OMIT, description: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Create project Parameters ---------- name : str visibility : typing.Optional[ProjectWriteVisibility] description : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/projects", method="POST", json={ "name": name, "visibility": visibility, "description": description, }, 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(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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_project_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ProjectPublic]: """ Get project by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProjectPublic] Project resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectPublic, parse_obj_as( type_=ProjectPublic, # 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 delete_project_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete project 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/projects/{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 == 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 update_project( self, id: str, *, name: typing.Optional[str] = OMIT, description: typing.Optional[str] = OMIT, visibility: typing.Optional[ProjectUpdateVisibility] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Update project by id Parameters ---------- id : str name : typing.Optional[str] description : typing.Optional[str] visibility : typing.Optional[ProjectUpdateVisibility] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "description": description, "visibility": visibility, }, 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(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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_projects_batch( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete projects batch 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/projects/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 find_feedback_score_names_by_project_ids( self, *, project_ids: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[FeedbackScoreNames]: """ Find Feedback Score names By Project Ids Parameters ---------- project_ids : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[FeedbackScoreNames] Feedback Scores resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/projects/feedback-scores/names", method="GET", params={ "project_ids": project_ids, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( FeedbackScoreNames, parse_obj_as( type_=FeedbackScoreNames, # 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 find_token_usage_names( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[TokenUsageNames]: """ Find Token Usage names Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[TokenUsageNames] Token Usage names resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}/token-usage/names", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( TokenUsageNames, parse_obj_as( type_=TokenUsageNames, # 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_project_metrics( self, id: str, *, metric_type: typing.Optional[ProjectMetricRequestPublicMetricType] = OMIT, interval: typing.Optional[ProjectMetricRequestPublicInterval] = OMIT, interval_start: typing.Optional[dt.datetime] = OMIT, interval_end: typing.Optional[dt.datetime] = OMIT, span_filters: typing.Optional[typing.Sequence[SpanFilterPublic]] = OMIT, trace_filters: typing.Optional[typing.Sequence[TraceFilterPublic]] = OMIT, thread_filters: typing.Optional[typing.Sequence[TraceThreadFilterPublic]] = OMIT, breakdown: typing.Optional[BreakdownConfigPublic] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ProjectMetricResponsePublic]: """ Gets specified metrics for a project Parameters ---------- id : str metric_type : typing.Optional[ProjectMetricRequestPublicMetricType] interval : typing.Optional[ProjectMetricRequestPublicInterval] interval_start : typing.Optional[dt.datetime] interval_end : typing.Optional[dt.datetime] span_filters : typing.Optional[typing.Sequence[SpanFilterPublic]] trace_filters : typing.Optional[typing.Sequence[TraceFilterPublic]] thread_filters : typing.Optional[typing.Sequence[TraceThreadFilterPublic]] breakdown : typing.Optional[BreakdownConfigPublic] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProjectMetricResponsePublic] Project Metrics """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/projects/{jsonable_encoder(id)}/metrics", method="POST", json={ "metric_type": metric_type, "interval": interval, "interval_start": interval_start, "interval_end": interval_end, "span_filters": convert_and_respect_annotation_metadata( object_=span_filters, annotation=typing.Sequence[SpanFilterPublic], direction="write" ), "trace_filters": convert_and_respect_annotation_metadata( object_=trace_filters, annotation=typing.Sequence[TraceFilterPublic], direction="write" ), "thread_filters": convert_and_respect_annotation_metadata( object_=thread_filters, annotation=typing.Sequence[TraceThreadFilterPublic], direction="write" ), "breakdown": convert_and_respect_annotation_metadata( object_=breakdown, annotation=BreakdownConfigPublic, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectMetricResponsePublic, parse_obj_as( type_=ProjectMetricResponsePublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) 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(), ), ), ) 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_project_stats( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, name: typing.Optional[str] = None, sorting: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ProjectStatsSummary]: """ Get Project Stats Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] name : typing.Optional[str] sorting : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProjectStatsSummary] Project Stats """ _response = await self._client_wrapper.httpx_client.request( "v1/private/projects/stats", method="GET", params={ "page": page, "size": size, "name": name, "sorting": sorting, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectStatsSummary, parse_obj_as( type_=ProjectStatsSummary, # 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 retrieve_project( self, *, name: str, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ProjectDetailed]: """ Retrieve project Parameters ---------- name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProjectDetailed] Project resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/projects/retrieve", method="POST", json={ "name": name, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProjectDetailed, parse_obj_as( type_=ProjectDetailed, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) 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(), ), ), ) 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(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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)