# This file was auto-generated by Fern from our API Definition. 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.request_options import RequestOptions from ..core.serialization import convert_and_respect_annotation_metadata from ..core.unchecked_base_model import construct_type from ..errors.bad_request_error import BadRequestError from ..errors.forbidden_error import ForbiddenError from ..errors.internal_server_error import InternalServerError from ..errors.not_found_error import NotFoundError from ..errors.service_unavailable_error import ServiceUnavailableError from ..errors.unauthorized_error import UnauthorizedError from .finetuning.types.create_finetuned_model_response import CreateFinetunedModelResponse from .finetuning.types.delete_finetuned_model_response import DeleteFinetunedModelResponse from .finetuning.types.finetuned_model import FinetunedModel from .finetuning.types.get_finetuned_model_response import GetFinetunedModelResponse from .finetuning.types.list_events_response import ListEventsResponse from .finetuning.types.list_finetuned_models_response import ListFinetunedModelsResponse from .finetuning.types.list_training_step_metrics_response import ListTrainingStepMetricsResponse from .finetuning.types.settings import Settings from .finetuning.types.status import Status from .finetuning.types.update_finetuned_model_response import UpdateFinetunedModelResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawFinetuningClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list_finetuned_models( self, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ListFinetunedModelsResponse]: """ Returns a list of fine-tuned models that the user has access to. Parameters ---------- page_size : typing.Optional[int] Maximum number of results to be returned by the server. If 0, defaults to 50. page_token : typing.Optional[str] Request a specific page of the list results. order_by : typing.Optional[str] Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ListFinetunedModelsResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListFinetunedModelsResponse, construct_type( type_=ListFinetunedModelsResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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_finetuned_model( self, *, request: FinetunedModel, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[CreateFinetunedModelResponse]: """ Creates a new fine-tuned model. The model will be trained on the dataset specified in the request body. The training process may take some time, and the model will be available once the training is complete. Parameters ---------- request : FinetunedModel request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[CreateFinetunedModelResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="POST", json=convert_and_respect_annotation_metadata(object_=request, annotation=FinetunedModel, direction="write"), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateFinetunedModelResponse, construct_type( type_=CreateFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[GetFinetunedModelResponse]: """ Retrieve a fine-tuned model by its ID. Parameters ---------- id : str The fine-tuned model ID. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetFinetunedModelResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetFinetunedModelResponse, construct_type( type_=GetFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteFinetunedModelResponse]: """ Deletes a fine-tuned model. The model will be removed from the system and will no longer be available for use. This operation is irreversible. Parameters ---------- id : str The fine-tuned model ID. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteFinetunedModelResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteFinetunedModelResponse, construct_type( type_=DeleteFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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_finetuned_model( self, id: str, *, name: str, settings: Settings, status: typing.Optional[Status] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[UpdateFinetunedModelResponse]: """ Updates the fine-tuned model with the given ID. The model will be updated with the new settings and name provided in the request body. Parameters ---------- id : str FinetunedModel ID. name : str FinetunedModel name (e.g. `foobar`). settings : Settings FinetunedModel settings such as dataset, hyperparameters... status : typing.Optional[Status] Current stage in the life-cycle of the fine-tuned model. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[UpdateFinetunedModelResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "settings": convert_and_respect_annotation_metadata( object_=settings, annotation=Settings, direction="write" ), "status": status, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( UpdateFinetunedModelResponse, construct_type( type_=UpdateFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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 list_events( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ListEventsResponse]: """ Returns a list of events that occurred during the life-cycle of the fine-tuned model. The events are ordered by creation time, with the most recent event first. The list can be paginated using `page_size` and `page_token` parameters. Parameters ---------- finetuned_model_id : str The parent fine-tuned model ID. page_size : typing.Optional[int] Maximum number of results to be returned by the server. If 0, defaults to 50. page_token : typing.Optional[str] Request a specific page of the list results. order_by : typing.Optional[str] Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ListEventsResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/events", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListEventsResponse, construct_type( type_=ListEventsResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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 list_training_step_metrics( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ListTrainingStepMetricsResponse]: """ Returns a list of metrics measured during the training of a fine-tuned model. The metrics are ordered by step number, with the most recent step first. The list can be paginated using `page_size` and `page_token` parameters. Parameters ---------- finetuned_model_id : str The parent fine-tuned model ID. page_size : typing.Optional[int] Maximum number of results to be returned by the server. If 0, defaults to 50. page_token : typing.Optional[str] Request a specific page of the list results. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ListTrainingStepMetricsResponse] A successful response. """ _response = self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/training-step-metrics", method="GET", params={ "page_size": page_size, "page_token": page_token, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListTrainingStepMetricsResponse, construct_type( type_=ListTrainingStepMetricsResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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 AsyncRawFinetuningClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list_finetuned_models( self, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ListFinetunedModelsResponse]: """ Returns a list of fine-tuned models that the user has access to. Parameters ---------- page_size : typing.Optional[int] Maximum number of results to be returned by the server. If 0, defaults to 50. page_token : typing.Optional[str] Request a specific page of the list results. order_by : typing.Optional[str] Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ListFinetunedModelsResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListFinetunedModelsResponse, construct_type( type_=ListFinetunedModelsResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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_finetuned_model( self, *, request: FinetunedModel, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[CreateFinetunedModelResponse]: """ Creates a new fine-tuned model. The model will be trained on the dataset specified in the request body. The training process may take some time, and the model will be available once the training is complete. Parameters ---------- request : FinetunedModel request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[CreateFinetunedModelResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( "v1/finetuning/finetuned-models", method="POST", json=convert_and_respect_annotation_metadata(object_=request, annotation=FinetunedModel, direction="write"), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreateFinetunedModelResponse, construct_type( type_=CreateFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[GetFinetunedModelResponse]: """ Retrieve a fine-tuned model by its ID. Parameters ---------- id : str The fine-tuned model ID. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetFinetunedModelResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetFinetunedModelResponse, construct_type( type_=GetFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete_finetuned_model( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteFinetunedModelResponse]: """ Deletes a fine-tuned model. The model will be removed from the system and will no longer be available for use. This operation is irreversible. Parameters ---------- id : str The fine-tuned model ID. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteFinetunedModelResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteFinetunedModelResponse, construct_type( type_=DeleteFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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_finetuned_model( self, id: str, *, name: str, settings: Settings, status: typing.Optional[Status] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[UpdateFinetunedModelResponse]: """ Updates the fine-tuned model with the given ID. The model will be updated with the new settings and name provided in the request body. Parameters ---------- id : str FinetunedModel ID. name : str FinetunedModel name (e.g. `foobar`). settings : Settings FinetunedModel settings such as dataset, hyperparameters... status : typing.Optional[Status] Current stage in the life-cycle of the fine-tuned model. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[UpdateFinetunedModelResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(id)}", method="PATCH", json={ "name": name, "settings": convert_and_respect_annotation_metadata( object_=settings, annotation=Settings, direction="write" ), "status": status, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( UpdateFinetunedModelResponse, construct_type( type_=UpdateFinetunedModelResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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 list_events( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, order_by: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ListEventsResponse]: """ Returns a list of events that occurred during the life-cycle of the fine-tuned model. The events are ordered by creation time, with the most recent event first. The list can be paginated using `page_size` and `page_token` parameters. Parameters ---------- finetuned_model_id : str The parent fine-tuned model ID. page_size : typing.Optional[int] Maximum number of results to be returned by the server. If 0, defaults to 50. page_token : typing.Optional[str] Request a specific page of the list results. order_by : typing.Optional[str] Comma separated list of fields. For example: "created_at,name". The default sorting order is ascending. To specify descending order for a field, append " desc" to the field name. For example: "created_at desc,name". Supported sorting fields: - created_at (default) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ListEventsResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/events", method="GET", params={ "page_size": page_size, "page_token": page_token, "order_by": order_by, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListEventsResponse, construct_type( type_=ListEventsResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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 list_training_step_metrics( self, finetuned_model_id: str, *, page_size: typing.Optional[int] = None, page_token: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ListTrainingStepMetricsResponse]: """ Returns a list of metrics measured during the training of a fine-tuned model. The metrics are ordered by step number, with the most recent step first. The list can be paginated using `page_size` and `page_token` parameters. Parameters ---------- finetuned_model_id : str The parent fine-tuned model ID. page_size : typing.Optional[int] Maximum number of results to be returned by the server. If 0, defaults to 50. page_token : typing.Optional[str] Request a specific page of the list results. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ListTrainingStepMetricsResponse] A successful response. """ _response = await self._client_wrapper.httpx_client.request( f"v1/finetuning/finetuned-models/{jsonable_encoder(finetuned_model_id)}/training-step-metrics", method="GET", params={ "page_size": page_size, "page_token": page_token, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ListTrainingStepMetricsResponse, construct_type( type_=ListTrainingStepMetricsResponse, # 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( 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)