# 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.unchecked_base_model import construct_type from ..types.prediction import Prediction # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawPredictionsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, project: typing.Optional[int] = None, task: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.List[Prediction]]: """ List all predictions and their IDs. Parameters ---------- project : typing.Optional[int] Filter predictions by project ID task : typing.Optional[int] Filter predictions by task ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.List[Prediction]] Predictions list """ _response = self._client_wrapper.httpx_client.request( "api/predictions/", method="GET", params={ "project": project, "task": task, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[Prediction], construct_type( type_=typing.List[Prediction], # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def create( self, *, model_version: typing.Optional[str] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, score: typing.Optional[float] = OMIT, task: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Prediction]: """ Create a prediction for a specific task. Parameters ---------- model_version : typing.Optional[str] Model version - tag for predictions that can be used to filter tasks in Data Manager, as well as select specific model version for showing preannotations in the labeling interface result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Prediction result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/predictions) score : typing.Optional[float] Prediction score. Can be used in Data Manager to sort task by model confidence. Task with the lowest score will be shown first. task : typing.Optional[int] Task ID for which the prediction is created request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Prediction] Created prediction """ _response = self._client_wrapper.httpx_client.request( "api/predictions/", method="POST", json={ "model_version": model_version, "result": result, "score": score, "task": task, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prediction, construct_type( type_=Prediction, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[Prediction]: """ Get details about a specific prediction by its ID. Parameters ---------- id : int Prediction ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Prediction] Prediction details """ _response = self._client_wrapper.httpx_client.request( f"api/predictions/{jsonable_encoder(id)}/", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prediction, construct_type( type_=Prediction, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]: """ Delete a prediction by prediction ID. Parameters ---------- id : int Prediction ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"api/predictions/{jsonable_encoder(id)}/", method="DELETE", request_options=request_options, ) 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 update( self, id: int, *, model_version: typing.Optional[str] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, score: typing.Optional[float] = OMIT, task: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Prediction]: """ Update prediction data by prediction ID. Parameters ---------- id : int Prediction ID model_version : typing.Optional[str] Model version - tag for predictions that can be used to filter tasks in Data Manager, as well as select specific model version for showing preannotations in the labeling interface result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Prediction result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/predictions) score : typing.Optional[float] Prediction score. Can be used in Data Manager to sort task by model confidence. Task with the lowest score will be shown first. task : typing.Optional[int] Task ID for which the prediction is created request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Prediction] Updated prediction """ _response = self._client_wrapper.httpx_client.request( f"api/predictions/{jsonable_encoder(id)}/", method="PATCH", json={ "model_version": model_version, "result": result, "score": score, "task": task, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prediction, construct_type( type_=Prediction, # 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) class AsyncRawPredictionsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, project: typing.Optional[int] = None, task: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.List[Prediction]]: """ List all predictions and their IDs. Parameters ---------- project : typing.Optional[int] Filter predictions by project ID task : typing.Optional[int] Filter predictions by task ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.List[Prediction]] Predictions list """ _response = await self._client_wrapper.httpx_client.request( "api/predictions/", method="GET", params={ "project": project, "task": task, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[Prediction], construct_type( type_=typing.List[Prediction], # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def create( self, *, model_version: typing.Optional[str] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, score: typing.Optional[float] = OMIT, task: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Prediction]: """ Create a prediction for a specific task. Parameters ---------- model_version : typing.Optional[str] Model version - tag for predictions that can be used to filter tasks in Data Manager, as well as select specific model version for showing preannotations in the labeling interface result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Prediction result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/predictions) score : typing.Optional[float] Prediction score. Can be used in Data Manager to sort task by model confidence. Task with the lowest score will be shown first. task : typing.Optional[int] Task ID for which the prediction is created request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Prediction] Created prediction """ _response = await self._client_wrapper.httpx_client.request( "api/predictions/", method="POST", json={ "model_version": model_version, "result": result, "score": score, "task": task, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prediction, construct_type( type_=Prediction, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get( self, id: int, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[Prediction]: """ Get details about a specific prediction by its ID. Parameters ---------- id : int Prediction ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Prediction] Prediction details """ _response = await self._client_wrapper.httpx_client.request( f"api/predictions/{jsonable_encoder(id)}/", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prediction, construct_type( type_=Prediction, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete( self, id: int, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete a prediction by prediction ID. Parameters ---------- id : int Prediction ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"api/predictions/{jsonable_encoder(id)}/", method="DELETE", request_options=request_options, ) 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 update( self, id: int, *, model_version: typing.Optional[str] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, score: typing.Optional[float] = OMIT, task: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Prediction]: """ Update prediction data by prediction ID. Parameters ---------- id : int Prediction ID model_version : typing.Optional[str] Model version - tag for predictions that can be used to filter tasks in Data Manager, as well as select specific model version for showing preannotations in the labeling interface result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Prediction result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/predictions) score : typing.Optional[float] Prediction score. Can be used in Data Manager to sort task by model confidence. Task with the lowest score will be shown first. task : typing.Optional[int] Task ID for which the prediction is created request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Prediction] Updated prediction """ _response = await self._client_wrapper.httpx_client.request( f"api/predictions/{jsonable_encoder(id)}/", method="PATCH", json={ "model_version": model_version, "result": result, "score": score, "task": task, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prediction, construct_type( type_=Prediction, # 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)