# 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 ...types.task_assignment import TaskAssignment
from .types.assign_assignments_request_type import AssignAssignmentsRequestType
from .types.bulk_assign_assignments_request_filters import BulkAssignAssignmentsRequestFilters
from .types.bulk_assign_assignments_request_selected_items import BulkAssignAssignmentsRequestSelectedItems
from .types.bulk_assign_assignments_request_type import BulkAssignAssignmentsRequestType
from .types.bulk_assign_assignments_response import BulkAssignAssignmentsResponse
from .types.delete_assignments_request_type import DeleteAssignmentsRequestType
from .types.update_assignments_request_type import UpdateAssignmentsRequestType
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class RawAssignmentsClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def bulk_assign(
self,
id: int,
*,
selected_items: BulkAssignAssignmentsRequestSelectedItems,
type: BulkAssignAssignmentsRequestType,
users: typing.Sequence[int],
filters: typing.Optional[BulkAssignAssignmentsRequestFilters] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[BulkAssignAssignmentsResponse]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Assign multiple users to a collection of tasks within a specific project.
Parameters
----------
id : int
selected_items : BulkAssignAssignmentsRequestSelectedItems
Task selection by IDs. If filters are applied, the selection will be applied to the filtered tasks.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]}`
type : BulkAssignAssignmentsRequestType
Assignment type. Use AN for annotate or RE for review.
users : typing.Sequence[int]
List of user IDs to assign
filters : typing.Optional[BulkAssignAssignmentsRequestFilters]
Filters to apply on tasks. You can use [the helper class `Filters` from this page](https://labelstud.io/sdk/data_manager.html) to create Data Manager Filters.
Example: `{"conjunction": "or", "items": [{"filter": "filter:tasks:completed_at", "operator": "greater", "type": "Datetime", "value": "2021-01-01T00:00:00.000Z"}]}`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[BulkAssignAssignmentsResponse]
Success
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/assignees",
method="POST",
json={
"filters": convert_and_respect_annotation_metadata(
object_=filters, annotation=BulkAssignAssignmentsRequestFilters, direction="write"
),
"selectedItems": convert_and_respect_annotation_metadata(
object_=selected_items, annotation=BulkAssignAssignmentsRequestSelectedItems, direction="write"
),
"type": type,
"users": users,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
BulkAssignAssignmentsResponse,
construct_type(
type_=BulkAssignAssignmentsResponse, # 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(),
),
),
)
_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, task_pk: int, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[typing.List[TaskAssignment]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Retrieve a list of tasks and assignees for those tasks for a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[typing.List[TaskAssignment]]
List of assignments for the task
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[TaskAssignment],
construct_type(
type_=typing.List[TaskAssignment], # 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 assign(
self,
id: int,
task_pk: int,
*,
type: AssignAssignmentsRequestType,
users: typing.Sequence[int],
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[TaskAssignment]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Assign a user to a task in a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
type : AssignAssignmentsRequestType
Assignment type. Use AN for annotate or RE for review.
users : typing.Sequence[int]
List of user IDs to assign
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[TaskAssignment]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="POST",
json={
"type": type,
"users": users,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
TaskAssignment,
construct_type(
type_=TaskAssignment, # 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,
task_pk: int,
*,
type: typing.Optional[DeleteAssignmentsRequestType] = None,
users: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[None]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Remove assignees for a task within a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
type : typing.Optional[DeleteAssignmentsRequestType]
Assignment type to delete (optional). If omitted, deletes all assignments for the task.
users : typing.Optional[str]
Comma separated list of user IDs to delete, as a string. If omitted, deletes all assignees for the given type.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="DELETE",
params={
"type": type,
"users": users,
},
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,
task_pk: int,
*,
type: UpdateAssignmentsRequestType,
users: typing.Sequence[int],
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[TaskAssignment]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update the assignee for a task in a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
type : UpdateAssignmentsRequestType
Assignment type. Use AN for annotate or RE for review.
users : typing.Sequence[int]
List of user IDs to assign
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[TaskAssignment]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="PATCH",
json={
"type": type,
"users": users,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
TaskAssignment,
construct_type(
type_=TaskAssignment, # 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 AsyncRawAssignmentsClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def bulk_assign(
self,
id: int,
*,
selected_items: BulkAssignAssignmentsRequestSelectedItems,
type: BulkAssignAssignmentsRequestType,
users: typing.Sequence[int],
filters: typing.Optional[BulkAssignAssignmentsRequestFilters] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[BulkAssignAssignmentsResponse]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Assign multiple users to a collection of tasks within a specific project.
Parameters
----------
id : int
selected_items : BulkAssignAssignmentsRequestSelectedItems
Task selection by IDs. If filters are applied, the selection will be applied to the filtered tasks.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]}`
type : BulkAssignAssignmentsRequestType
Assignment type. Use AN for annotate or RE for review.
users : typing.Sequence[int]
List of user IDs to assign
filters : typing.Optional[BulkAssignAssignmentsRequestFilters]
Filters to apply on tasks. You can use [the helper class `Filters` from this page](https://labelstud.io/sdk/data_manager.html) to create Data Manager Filters.
Example: `{"conjunction": "or", "items": [{"filter": "filter:tasks:completed_at", "operator": "greater", "type": "Datetime", "value": "2021-01-01T00:00:00.000Z"}]}`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[BulkAssignAssignmentsResponse]
Success
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/assignees",
method="POST",
json={
"filters": convert_and_respect_annotation_metadata(
object_=filters, annotation=BulkAssignAssignmentsRequestFilters, direction="write"
),
"selectedItems": convert_and_respect_annotation_metadata(
object_=selected_items, annotation=BulkAssignAssignmentsRequestSelectedItems, direction="write"
),
"type": type,
"users": users,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
BulkAssignAssignmentsResponse,
construct_type(
type_=BulkAssignAssignmentsResponse, # 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(),
),
),
)
_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, task_pk: int, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[typing.List[TaskAssignment]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Retrieve a list of tasks and assignees for those tasks for a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[typing.List[TaskAssignment]]
List of assignments for the task
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[TaskAssignment],
construct_type(
type_=typing.List[TaskAssignment], # 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 assign(
self,
id: int,
task_pk: int,
*,
type: AssignAssignmentsRequestType,
users: typing.Sequence[int],
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[TaskAssignment]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Assign a user to a task in a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
type : AssignAssignmentsRequestType
Assignment type. Use AN for annotate or RE for review.
users : typing.Sequence[int]
List of user IDs to assign
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[TaskAssignment]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="POST",
json={
"type": type,
"users": users,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
TaskAssignment,
construct_type(
type_=TaskAssignment, # 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,
task_pk: int,
*,
type: typing.Optional[DeleteAssignmentsRequestType] = None,
users: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[None]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Remove assignees for a task within a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
type : typing.Optional[DeleteAssignmentsRequestType]
Assignment type to delete (optional). If omitted, deletes all assignments for the task.
users : typing.Optional[str]
Comma separated list of user IDs to delete, as a string. If omitted, deletes all assignees for the given type.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="DELETE",
params={
"type": type,
"users": users,
},
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,
task_pk: int,
*,
type: UpdateAssignmentsRequestType,
users: typing.Sequence[int],
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[TaskAssignment]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update the assignee for a task in a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
task_pk : int
A unique integer value identifying this task.
type : UpdateAssignmentsRequestType
Assignment type. Use AN for annotate or RE for review.
users : typing.Sequence[int]
List of user IDs to assign
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[TaskAssignment]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(id)}/tasks/{jsonable_encoder(task_pk)}/assignees",
method="PATCH",
json={
"type": type,
"users": users,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
TaskAssignment,
construct_type(
type_=TaskAssignment, # 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)