# 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.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 ..types.annotation import Annotation from ..types.last_action_enum import LastActionEnum from ..types.selected_items_request import SelectedItemsRequest from .types.create_bulk_annotations_response_item import CreateBulkAnnotationsResponseItem from .types.delete_bulk_annotations_response import DeleteBulkAnnotationsResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawAnnotationsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def delete_bulk( self, *, ids: typing.Sequence[int], project: int, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteBulkAnnotationsResponse]: """ Delete multiple annotations by their IDs. The deletion is processed synchronously. Returns the count of deleted annotations in the response. Parameters ---------- ids : typing.Sequence[int] List of annotation IDs to delete project : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteBulkAnnotationsResponse] Annotations deleted successfully """ _response = self._client_wrapper.httpx_client.request( "api/annotations/bulk-delete/", method="POST", json={ "ids": ids, "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteBulkAnnotationsResponse, construct_type( type_=DeleteBulkAnnotationsResponse, # 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.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( 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 create_bulk( self, *, bulk_created: typing.Optional[bool] = OMIT, completed_by: typing.Optional[int] = OMIT, draft_created_at: typing.Optional[dt.datetime] = OMIT, ground_truth: typing.Optional[bool] = OMIT, import_id: typing.Optional[int] = OMIT, last_action: typing.Optional[LastActionEnum] = OMIT, last_created_by: typing.Optional[int] = OMIT, lead_time: typing.Optional[float] = OMIT, parent_annotation: typing.Optional[int] = OMIT, parent_prediction: typing.Optional[int] = OMIT, project: typing.Optional[int] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, selected_items: typing.Optional[SelectedItemsRequest] = OMIT, task: typing.Optional[int] = OMIT, tasks: typing.Optional[typing.Sequence[int]] = OMIT, unique_id: typing.Optional[str] = OMIT, updated_by: typing.Optional[int] = OMIT, was_cancelled: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.List[CreateBulkAnnotationsResponseItem]]: """ Create multiple annotations at once Parameters ---------- bulk_created : typing.Optional[bool] Annotation was created in bulk mode completed_by : typing.Optional[int] draft_created_at : typing.Optional[dt.datetime] Draft creation time ground_truth : typing.Optional[bool] This annotation is a Ground Truth (ground_truth) import_id : typing.Optional[int] Original annotation ID that was at the import step or NULL if this annotation wasn't imported last_action : typing.Optional[LastActionEnum] Action which was performed in the last annotation history item * `prediction` - Created from prediction * `propagated_annotation` - Created from another annotation * `imported` - Imported * `submitted` - Submitted * `updated` - Updated * `skipped` - Skipped * `accepted` - Accepted * `rejected` - Rejected * `fixed_and_accepted` - Fixed and accepted * `deleted_review` - Deleted review last_created_by : typing.Optional[int] User who created the last annotation history item lead_time : typing.Optional[float] How much time it took to annotate the task parent_annotation : typing.Optional[int] Points to the parent annotation from which this annotation was created parent_prediction : typing.Optional[int] Points to the prediction from which this annotation was created project : typing.Optional[int] Project ID for this annotation result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] List of annotation results for the task selected_items : typing.Optional[SelectedItemsRequest] task : typing.Optional[int] Corresponding task for this annotation tasks : typing.Optional[typing.Sequence[int]] unique_id : typing.Optional[str] updated_by : typing.Optional[int] Last user who updated this annotation was_cancelled : typing.Optional[bool] User skipped the task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.List[CreateBulkAnnotationsResponseItem]] Bulk annotations created successfully """ _response = self._client_wrapper.httpx_client.request( "api/annotations/bulk/", method="POST", json={ "bulk_created": bulk_created, "completed_by": completed_by, "draft_created_at": draft_created_at, "ground_truth": ground_truth, "import_id": import_id, "last_action": last_action, "last_created_by": last_created_by, "lead_time": lead_time, "parent_annotation": parent_annotation, "parent_prediction": parent_prediction, "project": project, "result": result, "selected_items": convert_and_respect_annotation_metadata( object_=selected_items, annotation=typing.Optional[SelectedItemsRequest], direction="write" ), "task": task, "tasks": tasks, "unique_id": unique_id, "updated_by": updated_by, "was_cancelled": was_cancelled, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[CreateBulkAnnotationsResponseItem], construct_type( type_=typing.List[CreateBulkAnnotationsResponseItem], # 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[Annotation]: """ Retrieve a specific annotation for a task using the annotation result ID. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Annotation] Retrieved annotation """ _response = self._client_wrapper.httpx_client.request( f"api/annotations/{jsonable_encoder(id)}/", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Annotation, construct_type( type_=Annotation, # 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 an annotation. This action can't be undone! Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"api/annotations/{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, *, completed_by: typing.Optional[int] = OMIT, ground_truth: typing.Optional[bool] = OMIT, lead_time: typing.Optional[float] = OMIT, project: typing.Optional[int] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, task: typing.Optional[int] = OMIT, updated_by: typing.Optional[int] = OMIT, was_cancelled: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Annotation]: """ Update existing attributes on an annotation. Parameters ---------- id : int completed_by : typing.Optional[int] User ID of the person who created this annotation ground_truth : typing.Optional[bool] This annotation is a Ground Truth lead_time : typing.Optional[float] How much time it took to annotate the task (in seconds) project : typing.Optional[int] Project ID for this annotation result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Labeling result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/task_format) task : typing.Optional[int] Corresponding task for this annotation updated_by : typing.Optional[int] Last user who updated this annotation was_cancelled : typing.Optional[bool] User skipped the task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Annotation] Updated annotation """ _response = self._client_wrapper.httpx_client.request( f"api/annotations/{jsonable_encoder(id)}/", method="PATCH", json={ "completed_by": completed_by, "ground_truth": ground_truth, "lead_time": lead_time, "project": project, "result": result, "task": task, "updated_by": updated_by, "was_cancelled": was_cancelled, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Annotation, construct_type( type_=Annotation, # 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 list( self, id: int, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.List[Annotation]]: """ List all annotations for a task. Parameters ---------- id : int Task ID ordering : typing.Optional[str] Which field to use when ordering the results. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.List[Annotation]] Annotation """ _response = self._client_wrapper.httpx_client.request( f"api/tasks/{jsonable_encoder(id)}/annotations/", method="GET", params={ "ordering": ordering, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[Annotation], construct_type( type_=typing.List[Annotation], # 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, id: int, *, completed_by: typing.Optional[int] = OMIT, ground_truth: typing.Optional[bool] = OMIT, lead_time: typing.Optional[float] = OMIT, project: typing.Optional[int] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, task: typing.Optional[int] = OMIT, updated_by: typing.Optional[int] = OMIT, was_cancelled: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Annotation]: """ Add annotations to a task like an annotator does. The content of the result field depends on your labeling configuration. For example, send the following data as part of your POST request to send an empty annotation with the ID of the user who completed the task: ```json { "result": {}, "was_cancelled": true, "ground_truth": true, "lead_time": 0, "task": 0 "completed_by": 123 } ``` Parameters ---------- id : int Task ID completed_by : typing.Optional[int] User ID of the person who created this annotation ground_truth : typing.Optional[bool] This annotation is a Ground Truth lead_time : typing.Optional[float] How much time it took to annotate the task (in seconds) project : typing.Optional[int] Project ID for this annotation result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Labeling result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/task_format) task : typing.Optional[int] Corresponding task for this annotation updated_by : typing.Optional[int] Last user who updated this annotation was_cancelled : typing.Optional[bool] User skipped the task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Annotation] Created annotation """ _response = self._client_wrapper.httpx_client.request( f"api/tasks/{jsonable_encoder(id)}/annotations/", method="POST", json={ "completed_by": completed_by, "ground_truth": ground_truth, "lead_time": lead_time, "project": project, "result": result, "task": task, "updated_by": updated_by, "was_cancelled": was_cancelled, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Annotation, construct_type( type_=Annotation, # 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 AsyncRawAnnotationsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def delete_bulk( self, *, ids: typing.Sequence[int], project: int, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteBulkAnnotationsResponse]: """ Delete multiple annotations by their IDs. The deletion is processed synchronously. Returns the count of deleted annotations in the response. Parameters ---------- ids : typing.Sequence[int] List of annotation IDs to delete project : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteBulkAnnotationsResponse] Annotations deleted successfully """ _response = await self._client_wrapper.httpx_client.request( "api/annotations/bulk-delete/", method="POST", json={ "ids": ids, "project": project, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteBulkAnnotationsResponse, construct_type( type_=DeleteBulkAnnotationsResponse, # 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.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( 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 create_bulk( self, *, bulk_created: typing.Optional[bool] = OMIT, completed_by: typing.Optional[int] = OMIT, draft_created_at: typing.Optional[dt.datetime] = OMIT, ground_truth: typing.Optional[bool] = OMIT, import_id: typing.Optional[int] = OMIT, last_action: typing.Optional[LastActionEnum] = OMIT, last_created_by: typing.Optional[int] = OMIT, lead_time: typing.Optional[float] = OMIT, parent_annotation: typing.Optional[int] = OMIT, parent_prediction: typing.Optional[int] = OMIT, project: typing.Optional[int] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, selected_items: typing.Optional[SelectedItemsRequest] = OMIT, task: typing.Optional[int] = OMIT, tasks: typing.Optional[typing.Sequence[int]] = OMIT, unique_id: typing.Optional[str] = OMIT, updated_by: typing.Optional[int] = OMIT, was_cancelled: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.List[CreateBulkAnnotationsResponseItem]]: """ Create multiple annotations at once Parameters ---------- bulk_created : typing.Optional[bool] Annotation was created in bulk mode completed_by : typing.Optional[int] draft_created_at : typing.Optional[dt.datetime] Draft creation time ground_truth : typing.Optional[bool] This annotation is a Ground Truth (ground_truth) import_id : typing.Optional[int] Original annotation ID that was at the import step or NULL if this annotation wasn't imported last_action : typing.Optional[LastActionEnum] Action which was performed in the last annotation history item * `prediction` - Created from prediction * `propagated_annotation` - Created from another annotation * `imported` - Imported * `submitted` - Submitted * `updated` - Updated * `skipped` - Skipped * `accepted` - Accepted * `rejected` - Rejected * `fixed_and_accepted` - Fixed and accepted * `deleted_review` - Deleted review last_created_by : typing.Optional[int] User who created the last annotation history item lead_time : typing.Optional[float] How much time it took to annotate the task parent_annotation : typing.Optional[int] Points to the parent annotation from which this annotation was created parent_prediction : typing.Optional[int] Points to the prediction from which this annotation was created project : typing.Optional[int] Project ID for this annotation result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] List of annotation results for the task selected_items : typing.Optional[SelectedItemsRequest] task : typing.Optional[int] Corresponding task for this annotation tasks : typing.Optional[typing.Sequence[int]] unique_id : typing.Optional[str] updated_by : typing.Optional[int] Last user who updated this annotation was_cancelled : typing.Optional[bool] User skipped the task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.List[CreateBulkAnnotationsResponseItem]] Bulk annotations created successfully """ _response = await self._client_wrapper.httpx_client.request( "api/annotations/bulk/", method="POST", json={ "bulk_created": bulk_created, "completed_by": completed_by, "draft_created_at": draft_created_at, "ground_truth": ground_truth, "import_id": import_id, "last_action": last_action, "last_created_by": last_created_by, "lead_time": lead_time, "parent_annotation": parent_annotation, "parent_prediction": parent_prediction, "project": project, "result": result, "selected_items": convert_and_respect_annotation_metadata( object_=selected_items, annotation=typing.Optional[SelectedItemsRequest], direction="write" ), "task": task, "tasks": tasks, "unique_id": unique_id, "updated_by": updated_by, "was_cancelled": was_cancelled, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[CreateBulkAnnotationsResponseItem], construct_type( type_=typing.List[CreateBulkAnnotationsResponseItem], # 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[Annotation]: """ Retrieve a specific annotation for a task using the annotation result ID. Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Annotation] Retrieved annotation """ _response = await self._client_wrapper.httpx_client.request( f"api/annotations/{jsonable_encoder(id)}/", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Annotation, construct_type( type_=Annotation, # 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 an annotation. This action can't be undone! Parameters ---------- id : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"api/annotations/{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, *, completed_by: typing.Optional[int] = OMIT, ground_truth: typing.Optional[bool] = OMIT, lead_time: typing.Optional[float] = OMIT, project: typing.Optional[int] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, task: typing.Optional[int] = OMIT, updated_by: typing.Optional[int] = OMIT, was_cancelled: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Annotation]: """ Update existing attributes on an annotation. Parameters ---------- id : int completed_by : typing.Optional[int] User ID of the person who created this annotation ground_truth : typing.Optional[bool] This annotation is a Ground Truth lead_time : typing.Optional[float] How much time it took to annotate the task (in seconds) project : typing.Optional[int] Project ID for this annotation result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Labeling result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/task_format) task : typing.Optional[int] Corresponding task for this annotation updated_by : typing.Optional[int] Last user who updated this annotation was_cancelled : typing.Optional[bool] User skipped the task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Annotation] Updated annotation """ _response = await self._client_wrapper.httpx_client.request( f"api/annotations/{jsonable_encoder(id)}/", method="PATCH", json={ "completed_by": completed_by, "ground_truth": ground_truth, "lead_time": lead_time, "project": project, "result": result, "task": task, "updated_by": updated_by, "was_cancelled": was_cancelled, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Annotation, construct_type( type_=Annotation, # 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 list( self, id: int, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.List[Annotation]]: """ List all annotations for a task. Parameters ---------- id : int Task ID ordering : typing.Optional[str] Which field to use when ordering the results. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.List[Annotation]] Annotation """ _response = await self._client_wrapper.httpx_client.request( f"api/tasks/{jsonable_encoder(id)}/annotations/", method="GET", params={ "ordering": ordering, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[Annotation], construct_type( type_=typing.List[Annotation], # 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, id: int, *, completed_by: typing.Optional[int] = OMIT, ground_truth: typing.Optional[bool] = OMIT, lead_time: typing.Optional[float] = OMIT, project: typing.Optional[int] = OMIT, result: typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] = OMIT, task: typing.Optional[int] = OMIT, updated_by: typing.Optional[int] = OMIT, was_cancelled: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Annotation]: """ Add annotations to a task like an annotator does. The content of the result field depends on your labeling configuration. For example, send the following data as part of your POST request to send an empty annotation with the ID of the user who completed the task: ```json { "result": {}, "was_cancelled": true, "ground_truth": true, "lead_time": 0, "task": 0 "completed_by": 123 } ``` Parameters ---------- id : int Task ID completed_by : typing.Optional[int] User ID of the person who created this annotation ground_truth : typing.Optional[bool] This annotation is a Ground Truth lead_time : typing.Optional[float] How much time it took to annotate the task (in seconds) project : typing.Optional[int] Project ID for this annotation result : typing.Optional[typing.Sequence[typing.Dict[str, typing.Any]]] Labeling result in JSON format. Read more about the format in [the Label Studio documentation.](https://labelstud.io/guide/task_format) task : typing.Optional[int] Corresponding task for this annotation updated_by : typing.Optional[int] Last user who updated this annotation was_cancelled : typing.Optional[bool] User skipped the task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Annotation] Created annotation """ _response = await self._client_wrapper.httpx_client.request( f"api/tasks/{jsonable_encoder(id)}/annotations/", method="POST", json={ "completed_by": completed_by, "ground_truth": ground_truth, "lead_time": lead_time, "project": project, "result": result, "task": task, "updated_by": updated_by, "was_cancelled": was_cancelled, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Annotation, construct_type( type_=Annotation, # 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)