# This file was auto-generated by Fern from our API Definition. import contextlib 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.experiment_group_aggregations_response import ExperimentGroupAggregationsResponse from ..types.experiment_group_response import ExperimentGroupResponse from ..types.experiment_item import ExperimentItem from ..types.experiment_item_bulk_record_experiment_item_bulk_write_view import ( ExperimentItemBulkRecordExperimentItemBulkWriteView, ) from ..types.experiment_item_public import ExperimentItemPublic from ..types.experiment_page_public import ExperimentPagePublic from ..types.experiment_public import ExperimentPublic from ..types.experiment_score import ExperimentScore from ..types.experiment_score_write import ExperimentScoreWrite from ..types.experiment_update import ExperimentUpdate from ..types.experiment_update_status import ExperimentUpdateStatus from ..types.experiment_update_type import ExperimentUpdateType from ..types.feedback_score_names_public import FeedbackScoreNamesPublic from ..types.json_list_string_write import JsonListStringWrite from ..types.json_node import JsonNode from ..types.prompt_version_link_write import PromptVersionLinkWrite from .types.experiment_write_evaluation_method import ExperimentWriteEvaluationMethod from .types.experiment_write_status import ExperimentWriteStatus from .types.experiment_write_type import ExperimentWriteType # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawExperimentsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def batch_update_experiments( self, *, ids: typing.Sequence[str], update: ExperimentUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Update multiple experiments Parameters ---------- ids : typing.Sequence[str] List of experiment IDs to update (max 1000) update : ExperimentUpdate 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/experiments/batch", method="PATCH", json={ "ids": ids, "update": convert_and_respect_annotation_metadata( object_=update, annotation=ExperimentUpdate, 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 find_experiments( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, dataset_id: typing.Optional[str] = None, optimization_id: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_deleted: typing.Optional[bool] = None, prompt_id: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, filters: typing.Optional[str] = None, experiment_ids: typing.Optional[str] = None, force_sorting: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ExperimentPagePublic]: """ Find experiments Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] dataset_id : typing.Optional[str] optimization_id : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] dataset_deleted : typing.Optional[bool] prompt_id : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] sorting : typing.Optional[str] filters : typing.Optional[str] experiment_ids : typing.Optional[str] force_sorting : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentPagePublic] Experiments resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments", method="GET", params={ "page": page, "size": size, "datasetId": dataset_id, "optimization_id": optimization_id, "types": types, "name": name, "dataset_deleted": dataset_deleted, "prompt_id": prompt_id, "project_id": project_id, "project_deleted": project_deleted, "sorting": sorting, "filters": filters, "experiment_ids": experiment_ids, "force_sorting": force_sorting, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentPagePublic, parse_obj_as( type_=ExperimentPagePublic, # 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(), ), ), ) _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_experiment( self, *, id: typing.Optional[str] = OMIT, dataset_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, type: typing.Optional[ExperimentWriteType] = OMIT, evaluation_method: typing.Optional[ExperimentWriteEvaluationMethod] = OMIT, optimization_id: typing.Optional[str] = OMIT, status: typing.Optional[ExperimentWriteStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScoreWrite]] = OMIT, prompt_version: typing.Optional[PromptVersionLinkWrite] = OMIT, prompt_versions: typing.Optional[typing.Sequence[PromptVersionLinkWrite]] = OMIT, dataset_version_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Create experiment Parameters ---------- id : typing.Optional[str] dataset_name : typing.Optional[str] project_id : typing.Optional[str] Project ID. Takes precedence over project_name when both are provided. project_name : typing.Optional[str] Project name. Creates project if it doesn't exist. Ignored when project_id is provided. name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] tags : typing.Optional[typing.Sequence[str]] type : typing.Optional[ExperimentWriteType] evaluation_method : typing.Optional[ExperimentWriteEvaluationMethod] optimization_id : typing.Optional[str] status : typing.Optional[ExperimentWriteStatus] experiment_scores : typing.Optional[typing.Sequence[ExperimentScoreWrite]] prompt_version : typing.Optional[PromptVersionLinkWrite] prompt_versions : typing.Optional[typing.Sequence[PromptVersionLinkWrite]] dataset_version_id : typing.Optional[str] ID of the dataset version this experiment is linked to. If not provided at creation, experiment will be automatically linked to the latest version. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments", method="POST", json={ "id": id, "dataset_name": dataset_name, "project_id": project_id, "project_name": project_name, "name": name, "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "tags": tags, "type": type, "evaluation_method": evaluation_method, "optimization_id": optimization_id, "status": status, "experiment_scores": convert_and_respect_annotation_metadata( object_=experiment_scores, annotation=typing.Sequence[ExperimentScoreWrite], direction="write" ), "prompt_version": convert_and_respect_annotation_metadata( object_=prompt_version, annotation=PromptVersionLinkWrite, direction="write" ), "prompt_versions": convert_and_respect_annotation_metadata( object_=prompt_versions, annotation=typing.Sequence[PromptVersionLinkWrite], direction="write" ), "dataset_version_id": dataset_version_id, }, 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_experiment_items( self, *, experiment_items: typing.Sequence[ExperimentItem], request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Create experiment items Parameters ---------- experiment_items : typing.Sequence[ExperimentItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments/items", method="POST", json={ "experiment_items": convert_and_respect_annotation_metadata( object_=experiment_items, annotation=typing.Sequence[ExperimentItem], 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 delete_experiment_items( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete experiment items 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/experiments/items/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_experiments_by_id( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete experiments by id 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/experiments/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 experiment_items_bulk( self, *, experiment_name: str, dataset_name: str, items: typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView], experiment_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Record experiment items in bulk with traces, spans, and feedback scores. Maximum request size is 4MB. Parameters ---------- experiment_name : str dataset_name : str items : typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView] experiment_id : typing.Optional[str] Optional experiment ID. If provided, items will be added to the existing experiment and experimentName will be ignored. If not provided or experiment with that ID doesn't exist, a new experiment will be created with the given experimentName request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments/items/bulk", method="PUT", json={ "experiment_name": experiment_name, "dataset_name": dataset_name, "experiment_id": experiment_id, "items": convert_and_respect_annotation_metadata( object_=items, annotation=typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView], 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) 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 == 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(), ), ), ) 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 find_feedback_score_names( self, *, experiment_ids: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[FeedbackScoreNamesPublic]: """ Find Feedback Score names Parameters ---------- experiment_ids : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[FeedbackScoreNamesPublic] Feedback Scores resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments/feedback-scores/names", method="GET", params={ "experiment_ids": experiment_ids, }, 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 find_experiment_groups( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ExperimentGroupResponse]: """ Find experiments grouped by specified fields Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentGroupResponse] Experiment groups """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments/groups", method="GET", params={ "groups": groups, "types": types, "name": name, "project_id": project_id, "project_deleted": project_deleted, "filters": filters, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentGroupResponse, parse_obj_as( type_=ExperimentGroupResponse, # 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(), ), ), ) _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_experiment_groups_aggregations( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ExperimentGroupAggregationsResponse]: """ Find experiments grouped by specified fields with aggregation metrics Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentGroupAggregationsResponse] Experiment groups with aggregations """ _response = self._client_wrapper.httpx_client.request( "v1/private/experiments/groups/aggregations", method="GET", params={ "groups": groups, "types": types, "name": name, "project_id": project_id, "project_deleted": project_deleted, "filters": filters, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentGroupAggregationsResponse, parse_obj_as( type_=ExperimentGroupAggregationsResponse, # 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(), ), ), ) _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 finish_experiments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Finish experiments and trigger alert events 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/experiments/finish", 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) 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_experiment_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ExperimentPublic]: """ Get experiment by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentPublic] Experiment resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/experiments/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentPublic, parse_obj_as( type_=ExperimentPublic, # 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 update_experiment( self, id: str, *, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonNode] = 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, type: typing.Optional[ExperimentUpdateType] = OMIT, status: typing.Optional[ExperimentUpdateStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScore]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Update experiment by id Parameters ---------- id : str name : typing.Optional[str] metadata : typing.Optional[JsonNode] 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 type : typing.Optional[ExperimentUpdateType] status : typing.Optional[ExperimentUpdateStatus] The status of the experiment experiment_scores : typing.Optional[typing.Sequence[ExperimentScore]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/experiments/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "metadata": metadata, "tags": tags, "tags_to_add": tags_to_add, "tags_to_remove": tags_to_remove, "type": type, "status": status, "experiment_scores": convert_and_respect_annotation_metadata( object_=experiment_scores, annotation=typing.Sequence[ExperimentScore], 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) 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_experiment_item_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ExperimentItemPublic]: """ Get experiment item by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentItemPublic] Experiment item resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/experiments/items/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentItemPublic, parse_obj_as( type_=ExperimentItemPublic, # 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) @contextlib.contextmanager def stream_experiment_items( self, *, experiment_name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[HttpResponse[typing.Iterator[bytes]]]: """ Stream experiment items Parameters ---------- experiment_name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] truncate : typing.Optional[bool] Truncate image included in either input, output or metadata 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]]] Experiment items stream or error during process """ with self._client_wrapper.httpx_client.stream( "v1/private/experiments/items/stream", method="POST", json={ "experiment_name": experiment_name, "limit": limit, "last_retrieved_id": last_retrieved_id, "truncate": truncate, }, 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() _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() @contextlib.contextmanager def stream_experiments( self, *, name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[HttpResponse[typing.Iterator[bytes]]]: """ Stream experiments Parameters ---------- name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] 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]]] Experiments stream or error during process """ with self._client_wrapper.httpx_client.stream( "v1/private/experiments/stream", method="POST", json={ "name": name, "limit": limit, "last_retrieved_id": last_retrieved_id, }, 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() _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() class AsyncRawExperimentsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def batch_update_experiments( self, *, ids: typing.Sequence[str], update: ExperimentUpdate, merge_tags: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Update multiple experiments Parameters ---------- ids : typing.Sequence[str] List of experiment IDs to update (max 1000) update : ExperimentUpdate 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/experiments/batch", method="PATCH", json={ "ids": ids, "update": convert_and_respect_annotation_metadata( object_=update, annotation=ExperimentUpdate, 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 find_experiments( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, dataset_id: typing.Optional[str] = None, optimization_id: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_deleted: typing.Optional[bool] = None, prompt_id: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, sorting: typing.Optional[str] = None, filters: typing.Optional[str] = None, experiment_ids: typing.Optional[str] = None, force_sorting: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ExperimentPagePublic]: """ Find experiments Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] dataset_id : typing.Optional[str] optimization_id : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] dataset_deleted : typing.Optional[bool] prompt_id : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] sorting : typing.Optional[str] filters : typing.Optional[str] experiment_ids : typing.Optional[str] force_sorting : typing.Optional[bool] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentPagePublic] Experiments resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments", method="GET", params={ "page": page, "size": size, "datasetId": dataset_id, "optimization_id": optimization_id, "types": types, "name": name, "dataset_deleted": dataset_deleted, "prompt_id": prompt_id, "project_id": project_id, "project_deleted": project_deleted, "sorting": sorting, "filters": filters, "experiment_ids": experiment_ids, "force_sorting": force_sorting, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentPagePublic, parse_obj_as( type_=ExperimentPagePublic, # 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(), ), ), ) _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_experiment( self, *, id: typing.Optional[str] = OMIT, dataset_name: typing.Optional[str] = OMIT, project_id: typing.Optional[str] = OMIT, project_name: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, tags: typing.Optional[typing.Sequence[str]] = OMIT, type: typing.Optional[ExperimentWriteType] = OMIT, evaluation_method: typing.Optional[ExperimentWriteEvaluationMethod] = OMIT, optimization_id: typing.Optional[str] = OMIT, status: typing.Optional[ExperimentWriteStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScoreWrite]] = OMIT, prompt_version: typing.Optional[PromptVersionLinkWrite] = OMIT, prompt_versions: typing.Optional[typing.Sequence[PromptVersionLinkWrite]] = OMIT, dataset_version_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Create experiment Parameters ---------- id : typing.Optional[str] dataset_name : typing.Optional[str] project_id : typing.Optional[str] Project ID. Takes precedence over project_name when both are provided. project_name : typing.Optional[str] Project name. Creates project if it doesn't exist. Ignored when project_id is provided. name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] tags : typing.Optional[typing.Sequence[str]] type : typing.Optional[ExperimentWriteType] evaluation_method : typing.Optional[ExperimentWriteEvaluationMethod] optimization_id : typing.Optional[str] status : typing.Optional[ExperimentWriteStatus] experiment_scores : typing.Optional[typing.Sequence[ExperimentScoreWrite]] prompt_version : typing.Optional[PromptVersionLinkWrite] prompt_versions : typing.Optional[typing.Sequence[PromptVersionLinkWrite]] dataset_version_id : typing.Optional[str] ID of the dataset version this experiment is linked to. If not provided at creation, experiment will be automatically linked to the latest version. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments", method="POST", json={ "id": id, "dataset_name": dataset_name, "project_id": project_id, "project_name": project_name, "name": name, "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "tags": tags, "type": type, "evaluation_method": evaluation_method, "optimization_id": optimization_id, "status": status, "experiment_scores": convert_and_respect_annotation_metadata( object_=experiment_scores, annotation=typing.Sequence[ExperimentScoreWrite], direction="write" ), "prompt_version": convert_and_respect_annotation_metadata( object_=prompt_version, annotation=PromptVersionLinkWrite, direction="write" ), "prompt_versions": convert_and_respect_annotation_metadata( object_=prompt_versions, annotation=typing.Sequence[PromptVersionLinkWrite], direction="write" ), "dataset_version_id": dataset_version_id, }, 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_experiment_items( self, *, experiment_items: typing.Sequence[ExperimentItem], request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Create experiment items Parameters ---------- experiment_items : typing.Sequence[ExperimentItem] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments/items", method="POST", json={ "experiment_items": convert_and_respect_annotation_metadata( object_=experiment_items, annotation=typing.Sequence[ExperimentItem], 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 delete_experiment_items( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete experiment items 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/experiments/items/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_experiments_by_id( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete experiments by id 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/experiments/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 experiment_items_bulk( self, *, experiment_name: str, dataset_name: str, items: typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView], experiment_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Record experiment items in bulk with traces, spans, and feedback scores. Maximum request size is 4MB. Parameters ---------- experiment_name : str dataset_name : str items : typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView] experiment_id : typing.Optional[str] Optional experiment ID. If provided, items will be added to the existing experiment and experimentName will be ignored. If not provided or experiment with that ID doesn't exist, a new experiment will be created with the given experimentName request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments/items/bulk", method="PUT", json={ "experiment_name": experiment_name, "dataset_name": dataset_name, "experiment_id": experiment_id, "items": convert_and_respect_annotation_metadata( object_=items, annotation=typing.Sequence[ExperimentItemBulkRecordExperimentItemBulkWriteView], 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) 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 == 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(), ), ), ) 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 find_feedback_score_names( self, *, experiment_ids: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[FeedbackScoreNamesPublic]: """ Find Feedback Score names Parameters ---------- experiment_ids : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[FeedbackScoreNamesPublic] Feedback Scores resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments/feedback-scores/names", method="GET", params={ "experiment_ids": experiment_ids, }, 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 find_experiment_groups( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ExperimentGroupResponse]: """ Find experiments grouped by specified fields Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentGroupResponse] Experiment groups """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments/groups", method="GET", params={ "groups": groups, "types": types, "name": name, "project_id": project_id, "project_deleted": project_deleted, "filters": filters, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentGroupResponse, parse_obj_as( type_=ExperimentGroupResponse, # 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(), ), ), ) _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_experiment_groups_aggregations( self, *, groups: typing.Optional[str] = None, types: typing.Optional[str] = None, name: typing.Optional[str] = None, project_id: typing.Optional[str] = None, project_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ExperimentGroupAggregationsResponse]: """ Find experiments grouped by specified fields with aggregation metrics Parameters ---------- groups : typing.Optional[str] types : typing.Optional[str] name : typing.Optional[str] project_id : typing.Optional[str] project_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentGroupAggregationsResponse] Experiment groups with aggregations """ _response = await self._client_wrapper.httpx_client.request( "v1/private/experiments/groups/aggregations", method="GET", params={ "groups": groups, "types": types, "name": name, "project_id": project_id, "project_deleted": project_deleted, "filters": filters, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentGroupAggregationsResponse, parse_obj_as( type_=ExperimentGroupAggregationsResponse, # 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(), ), ), ) _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 finish_experiments( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Finish experiments and trigger alert events 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/experiments/finish", 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) 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_experiment_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ExperimentPublic]: """ Get experiment by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentPublic] Experiment resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/experiments/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentPublic, parse_obj_as( type_=ExperimentPublic, # 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 update_experiment( self, id: str, *, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonNode] = 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, type: typing.Optional[ExperimentUpdateType] = OMIT, status: typing.Optional[ExperimentUpdateStatus] = OMIT, experiment_scores: typing.Optional[typing.Sequence[ExperimentScore]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Update experiment by id Parameters ---------- id : str name : typing.Optional[str] metadata : typing.Optional[JsonNode] 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 type : typing.Optional[ExperimentUpdateType] status : typing.Optional[ExperimentUpdateStatus] The status of the experiment experiment_scores : typing.Optional[typing.Sequence[ExperimentScore]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/experiments/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "metadata": metadata, "tags": tags, "tags_to_add": tags_to_add, "tags_to_remove": tags_to_remove, "type": type, "status": status, "experiment_scores": convert_and_respect_annotation_metadata( object_=experiment_scores, annotation=typing.Sequence[ExperimentScore], 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) 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_experiment_item_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ExperimentItemPublic]: """ Get experiment item by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentItemPublic] Experiment item resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/experiments/items/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentItemPublic, parse_obj_as( type_=ExperimentItemPublic, # 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) @contextlib.asynccontextmanager async def stream_experiment_items( self, *, experiment_name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, truncate: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]]: """ Stream experiment items Parameters ---------- experiment_name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] truncate : typing.Optional[bool] Truncate image included in either input, output or metadata 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]]] Experiment items stream or error during process """ async with self._client_wrapper.httpx_client.stream( "v1/private/experiments/items/stream", method="POST", json={ "experiment_name": experiment_name, "limit": limit, "last_retrieved_id": last_retrieved_id, "truncate": truncate, }, 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() _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() @contextlib.asynccontextmanager async def stream_experiments( self, *, name: str, limit: typing.Optional[int] = OMIT, last_retrieved_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]]: """ Stream experiments Parameters ---------- name : str limit : typing.Optional[int] last_retrieved_id : typing.Optional[str] 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]]] Experiments stream or error during process """ async with self._client_wrapper.httpx_client.stream( "v1/private/experiments/stream", method="POST", json={ "name": name, "limit": limit, "last_retrieved_id": last_retrieved_id, }, 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() _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()