# 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.not_found_error import NotFoundError from ..errors.not_implemented_error import NotImplementedError from ..types.json_list_string_write import JsonListStringWrite from ..types.optimization_page_public import OptimizationPagePublic from ..types.optimization_public import OptimizationPublic from ..types.optimization_studio_config_write import OptimizationStudioConfigWrite from ..types.optimization_studio_log import OptimizationStudioLog from ..types.optimization_write_status import OptimizationWriteStatus from .types.optimization_update_status import OptimizationUpdateStatus # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawOptimizationsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def cancel_studio_optimizations( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Cancel Studio optimizations 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/optimizations/studio/{jsonable_encoder(id)}/cancel", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 501: raise NotImplementedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def find_optimizations( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, dataset_id: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_name: typing.Optional[str] = None, dataset_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[OptimizationPagePublic]: """ Find optimizations Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] dataset_id : typing.Optional[str] name : typing.Optional[str] dataset_name : typing.Optional[str] dataset_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[OptimizationPagePublic] Optimizations resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/optimizations", method="GET", params={ "page": page, "size": size, "dataset_id": dataset_id, "name": name, "dataset_name": dataset_name, "dataset_deleted": dataset_deleted, "filters": filters, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( OptimizationPagePublic, parse_obj_as( type_=OptimizationPagePublic, # 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_optimization( self, *, dataset_name: str, objective_name: str, status: OptimizationWriteStatus, id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, studio_config: typing.Optional[OptimizationStudioConfigWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Create optimization Parameters ---------- dataset_name : str objective_name : str status : OptimizationWriteStatus id : typing.Optional[str] name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] studio_config : typing.Optional[OptimizationStudioConfigWrite] last_updated_at : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/optimizations", method="POST", json={ "id": id, "name": name, "dataset_name": dataset_name, "objective_name": objective_name, "status": status, "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "studio_config": convert_and_respect_annotation_metadata( object_=studio_config, annotation=OptimizationStudioConfigWrite, direction="write" ), "last_updated_at": last_updated_at, }, 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 upsert_optimization( self, *, dataset_name: str, objective_name: str, status: OptimizationWriteStatus, id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, studio_config: typing.Optional[OptimizationStudioConfigWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Upsert optimization Parameters ---------- dataset_name : str objective_name : str status : OptimizationWriteStatus id : typing.Optional[str] name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] studio_config : typing.Optional[OptimizationStudioConfigWrite] last_updated_at : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/optimizations", method="PUT", json={ "id": id, "name": name, "dataset_name": dataset_name, "objective_name": objective_name, "status": status, "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "studio_config": convert_and_respect_annotation_metadata( object_=studio_config, annotation=OptimizationStudioConfigWrite, direction="write" ), "last_updated_at": last_updated_at, }, 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_optimizations_by_id( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete optimizations 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/optimizations/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 get_optimization_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[OptimizationPublic]: """ Get optimization by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[OptimizationPublic] Optimization resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/optimizations/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( OptimizationPublic, parse_obj_as( type_=OptimizationPublic, # 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_optimizations_by_id( self, id: str, *, name: typing.Optional[str] = OMIT, status: typing.Optional[OptimizationUpdateStatus] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Update optimization by id Parameters ---------- id : str name : typing.Optional[str] status : typing.Optional[OptimizationUpdateStatus] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/optimizations/{jsonable_encoder(id)}", method="PUT", json={ "name": name, "status": status, }, 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 get_studio_optimization_logs( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[OptimizationStudioLog]: """ Get presigned S3 URL for downloading optimization logs Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[OptimizationStudioLog] Logs response """ _response = self._client_wrapper.httpx_client.request( f"v1/private/optimizations/studio/{jsonable_encoder(id)}/logs", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( OptimizationStudioLog, parse_obj_as( type_=OptimizationStudioLog, # 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) class AsyncRawOptimizationsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def cancel_studio_optimizations( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Cancel Studio optimizations 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/optimizations/studio/{jsonable_encoder(id)}/cancel", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 501: raise NotImplementedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def find_optimizations( self, *, page: typing.Optional[int] = None, size: typing.Optional[int] = None, dataset_id: typing.Optional[str] = None, name: typing.Optional[str] = None, dataset_name: typing.Optional[str] = None, dataset_deleted: typing.Optional[bool] = None, filters: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[OptimizationPagePublic]: """ Find optimizations Parameters ---------- page : typing.Optional[int] size : typing.Optional[int] dataset_id : typing.Optional[str] name : typing.Optional[str] dataset_name : typing.Optional[str] dataset_deleted : typing.Optional[bool] filters : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[OptimizationPagePublic] Optimizations resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/optimizations", method="GET", params={ "page": page, "size": size, "dataset_id": dataset_id, "name": name, "dataset_name": dataset_name, "dataset_deleted": dataset_deleted, "filters": filters, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( OptimizationPagePublic, parse_obj_as( type_=OptimizationPagePublic, # 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_optimization( self, *, dataset_name: str, objective_name: str, status: OptimizationWriteStatus, id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, studio_config: typing.Optional[OptimizationStudioConfigWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Create optimization Parameters ---------- dataset_name : str objective_name : str status : OptimizationWriteStatus id : typing.Optional[str] name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] studio_config : typing.Optional[OptimizationStudioConfigWrite] last_updated_at : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/optimizations", method="POST", json={ "id": id, "name": name, "dataset_name": dataset_name, "objective_name": objective_name, "status": status, "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "studio_config": convert_and_respect_annotation_metadata( object_=studio_config, annotation=OptimizationStudioConfigWrite, direction="write" ), "last_updated_at": last_updated_at, }, 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 upsert_optimization( self, *, dataset_name: str, objective_name: str, status: OptimizationWriteStatus, id: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, metadata: typing.Optional[JsonListStringWrite] = OMIT, studio_config: typing.Optional[OptimizationStudioConfigWrite] = OMIT, last_updated_at: typing.Optional[dt.datetime] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Upsert optimization Parameters ---------- dataset_name : str objective_name : str status : OptimizationWriteStatus id : typing.Optional[str] name : typing.Optional[str] metadata : typing.Optional[JsonListStringWrite] studio_config : typing.Optional[OptimizationStudioConfigWrite] last_updated_at : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/optimizations", method="PUT", json={ "id": id, "name": name, "dataset_name": dataset_name, "objective_name": objective_name, "status": status, "metadata": convert_and_respect_annotation_metadata( object_=metadata, annotation=JsonListStringWrite, direction="write" ), "studio_config": convert_and_respect_annotation_metadata( object_=studio_config, annotation=OptimizationStudioConfigWrite, direction="write" ), "last_updated_at": last_updated_at, }, 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_optimizations_by_id( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete optimizations 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/optimizations/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 get_optimization_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[OptimizationPublic]: """ Get optimization by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[OptimizationPublic] Optimization resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/optimizations/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( OptimizationPublic, parse_obj_as( type_=OptimizationPublic, # 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_optimizations_by_id( self, id: str, *, name: typing.Optional[str] = OMIT, status: typing.Optional[OptimizationUpdateStatus] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Update optimization by id Parameters ---------- id : str name : typing.Optional[str] status : typing.Optional[OptimizationUpdateStatus] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/optimizations/{jsonable_encoder(id)}", method="PUT", json={ "name": name, "status": status, }, 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 get_studio_optimization_logs( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[OptimizationStudioLog]: """ Get presigned S3 URL for downloading optimization logs Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[OptimizationStudioLog] Logs response """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/optimizations/studio/{jsonable_encoder(id)}/logs", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( OptimizationStudioLog, parse_obj_as( type_=OptimizationStudioLog, # 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)