# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.pagination import AsyncPager, SyncPager from ..core.request_options import RequestOptions from ..types.lse_task import LseTask from ..types.paginated_role_based_task_list import PaginatedRoleBasedTaskList from ..types.project_import import ProjectImport from ..types.role_based_task import RoleBasedTask from ..types.task_event import TaskEvent from .raw_client import AsyncRawTasksClient, RawTasksClient from .types.list_tasks_request_fields import ListTasksRequestFields # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class TasksClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawTasksClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawTasksClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawTasksClient """ return self._raw_client def create_many_status( self, id: int, import_pk: int, *, request_options: typing.Optional[RequestOptions] = None ) -> ProjectImport: """ Poll the status of an asynchronous project import operation. **Usage:** 1. When you POST to `/api/projects/{project_id}/import`, you'll receive a response like `{"import": }` 2. Use that `import_id` with this GET endpoint to check the import status 3. Poll this endpoint to see if the import has completed, is still processing, or has failed 4. **Import errors and failures will only be visible in this GET response**, not in the original POST request This endpoint returns detailed information about the import including task counts, status, and any error messages. Parameters ---------- id : int A unique integer value identifying this project import. import_pk : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectImport Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.create_many_status( id=1, import_pk=1, ) """ _response = self._raw_client.create_many_status(id, import_pk, request_options=request_options) return _response.data def delete_all_tasks(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete all tasks from a specific project. Parameters ---------- id : int A unique integer value identifying this project. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.delete_all_tasks( id=1, ) """ _response = self._raw_client.delete_all_tasks(id, request_options=request_options) return _response.data def list( self, *, fields: typing.Optional[ListTasksRequestFields] = None, include: typing.Optional[str] = None, only_annotated: typing.Optional[bool] = None, page: typing.Optional[int] = None, page_size: typing.Optional[int] = None, project: typing.Optional[int] = None, query: typing.Optional[str] = None, resolve_uri: typing.Optional[bool] = None, review: typing.Optional[bool] = None, selected_items: typing.Optional[str] = None, view: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> SyncPager[RoleBasedTask, PaginatedRoleBasedTaskList]: """ Retrieve a paginated list of tasks. The response format varies based on the user's role in the organization: - **Admin/Owner**: Full task details with all annotations, reviews, and metadata - **Reviewer**: Task details optimized for review workflow - **Annotator**: Task details filtered to show only user's own annotations and assignments Parameters ---------- fields : typing.Optional[ListTasksRequestFields] Set to "all" if you want to include annotations and predictions in the response. Defaults to task_only include : typing.Optional[str] Specify which fields to include in the response only_annotated : typing.Optional[bool] Filter to show only tasks that have annotations page : typing.Optional[int] A page number within the paginated result set. page_size : typing.Optional[int] Number of results to return per page. project : typing.Optional[int] Project ID query : typing.Optional[str] Additional query to filter tasks. It must be JSON encoded string of dict containing one of the following parameters: {"filters": ..., "selectedItems": ..., "ordering": ...}. Check Data Manager > Create View > see data field for more details about filters, selectedItems and ordering. filters: dict with "conjunction" string ("or" or "and") and list of filters in "items" array. Each filter is a dictionary with keys: "filter", "operator", "type", "value". Read more about available filters Example: {"conjunction": "or", "items": [{"filter": "filter:tasks:completed_at", "operator": "greater", "type": "Datetime", "value": "2021-01-01T00:00:00.000Z"}]} selectedItems: dictionary with keys: "all", "included", "excluded". If "all" is false, "included" must be used. If "all" is true, "excluded" must be used. Examples: {"all": false, "included": [1, 2, 3]} or {"all": true, "excluded": [4, 5]} ordering: list of fields to order by. Currently, ordering is supported by only one parameter. Example: ["completed_at"] resolve_uri : typing.Optional[bool] Resolve task data URIs using Cloud Storage review : typing.Optional[bool] Get tasks for review selected_items : typing.Optional[str] JSON string of selected task IDs for review workflow view : typing.Optional[int] View ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SyncPager[RoleBasedTask, PaginatedRoleBasedTaskList] Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) response = client.tasks.list() for item in response: yield item # alternatively, you can paginate page-by-page for page in response.iter_pages(): yield page """ return self._raw_client.list( fields=fields, include=include, only_annotated=only_annotated, page=page, page_size=page_size, project=project, query=query, resolve_uri=resolve_uri, review=review, selected_items=selected_items, view=view, request_options=request_options, ) def create( self, *, data: typing.Any, allow_skip: typing.Optional[bool] = OMIT, cancelled_annotations: typing.Optional[int] = OMIT, comment_authors: typing.Optional[typing.Sequence[int]] = OMIT, comment_count: typing.Optional[int] = OMIT, file_upload: typing.Optional[int] = OMIT, inner_id: typing.Optional[int] = OMIT, is_labeled: typing.Optional[bool] = OMIT, last_comment_updated_at: typing.Optional[dt.datetime] = OMIT, meta: typing.Optional[typing.Any] = OMIT, overlap: typing.Optional[int] = OMIT, project: typing.Optional[int] = OMIT, total_annotations: typing.Optional[int] = OMIT, total_predictions: typing.Optional[int] = OMIT, unresolved_comment_count: typing.Optional[int] = OMIT, updated_by: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> LseTask: """ Create a new task Parameters ---------- data : typing.Any allow_skip : typing.Optional[bool] Whether this task can be skipped. Set to False to make task unskippable. cancelled_annotations : typing.Optional[int] Number of total cancelled annotations for the current task comment_authors : typing.Optional[typing.Sequence[int]] Users who wrote comments comment_count : typing.Optional[int] Number of comments in the task including all annotations file_upload : typing.Optional[int] Uploaded file used as data source for this task inner_id : typing.Optional[int] Internal task ID in the project, starts with 1 is_labeled : typing.Optional[bool] True if the number of annotations for this task is greater than or equal to the number of maximum_completions for the project last_comment_updated_at : typing.Optional[dt.datetime] When the last comment was updated meta : typing.Optional[typing.Any] Meta is user imported (uploaded) data and can be useful as input for an ML Backend for embeddings, advanced vectors, and other info. It is passed to ML during training/predicting steps. overlap : typing.Optional[int] Number of distinct annotators that processed the current task project : typing.Optional[int] Project ID for this task total_annotations : typing.Optional[int] Number of total annotations for the current task except cancelled annotations total_predictions : typing.Optional[int] Number of total predictions for the current task unresolved_comment_count : typing.Optional[int] Number of unresolved comments in the task including all annotations updated_by : typing.Optional[int] Last annotator or reviewer who updated this task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LseTask Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.create( data={"image": "https://example.com/image.jpg", "text": "Hello, world!"}, project=1, ) """ _response = self._raw_client.create( data=data, allow_skip=allow_skip, cancelled_annotations=cancelled_annotations, comment_authors=comment_authors, comment_count=comment_count, file_upload=file_upload, inner_id=inner_id, is_labeled=is_labeled, last_comment_updated_at=last_comment_updated_at, meta=meta, overlap=overlap, project=project, total_annotations=total_annotations, total_predictions=total_predictions, unresolved_comment_count=unresolved_comment_count, updated_by=updated_by, request_options=request_options, ) return _response.data def get(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> RoleBasedTask: """ Get task data, metadata, annotations and other attributes for a specific labeling task by task ID. Parameters ---------- id : str Task ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- RoleBasedTask Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.get( id="id", ) """ _response = self._raw_client.get(id, request_options=request_options) return _response.data def delete(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete a task in Label Studio. This action cannot be undone! Parameters ---------- id : str Task ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.delete( id="id", ) """ _response = self._raw_client.delete(id, request_options=request_options) return _response.data def update( self, id: str, *, allow_skip: typing.Optional[bool] = OMIT, avg_lead_time: typing.Optional[float] = OMIT, cancelled_annotations: typing.Optional[int] = OMIT, comment_count: typing.Optional[int] = OMIT, completed_at: typing.Optional[dt.datetime] = OMIT, data: typing.Optional[typing.Any] = OMIT, draft_exists: typing.Optional[bool] = OMIT, ground_truth: typing.Optional[bool] = OMIT, inner_id: typing.Optional[int] = OMIT, is_labeled: typing.Optional[bool] = OMIT, last_comment_updated_at: typing.Optional[dt.datetime] = OMIT, meta: typing.Optional[typing.Any] = OMIT, overlap: typing.Optional[int] = OMIT, precomputed_agreement: typing.Optional[float] = OMIT, predictions_score: typing.Optional[float] = OMIT, project: typing.Optional[int] = OMIT, reviewed: typing.Optional[bool] = OMIT, reviews_accepted: typing.Optional[int] = OMIT, reviews_rejected: typing.Optional[int] = OMIT, total_annotations: typing.Optional[int] = OMIT, total_predictions: typing.Optional[int] = OMIT, unresolved_comment_count: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> RoleBasedTask: """ Update the attributes of an existing labeling task. Parameters ---------- id : str Task ID allow_skip : typing.Optional[bool] Whether this task can be skipped. Set to False to make task unskippable. avg_lead_time : typing.Optional[float] cancelled_annotations : typing.Optional[int] comment_count : typing.Optional[int] Number of comments in the task including all annotations completed_at : typing.Optional[dt.datetime] data : typing.Optional[typing.Any] User imported or uploaded data for a task. Data is formatted according to the project label config. You can find examples of data for your project on the Import page in the Label Studio Data Manager UI. draft_exists : typing.Optional[bool] ground_truth : typing.Optional[bool] inner_id : typing.Optional[int] is_labeled : typing.Optional[bool] True if the number of annotations for this task is greater than or equal to the number of maximum_completions for the project last_comment_updated_at : typing.Optional[dt.datetime] When the last comment was updated meta : typing.Optional[typing.Any] Meta is user imported (uploaded) data and can be useful as input for an ML Backend for embeddings, advanced vectors, and other info. It is passed to ML during training/predicting steps. overlap : typing.Optional[int] Number of distinct annotators that processed the current task precomputed_agreement : typing.Optional[float] Average agreement score for the task predictions_score : typing.Optional[float] project : typing.Optional[int] Project ID for this task reviewed : typing.Optional[bool] reviews_accepted : typing.Optional[int] reviews_rejected : typing.Optional[int] total_annotations : typing.Optional[int] total_predictions : typing.Optional[int] unresolved_comment_count : typing.Optional[int] Number of unresolved comments in the task including all annotations request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- RoleBasedTask Examples -------- from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.update( id="id", ) """ _response = self._raw_client.update( id, allow_skip=allow_skip, avg_lead_time=avg_lead_time, cancelled_annotations=cancelled_annotations, comment_count=comment_count, completed_at=completed_at, data=data, draft_exists=draft_exists, ground_truth=ground_truth, inner_id=inner_id, is_labeled=is_labeled, last_comment_updated_at=last_comment_updated_at, meta=meta, overlap=overlap, precomputed_agreement=precomputed_agreement, predictions_score=predictions_score, project=project, reviewed=reviewed, reviews_accepted=reviews_accepted, reviews_rejected=reviews_rejected, total_annotations=total_annotations, total_predictions=total_predictions, unresolved_comment_count=unresolved_comment_count, request_options=request_options, ) return _response.data def create_event( self, id: int, *, event_key: str, event_time: dt.datetime, annotation: typing.Optional[int] = OMIT, annotation_draft_id: typing.Optional[int] = OMIT, meta: typing.Optional[typing.Any] = OMIT, review: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> TaskEvent: """ Create a new task event to track user interactions and system events during annotation. This endpoint is designed to receive events from the frontend labeling interface to enable accurate lead time calculation and detailed annotation analytics. ## Event Types **Core Annotation Events:** - `annotation_loaded` - When annotation interface is loaded - `annotation_created` - When annotation is submitted - `annotation_updated` - When annotation is modified - `annotation_reviewed` - When annotation is reviewed **User Activity Events:** - `visibility_change` - When page visibility changes (tab switch, minimize) - `idle_detected` - When user goes idle - `idle_resumed` - When user returns from idle **Interaction Events:** - `region_finished_drawing` - When annotation region is completed - `region_deleted` - When annotation regions are removed - `hotkey_pressed` - When keyboard shortcuts are used **Media Events:** - `video_playback_start/end` - Video playback control - `audio_playback_start/end` - Audio playback control - `video_scrub` - Video timeline scrubbing ## Usage Events are automatically associated with the task specified in the URL path. The current user is automatically set as the actor. Project and organization are derived from the task context. ## Example Request ```json { "event_key": "annotation_loaded", "event_time": "2024-01-15T10:30:00Z", "annotation": 123, "meta": { "annotation_count": 5, "estimated_time": 300 } } ``` Parameters ---------- id : int Task ID to associate the event with event_key : str Event type identifier (e.g., "annotation_loaded", "region_finished_drawing") event_time : dt.datetime Timestamp when the event occurred (frontend time) annotation : typing.Optional[int] Annotation ID associated with this event annotation_draft_id : typing.Optional[int] Draft annotation ID associated with this event meta : typing.Optional[typing.Any] Additional event metadata (region data, hotkey info, etc.) review : typing.Optional[int] Review ID associated with this event request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TaskEvent Examples -------- import datetime from label_studio_sdk import LabelStudio client = LabelStudio( api_key="YOUR_API_KEY", ) client.tasks.create_event( id=1, event_key="event_key", event_time=datetime.datetime.fromisoformat( "2024-01-15 09:30:00+00:00", ), ) """ _response = self._raw_client.create_event( id, event_key=event_key, event_time=event_time, annotation=annotation, annotation_draft_id=annotation_draft_id, meta=meta, review=review, request_options=request_options, ) return _response.data class AsyncTasksClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawTasksClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawTasksClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawTasksClient """ return self._raw_client async def create_many_status( self, id: int, import_pk: int, *, request_options: typing.Optional[RequestOptions] = None ) -> ProjectImport: """ Poll the status of an asynchronous project import operation. **Usage:** 1. When you POST to `/api/projects/{project_id}/import`, you'll receive a response like `{"import": }` 2. Use that `import_id` with this GET endpoint to check the import status 3. Poll this endpoint to see if the import has completed, is still processing, or has failed 4. **Import errors and failures will only be visible in this GET response**, not in the original POST request This endpoint returns detailed information about the import including task counts, status, and any error messages. Parameters ---------- id : int A unique integer value identifying this project import. import_pk : int request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- ProjectImport Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.create_many_status( id=1, import_pk=1, ) asyncio.run(main()) """ _response = await self._raw_client.create_many_status(id, import_pk, request_options=request_options) return _response.data async def delete_all_tasks(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete all tasks from a specific project. Parameters ---------- id : int A unique integer value identifying this project. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.delete_all_tasks( id=1, ) asyncio.run(main()) """ _response = await self._raw_client.delete_all_tasks(id, request_options=request_options) return _response.data async def list( self, *, fields: typing.Optional[ListTasksRequestFields] = None, include: typing.Optional[str] = None, only_annotated: typing.Optional[bool] = None, page: typing.Optional[int] = None, page_size: typing.Optional[int] = None, project: typing.Optional[int] = None, query: typing.Optional[str] = None, resolve_uri: typing.Optional[bool] = None, review: typing.Optional[bool] = None, selected_items: typing.Optional[str] = None, view: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncPager[RoleBasedTask, PaginatedRoleBasedTaskList]: """ Retrieve a paginated list of tasks. The response format varies based on the user's role in the organization: - **Admin/Owner**: Full task details with all annotations, reviews, and metadata - **Reviewer**: Task details optimized for review workflow - **Annotator**: Task details filtered to show only user's own annotations and assignments Parameters ---------- fields : typing.Optional[ListTasksRequestFields] Set to "all" if you want to include annotations and predictions in the response. Defaults to task_only include : typing.Optional[str] Specify which fields to include in the response only_annotated : typing.Optional[bool] Filter to show only tasks that have annotations page : typing.Optional[int] A page number within the paginated result set. page_size : typing.Optional[int] Number of results to return per page. project : typing.Optional[int] Project ID query : typing.Optional[str] Additional query to filter tasks. It must be JSON encoded string of dict containing one of the following parameters: {"filters": ..., "selectedItems": ..., "ordering": ...}. Check Data Manager > Create View > see data field for more details about filters, selectedItems and ordering. filters: dict with "conjunction" string ("or" or "and") and list of filters in "items" array. Each filter is a dictionary with keys: "filter", "operator", "type", "value". Read more about available filters Example: {"conjunction": "or", "items": [{"filter": "filter:tasks:completed_at", "operator": "greater", "type": "Datetime", "value": "2021-01-01T00:00:00.000Z"}]} selectedItems: dictionary with keys: "all", "included", "excluded". If "all" is false, "included" must be used. If "all" is true, "excluded" must be used. Examples: {"all": false, "included": [1, 2, 3]} or {"all": true, "excluded": [4, 5]} ordering: list of fields to order by. Currently, ordering is supported by only one parameter. Example: ["completed_at"] resolve_uri : typing.Optional[bool] Resolve task data URIs using Cloud Storage review : typing.Optional[bool] Get tasks for review selected_items : typing.Optional[str] JSON string of selected task IDs for review workflow view : typing.Optional[int] View ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncPager[RoleBasedTask, PaginatedRoleBasedTaskList] Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: response = await client.tasks.list() async for item in response: yield item # alternatively, you can paginate page-by-page async for page in response.iter_pages(): yield page asyncio.run(main()) """ return await self._raw_client.list( fields=fields, include=include, only_annotated=only_annotated, page=page, page_size=page_size, project=project, query=query, resolve_uri=resolve_uri, review=review, selected_items=selected_items, view=view, request_options=request_options, ) async def create( self, *, data: typing.Any, allow_skip: typing.Optional[bool] = OMIT, cancelled_annotations: typing.Optional[int] = OMIT, comment_authors: typing.Optional[typing.Sequence[int]] = OMIT, comment_count: typing.Optional[int] = OMIT, file_upload: typing.Optional[int] = OMIT, inner_id: typing.Optional[int] = OMIT, is_labeled: typing.Optional[bool] = OMIT, last_comment_updated_at: typing.Optional[dt.datetime] = OMIT, meta: typing.Optional[typing.Any] = OMIT, overlap: typing.Optional[int] = OMIT, project: typing.Optional[int] = OMIT, total_annotations: typing.Optional[int] = OMIT, total_predictions: typing.Optional[int] = OMIT, unresolved_comment_count: typing.Optional[int] = OMIT, updated_by: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> LseTask: """ Create a new task Parameters ---------- data : typing.Any allow_skip : typing.Optional[bool] Whether this task can be skipped. Set to False to make task unskippable. cancelled_annotations : typing.Optional[int] Number of total cancelled annotations for the current task comment_authors : typing.Optional[typing.Sequence[int]] Users who wrote comments comment_count : typing.Optional[int] Number of comments in the task including all annotations file_upload : typing.Optional[int] Uploaded file used as data source for this task inner_id : typing.Optional[int] Internal task ID in the project, starts with 1 is_labeled : typing.Optional[bool] True if the number of annotations for this task is greater than or equal to the number of maximum_completions for the project last_comment_updated_at : typing.Optional[dt.datetime] When the last comment was updated meta : typing.Optional[typing.Any] Meta is user imported (uploaded) data and can be useful as input for an ML Backend for embeddings, advanced vectors, and other info. It is passed to ML during training/predicting steps. overlap : typing.Optional[int] Number of distinct annotators that processed the current task project : typing.Optional[int] Project ID for this task total_annotations : typing.Optional[int] Number of total annotations for the current task except cancelled annotations total_predictions : typing.Optional[int] Number of total predictions for the current task unresolved_comment_count : typing.Optional[int] Number of unresolved comments in the task including all annotations updated_by : typing.Optional[int] Last annotator or reviewer who updated this task request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LseTask Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.create( data={ "image": "https://example.com/image.jpg", "text": "Hello, world!", }, project=1, ) asyncio.run(main()) """ _response = await self._raw_client.create( data=data, allow_skip=allow_skip, cancelled_annotations=cancelled_annotations, comment_authors=comment_authors, comment_count=comment_count, file_upload=file_upload, inner_id=inner_id, is_labeled=is_labeled, last_comment_updated_at=last_comment_updated_at, meta=meta, overlap=overlap, project=project, total_annotations=total_annotations, total_predictions=total_predictions, unresolved_comment_count=unresolved_comment_count, updated_by=updated_by, request_options=request_options, ) return _response.data async def get(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> RoleBasedTask: """ Get task data, metadata, annotations and other attributes for a specific labeling task by task ID. Parameters ---------- id : str Task ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- RoleBasedTask Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.get( id="id", ) asyncio.run(main()) """ _response = await self._raw_client.get(id, request_options=request_options) return _response.data async def delete(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Delete a task in Label Studio. This action cannot be undone! Parameters ---------- id : str Task ID request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.delete( id="id", ) asyncio.run(main()) """ _response = await self._raw_client.delete(id, request_options=request_options) return _response.data async def update( self, id: str, *, allow_skip: typing.Optional[bool] = OMIT, avg_lead_time: typing.Optional[float] = OMIT, cancelled_annotations: typing.Optional[int] = OMIT, comment_count: typing.Optional[int] = OMIT, completed_at: typing.Optional[dt.datetime] = OMIT, data: typing.Optional[typing.Any] = OMIT, draft_exists: typing.Optional[bool] = OMIT, ground_truth: typing.Optional[bool] = OMIT, inner_id: typing.Optional[int] = OMIT, is_labeled: typing.Optional[bool] = OMIT, last_comment_updated_at: typing.Optional[dt.datetime] = OMIT, meta: typing.Optional[typing.Any] = OMIT, overlap: typing.Optional[int] = OMIT, precomputed_agreement: typing.Optional[float] = OMIT, predictions_score: typing.Optional[float] = OMIT, project: typing.Optional[int] = OMIT, reviewed: typing.Optional[bool] = OMIT, reviews_accepted: typing.Optional[int] = OMIT, reviews_rejected: typing.Optional[int] = OMIT, total_annotations: typing.Optional[int] = OMIT, total_predictions: typing.Optional[int] = OMIT, unresolved_comment_count: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> RoleBasedTask: """ Update the attributes of an existing labeling task. Parameters ---------- id : str Task ID allow_skip : typing.Optional[bool] Whether this task can be skipped. Set to False to make task unskippable. avg_lead_time : typing.Optional[float] cancelled_annotations : typing.Optional[int] comment_count : typing.Optional[int] Number of comments in the task including all annotations completed_at : typing.Optional[dt.datetime] data : typing.Optional[typing.Any] User imported or uploaded data for a task. Data is formatted according to the project label config. You can find examples of data for your project on the Import page in the Label Studio Data Manager UI. draft_exists : typing.Optional[bool] ground_truth : typing.Optional[bool] inner_id : typing.Optional[int] is_labeled : typing.Optional[bool] True if the number of annotations for this task is greater than or equal to the number of maximum_completions for the project last_comment_updated_at : typing.Optional[dt.datetime] When the last comment was updated meta : typing.Optional[typing.Any] Meta is user imported (uploaded) data and can be useful as input for an ML Backend for embeddings, advanced vectors, and other info. It is passed to ML during training/predicting steps. overlap : typing.Optional[int] Number of distinct annotators that processed the current task precomputed_agreement : typing.Optional[float] Average agreement score for the task predictions_score : typing.Optional[float] project : typing.Optional[int] Project ID for this task reviewed : typing.Optional[bool] reviews_accepted : typing.Optional[int] reviews_rejected : typing.Optional[int] total_annotations : typing.Optional[int] total_predictions : typing.Optional[int] unresolved_comment_count : typing.Optional[int] Number of unresolved comments in the task including all annotations request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- RoleBasedTask Examples -------- import asyncio from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.update( id="id", ) asyncio.run(main()) """ _response = await self._raw_client.update( id, allow_skip=allow_skip, avg_lead_time=avg_lead_time, cancelled_annotations=cancelled_annotations, comment_count=comment_count, completed_at=completed_at, data=data, draft_exists=draft_exists, ground_truth=ground_truth, inner_id=inner_id, is_labeled=is_labeled, last_comment_updated_at=last_comment_updated_at, meta=meta, overlap=overlap, precomputed_agreement=precomputed_agreement, predictions_score=predictions_score, project=project, reviewed=reviewed, reviews_accepted=reviews_accepted, reviews_rejected=reviews_rejected, total_annotations=total_annotations, total_predictions=total_predictions, unresolved_comment_count=unresolved_comment_count, request_options=request_options, ) return _response.data async def create_event( self, id: int, *, event_key: str, event_time: dt.datetime, annotation: typing.Optional[int] = OMIT, annotation_draft_id: typing.Optional[int] = OMIT, meta: typing.Optional[typing.Any] = OMIT, review: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> TaskEvent: """ Create a new task event to track user interactions and system events during annotation. This endpoint is designed to receive events from the frontend labeling interface to enable accurate lead time calculation and detailed annotation analytics. ## Event Types **Core Annotation Events:** - `annotation_loaded` - When annotation interface is loaded - `annotation_created` - When annotation is submitted - `annotation_updated` - When annotation is modified - `annotation_reviewed` - When annotation is reviewed **User Activity Events:** - `visibility_change` - When page visibility changes (tab switch, minimize) - `idle_detected` - When user goes idle - `idle_resumed` - When user returns from idle **Interaction Events:** - `region_finished_drawing` - When annotation region is completed - `region_deleted` - When annotation regions are removed - `hotkey_pressed` - When keyboard shortcuts are used **Media Events:** - `video_playback_start/end` - Video playback control - `audio_playback_start/end` - Audio playback control - `video_scrub` - Video timeline scrubbing ## Usage Events are automatically associated with the task specified in the URL path. The current user is automatically set as the actor. Project and organization are derived from the task context. ## Example Request ```json { "event_key": "annotation_loaded", "event_time": "2024-01-15T10:30:00Z", "annotation": 123, "meta": { "annotation_count": 5, "estimated_time": 300 } } ``` Parameters ---------- id : int Task ID to associate the event with event_key : str Event type identifier (e.g., "annotation_loaded", "region_finished_drawing") event_time : dt.datetime Timestamp when the event occurred (frontend time) annotation : typing.Optional[int] Annotation ID associated with this event annotation_draft_id : typing.Optional[int] Draft annotation ID associated with this event meta : typing.Optional[typing.Any] Additional event metadata (region data, hotkey info, etc.) review : typing.Optional[int] Review ID associated with this event request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TaskEvent Examples -------- import asyncio import datetime from label_studio_sdk import AsyncLabelStudio client = AsyncLabelStudio( api_key="YOUR_API_KEY", ) async def main() -> None: await client.tasks.create_event( id=1, event_key="event_key", event_time=datetime.datetime.fromisoformat( "2024-01-15 09:30:00+00:00", ), ) asyncio.run(main()) """ _response = await self._raw_client.create_event( id, event_key=event_key, event_time=event_time, annotation=annotation, annotation_draft_id=annotation_draft_id, meta=meta, review=review, request_options=request_options, ) return _response.data