# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from json.decoder import JSONDecodeError from ..commons.errors.access_denied_error import AccessDeniedError from ..commons.errors.error import Error from ..commons.errors.method_not_allowed_error import MethodNotAllowedError from ..commons.errors.not_found_error import NotFoundError from ..commons.errors.unauthorized_error import UnauthorizedError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.datetime_utils import serialize_datetime from ..core.http_response import AsyncHttpResponse, HttpResponse from ..core.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from .types.experiment_items_response import ExperimentItemsResponse from .types.experiments_response import ExperimentsResponse class RawExperimentsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, from_start_time: dt.datetime, fields: typing.Optional[str] = None, limit: typing.Optional[int] = None, score_limit: typing.Optional[int] = None, cursor: typing.Optional[str] = None, to_start_time: typing.Optional[dt.datetime] = None, id: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_id: typing.Optional[str] = None, filter: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ExperimentsResponse]: """ List experiments with cursor-based pagination. Results are ordered by latest experiment activity descending. Parameters ---------- from_start_time : dt.datetime Retrieve only experiments on or after this datetime. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: `core`, `metadata`, `scores`. If omitted, `core` is returned. limit : typing.Optional[int] Number of experiments to return per page. Maximum 100, default 50. score_limit : typing.Optional[int] Number of scores to return per experiment when `fields=scores` is requested. Maximum 50, default 50. cursor : typing.Optional[str] Versioned base64url cursor from the previous response page. to_start_time : typing.Optional[dt.datetime] Retrieve only experiments before this datetime. id : typing.Optional[str] Comma-separated list of experiment IDs. name : typing.Optional[str] Comma-separated list of experiment names. dataset_id : typing.Optional[str] Comma-separated list of dataset IDs. filter : typing.Optional[str] JSON string containing an array of structured filter conditions. Supported columns are `id`, `name`, and `datasetId`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentsResponse] """ _response = self._client_wrapper.httpx_client.request( "api/public/experiments", method="GET", params={ "fields": fields, "limit": limit, "scoreLimit": score_limit, "cursor": cursor, "fromStartTime": serialize_datetime(from_start_time), "toStartTime": serialize_datetime(to_start_time) if to_start_time is not None else None, "id": id, "name": name, "datasetId": dataset_id, "filter": filter, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentsResponse, parse_obj_as( type_=ExperimentsResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 list_items( self, *, from_start_time: dt.datetime, fields: typing.Optional[str] = None, limit: typing.Optional[int] = None, score_limit: typing.Optional[int] = None, cursor: typing.Optional[str] = None, to_start_time: typing.Optional[dt.datetime] = None, experiment_id: typing.Optional[str] = None, experiment_name: typing.Optional[str] = None, experiment_item_id: typing.Optional[str] = None, dataset_id: typing.Optional[str] = None, filter: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ExperimentItemsResponse]: """ List experiment items with cursor-based pagination. Use this endpoint to export experiment item inputs, outputs, expected outputs, metadata, and optionally item/trace scores. Results are ordered by time descending. Parameters ---------- from_start_time : dt.datetime Retrieve only experiment items started on or after this datetime. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: `core`, `dataset`, `io`, `metadata`, `itemMetadata`, `experimentMetadata`, `scores`. If omitted, `core,dataset` is returned. limit : typing.Optional[int] Number of experiment items to return per page. Maximum 100, default 50. score_limit : typing.Optional[int] Number of scores to return per experiment item when `fields=scores` is requested. Maximum 50, default 50. cursor : typing.Optional[str] Versioned base64url cursor from the previous response page. to_start_time : typing.Optional[dt.datetime] Retrieve only experiment items started before this datetime. experiment_id : typing.Optional[str] Comma-separated list of experiment IDs. experiment_name : typing.Optional[str] Comma-separated list of experiment names. experiment_item_id : typing.Optional[str] Comma-separated list of experiment item IDs. dataset_id : typing.Optional[str] Comma-separated list of dataset IDs. filter : typing.Optional[str] JSON string containing an array of structured filter conditions. Supported columns are `experimentId`, `experimentName`, `experimentItemId`, and `datasetId`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExperimentItemsResponse] """ _response = self._client_wrapper.httpx_client.request( "api/public/experiment-items", method="GET", params={ "fields": fields, "limit": limit, "scoreLimit": score_limit, "cursor": cursor, "fromStartTime": serialize_datetime(from_start_time), "toStartTime": serialize_datetime(to_start_time) if to_start_time is not None else None, "experimentId": experiment_id, "experimentName": experiment_name, "experimentItemId": experiment_item_id, "datasetId": dataset_id, "filter": filter, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentItemsResponse, parse_obj_as( type_=ExperimentItemsResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 AsyncRawExperimentsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, from_start_time: dt.datetime, fields: typing.Optional[str] = None, limit: typing.Optional[int] = None, score_limit: typing.Optional[int] = None, cursor: typing.Optional[str] = None, to_start_time: typing.Optional[dt.datetime] = None, id: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_id: typing.Optional[str] = None, filter: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ExperimentsResponse]: """ List experiments with cursor-based pagination. Results are ordered by latest experiment activity descending. Parameters ---------- from_start_time : dt.datetime Retrieve only experiments on or after this datetime. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: `core`, `metadata`, `scores`. If omitted, `core` is returned. limit : typing.Optional[int] Number of experiments to return per page. Maximum 100, default 50. score_limit : typing.Optional[int] Number of scores to return per experiment when `fields=scores` is requested. Maximum 50, default 50. cursor : typing.Optional[str] Versioned base64url cursor from the previous response page. to_start_time : typing.Optional[dt.datetime] Retrieve only experiments before this datetime. id : typing.Optional[str] Comma-separated list of experiment IDs. name : typing.Optional[str] Comma-separated list of experiment names. dataset_id : typing.Optional[str] Comma-separated list of dataset IDs. filter : typing.Optional[str] JSON string containing an array of structured filter conditions. Supported columns are `id`, `name`, and `datasetId`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentsResponse] """ _response = await self._client_wrapper.httpx_client.request( "api/public/experiments", method="GET", params={ "fields": fields, "limit": limit, "scoreLimit": score_limit, "cursor": cursor, "fromStartTime": serialize_datetime(from_start_time), "toStartTime": serialize_datetime(to_start_time) if to_start_time is not None else None, "id": id, "name": name, "datasetId": dataset_id, "filter": filter, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentsResponse, parse_obj_as( type_=ExperimentsResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 list_items( self, *, from_start_time: dt.datetime, fields: typing.Optional[str] = None, limit: typing.Optional[int] = None, score_limit: typing.Optional[int] = None, cursor: typing.Optional[str] = None, to_start_time: typing.Optional[dt.datetime] = None, experiment_id: typing.Optional[str] = None, experiment_name: typing.Optional[str] = None, experiment_item_id: typing.Optional[str] = None, dataset_id: typing.Optional[str] = None, filter: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ExperimentItemsResponse]: """ List experiment items with cursor-based pagination. Use this endpoint to export experiment item inputs, outputs, expected outputs, metadata, and optionally item/trace scores. Results are ordered by time descending. Parameters ---------- from_start_time : dt.datetime Retrieve only experiment items started on or after this datetime. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: `core`, `dataset`, `io`, `metadata`, `itemMetadata`, `experimentMetadata`, `scores`. If omitted, `core,dataset` is returned. limit : typing.Optional[int] Number of experiment items to return per page. Maximum 100, default 50. score_limit : typing.Optional[int] Number of scores to return per experiment item when `fields=scores` is requested. Maximum 50, default 50. cursor : typing.Optional[str] Versioned base64url cursor from the previous response page. to_start_time : typing.Optional[dt.datetime] Retrieve only experiment items started before this datetime. experiment_id : typing.Optional[str] Comma-separated list of experiment IDs. experiment_name : typing.Optional[str] Comma-separated list of experiment names. experiment_item_id : typing.Optional[str] Comma-separated list of experiment item IDs. dataset_id : typing.Optional[str] Comma-separated list of dataset IDs. filter : typing.Optional[str] JSON string containing an array of structured filter conditions. Supported columns are `experimentId`, `experimentName`, `experimentItemId`, and `datasetId`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExperimentItemsResponse] """ _response = await self._client_wrapper.httpx_client.request( "api/public/experiment-items", method="GET", params={ "fields": fields, "limit": limit, "scoreLimit": score_limit, "cursor": cursor, "fromStartTime": serialize_datetime(from_start_time), "toStartTime": serialize_datetime(to_start_time) if to_start_time is not None else None, "experimentId": experiment_id, "experimentName": experiment_name, "experimentItemId": experiment_item_id, "datasetId": dataset_id, "filter": filter, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExperimentItemsResponse, parse_obj_as( type_=ExperimentItemsResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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, )