# 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.batch_failed_predictions import BatchFailedPredictions
from ..types.batch_predictions import BatchPredictions
from ..types.model_interface import ModelInterface
from ..types.model_interface_serializer_get import ModelInterfaceSerializerGet
from ..types.paginated_all_roles_project_list_list import PaginatedAllRolesProjectListList
from ..types.paginated_project_subset_tasks_response_list import PaginatedProjectSubsetTasksResponseList
from ..types.project_subset_item import ProjectSubsetItem
from ..types.skill_name_enum import SkillNameEnum
from ..types.user_simple_request import UserSimpleRequest
from .types.compatible_projects_prompts_request_project_type import CompatibleProjectsPromptsRequestProjectType
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class RawPromptsClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def batch_failed_predictions(
self,
*,
failed_predictions: typing.Sequence[typing.Any],
modelrun_id: int,
num_failed_predictions: typing.Optional[int] = None,
job_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[BatchFailedPredictions]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new batch of failed predictions.
Parameters
----------
failed_predictions : typing.Sequence[typing.Any]
modelrun_id : int
num_failed_predictions : typing.Optional[int]
Number of failed predictions being sent (for telemetry only, has no effect)
job_id : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[BatchFailedPredictions]
"""
_response = self._client_wrapper.httpx_client.request(
"api/model-run/batch-failed-predictions",
method="POST",
params={
"num_failed_predictions": num_failed_predictions,
},
json={
"failed_predictions": failed_predictions,
"job_id": job_id,
"modelrun_id": modelrun_id,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
BatchFailedPredictions,
construct_type(
type_=BatchFailedPredictions, # 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 batch_predictions(
self,
*,
modelrun_id: int,
results: typing.Sequence[typing.Any],
num_predictions: typing.Optional[int] = None,
job_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[BatchPredictions]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new batch prediction.
Parameters
----------
modelrun_id : int
results : typing.Sequence[typing.Any]
num_predictions : typing.Optional[int]
Number of predictions being sent (for telemetry only, has no effect)
job_id : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[BatchPredictions]
"""
_response = self._client_wrapper.httpx_client.request(
"api/model-run/batch-predictions",
method="POST",
params={
"num_predictions": num_predictions,
},
json={
"job_id": job_id,
"modelrun_id": modelrun_id,
"results": results,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
BatchPredictions,
construct_type(
type_=BatchPredictions, # 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 subset_tasks(
self,
project_pk: int,
*,
include_total: typing.Optional[bool] = None,
model_run: typing.Optional[int] = None,
ordering: typing.Optional[str] = None,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
parent_model: typing.Optional[int] = None,
project_subset: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[PaginatedProjectSubsetTasksResponseList]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Provides list of tasks, based on project subset. Includes predictions for tasks. For the 'HasGT' subset, accuracy metrics will also be provided.
Parameters
----------
project_pk : int
include_total : typing.Optional[bool]
If true (default), includes task_count in response; if false, omits it.
model_run : typing.Optional[int]
A unique ID of a ModelRun
ordering : typing.Optional[str]
Which field to use when ordering the results.
page : typing.Optional[int]
A page number within the paginated result set.
page_size : typing.Optional[int]
Number of results to return per page.
parent_model : typing.Optional[int]
The ID of the parent model (ModelInterface) for this Inference Run
project_subset : typing.Optional[str]
The project subset to retrieve tasks for
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[PaginatedProjectSubsetTasksResponseList]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(project_pk)}/subset-tasks",
method="GET",
params={
"include_total": include_total,
"model_run": model_run,
"ordering": ordering,
"page": page,
"page_size": page_size,
"parent_model": parent_model,
"project_subset": project_subset,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
PaginatedProjectSubsetTasksResponseList,
construct_type(
type_=PaginatedProjectSubsetTasksResponseList, # 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 subsets(
self,
project_pk: int,
*,
ordering: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[typing.List[ProjectSubsetItem]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Provides list of available subsets for a project along with count of tasks in each subset
Parameters
----------
project_pk : int
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[typing.List[ProjectSubsetItem]]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(project_pk)}/subsets",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[ProjectSubsetItem],
construct_type(
type_=typing.List[ProjectSubsetItem], # 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, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[typing.List[ModelInterfaceSerializerGet]]:
"""
List all prompts.
Parameters
----------
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[typing.List[ModelInterfaceSerializerGet]]
"""
_response = self._client_wrapper.httpx_client.request(
"api/prompts/",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[ModelInterfaceSerializerGet],
construct_type(
type_=typing.List[ModelInterfaceSerializerGet], # 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,
*,
title: str,
associated_projects: typing.Optional[typing.Sequence[int]] = OMIT,
created_by: typing.Optional[UserSimpleRequest] = OMIT,
description: typing.Optional[str] = OMIT,
input_fields: typing.Optional[typing.Any] = OMIT,
organization: typing.Optional[int] = OMIT,
output_classes: typing.Optional[typing.Any] = OMIT,
skill_name: typing.Optional[SkillNameEnum] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[ModelInterface]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new prompt.
Parameters
----------
title : str
Model name
associated_projects : typing.Optional[typing.Sequence[int]]
created_by : typing.Optional[UserSimpleRequest]
User who created Dataset
description : typing.Optional[str]
Model description
input_fields : typing.Optional[typing.Any]
organization : typing.Optional[int]
output_classes : typing.Optional[typing.Any]
skill_name : typing.Optional[SkillNameEnum]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ModelInterface]
"""
_response = self._client_wrapper.httpx_client.request(
"api/prompts/",
method="POST",
json={
"associated_projects": associated_projects,
"created_by": convert_and_respect_annotation_metadata(
object_=created_by, annotation=UserSimpleRequest, direction="write"
),
"description": description,
"input_fields": input_fields,
"organization": organization,
"output_classes": output_classes,
"skill_name": skill_name,
"title": title,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelInterface,
construct_type(
type_=ModelInterface, # 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 compatible_projects(
self,
*,
ordering: typing.Optional[str] = None,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
project_type: typing.Optional[CompatibleProjectsPromptsRequestProjectType] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[PaginatedAllRolesProjectListList]:
"""
Retrieve a list of compatible project for prompt.
Parameters
----------
ordering : typing.Optional[str]
Which field to use when ordering the results.
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_type : typing.Optional[CompatibleProjectsPromptsRequestProjectType]
Skill to filter by
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[PaginatedAllRolesProjectListList]
"""
_response = self._client_wrapper.httpx_client.request(
"api/prompts/compatible-projects",
method="GET",
params={
"ordering": ordering,
"page": page,
"page_size": page_size,
"project_type": project_type,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
PaginatedAllRolesProjectListList,
construct_type(
type_=PaginatedAllRolesProjectListList, # 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: str, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[ModelInterfaceSerializerGet]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Retrieve a specific prompt.
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ModelInterfaceSerializerGet]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/prompts/{jsonable_encoder(id)}/",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelInterfaceSerializerGet,
construct_type(
type_=ModelInterfaceSerializerGet, # 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: str, *, 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)
Delete a prompt by ID
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/prompts/{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: str,
*,
associated_projects: typing.Optional[typing.Sequence[int]] = OMIT,
created_by: typing.Optional[UserSimpleRequest] = OMIT,
description: typing.Optional[str] = OMIT,
input_fields: typing.Optional[typing.Any] = OMIT,
organization: typing.Optional[int] = OMIT,
output_classes: typing.Optional[typing.Any] = OMIT,
skill_name: typing.Optional[SkillNameEnum] = OMIT,
title: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[ModelInterface]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update a specific prompt by ID.
Parameters
----------
id : str
associated_projects : typing.Optional[typing.Sequence[int]]
created_by : typing.Optional[UserSimpleRequest]
User who created Dataset
description : typing.Optional[str]
Model description
input_fields : typing.Optional[typing.Any]
organization : typing.Optional[int]
output_classes : typing.Optional[typing.Any]
skill_name : typing.Optional[SkillNameEnum]
title : typing.Optional[str]
Model name
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ModelInterface]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/prompts/{jsonable_encoder(id)}/",
method="PATCH",
json={
"associated_projects": associated_projects,
"created_by": convert_and_respect_annotation_metadata(
object_=created_by, annotation=UserSimpleRequest, direction="write"
),
"description": description,
"input_fields": input_fields,
"organization": organization,
"output_classes": output_classes,
"skill_name": skill_name,
"title": title,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelInterface,
construct_type(
type_=ModelInterface, # 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 AsyncRawPromptsClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def batch_failed_predictions(
self,
*,
failed_predictions: typing.Sequence[typing.Any],
modelrun_id: int,
num_failed_predictions: typing.Optional[int] = None,
job_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[BatchFailedPredictions]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new batch of failed predictions.
Parameters
----------
failed_predictions : typing.Sequence[typing.Any]
modelrun_id : int
num_failed_predictions : typing.Optional[int]
Number of failed predictions being sent (for telemetry only, has no effect)
job_id : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[BatchFailedPredictions]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/model-run/batch-failed-predictions",
method="POST",
params={
"num_failed_predictions": num_failed_predictions,
},
json={
"failed_predictions": failed_predictions,
"job_id": job_id,
"modelrun_id": modelrun_id,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
BatchFailedPredictions,
construct_type(
type_=BatchFailedPredictions, # 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 batch_predictions(
self,
*,
modelrun_id: int,
results: typing.Sequence[typing.Any],
num_predictions: typing.Optional[int] = None,
job_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[BatchPredictions]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new batch prediction.
Parameters
----------
modelrun_id : int
results : typing.Sequence[typing.Any]
num_predictions : typing.Optional[int]
Number of predictions being sent (for telemetry only, has no effect)
job_id : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[BatchPredictions]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/model-run/batch-predictions",
method="POST",
params={
"num_predictions": num_predictions,
},
json={
"job_id": job_id,
"modelrun_id": modelrun_id,
"results": results,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
BatchPredictions,
construct_type(
type_=BatchPredictions, # 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 subset_tasks(
self,
project_pk: int,
*,
include_total: typing.Optional[bool] = None,
model_run: typing.Optional[int] = None,
ordering: typing.Optional[str] = None,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
parent_model: typing.Optional[int] = None,
project_subset: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[PaginatedProjectSubsetTasksResponseList]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Provides list of tasks, based on project subset. Includes predictions for tasks. For the 'HasGT' subset, accuracy metrics will also be provided.
Parameters
----------
project_pk : int
include_total : typing.Optional[bool]
If true (default), includes task_count in response; if false, omits it.
model_run : typing.Optional[int]
A unique ID of a ModelRun
ordering : typing.Optional[str]
Which field to use when ordering the results.
page : typing.Optional[int]
A page number within the paginated result set.
page_size : typing.Optional[int]
Number of results to return per page.
parent_model : typing.Optional[int]
The ID of the parent model (ModelInterface) for this Inference Run
project_subset : typing.Optional[str]
The project subset to retrieve tasks for
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[PaginatedProjectSubsetTasksResponseList]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(project_pk)}/subset-tasks",
method="GET",
params={
"include_total": include_total,
"model_run": model_run,
"ordering": ordering,
"page": page,
"page_size": page_size,
"parent_model": parent_model,
"project_subset": project_subset,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
PaginatedProjectSubsetTasksResponseList,
construct_type(
type_=PaginatedProjectSubsetTasksResponseList, # 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 subsets(
self,
project_pk: int,
*,
ordering: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[typing.List[ProjectSubsetItem]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Provides list of available subsets for a project along with count of tasks in each subset
Parameters
----------
project_pk : int
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[typing.List[ProjectSubsetItem]]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/projects/{jsonable_encoder(project_pk)}/subsets",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[ProjectSubsetItem],
construct_type(
type_=typing.List[ProjectSubsetItem], # 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, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[typing.List[ModelInterfaceSerializerGet]]:
"""
List all prompts.
Parameters
----------
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[typing.List[ModelInterfaceSerializerGet]]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/prompts/",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[ModelInterfaceSerializerGet],
construct_type(
type_=typing.List[ModelInterfaceSerializerGet], # 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,
*,
title: str,
associated_projects: typing.Optional[typing.Sequence[int]] = OMIT,
created_by: typing.Optional[UserSimpleRequest] = OMIT,
description: typing.Optional[str] = OMIT,
input_fields: typing.Optional[typing.Any] = OMIT,
organization: typing.Optional[int] = OMIT,
output_classes: typing.Optional[typing.Any] = OMIT,
skill_name: typing.Optional[SkillNameEnum] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[ModelInterface]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new prompt.
Parameters
----------
title : str
Model name
associated_projects : typing.Optional[typing.Sequence[int]]
created_by : typing.Optional[UserSimpleRequest]
User who created Dataset
description : typing.Optional[str]
Model description
input_fields : typing.Optional[typing.Any]
organization : typing.Optional[int]
output_classes : typing.Optional[typing.Any]
skill_name : typing.Optional[SkillNameEnum]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ModelInterface]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/prompts/",
method="POST",
json={
"associated_projects": associated_projects,
"created_by": convert_and_respect_annotation_metadata(
object_=created_by, annotation=UserSimpleRequest, direction="write"
),
"description": description,
"input_fields": input_fields,
"organization": organization,
"output_classes": output_classes,
"skill_name": skill_name,
"title": title,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelInterface,
construct_type(
type_=ModelInterface, # 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 compatible_projects(
self,
*,
ordering: typing.Optional[str] = None,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
project_type: typing.Optional[CompatibleProjectsPromptsRequestProjectType] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[PaginatedAllRolesProjectListList]:
"""
Retrieve a list of compatible project for prompt.
Parameters
----------
ordering : typing.Optional[str]
Which field to use when ordering the results.
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_type : typing.Optional[CompatibleProjectsPromptsRequestProjectType]
Skill to filter by
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[PaginatedAllRolesProjectListList]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/prompts/compatible-projects",
method="GET",
params={
"ordering": ordering,
"page": page,
"page_size": page_size,
"project_type": project_type,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
PaginatedAllRolesProjectListList,
construct_type(
type_=PaginatedAllRolesProjectListList, # 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: str, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[ModelInterfaceSerializerGet]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Retrieve a specific prompt.
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ModelInterfaceSerializerGet]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/prompts/{jsonable_encoder(id)}/",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelInterfaceSerializerGet,
construct_type(
type_=ModelInterfaceSerializerGet, # 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: str, *, 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)
Delete a prompt by ID
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/prompts/{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: str,
*,
associated_projects: typing.Optional[typing.Sequence[int]] = OMIT,
created_by: typing.Optional[UserSimpleRequest] = OMIT,
description: typing.Optional[str] = OMIT,
input_fields: typing.Optional[typing.Any] = OMIT,
organization: typing.Optional[int] = OMIT,
output_classes: typing.Optional[typing.Any] = OMIT,
skill_name: typing.Optional[SkillNameEnum] = OMIT,
title: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[ModelInterface]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update a specific prompt by ID.
Parameters
----------
id : str
associated_projects : typing.Optional[typing.Sequence[int]]
created_by : typing.Optional[UserSimpleRequest]
User who created Dataset
description : typing.Optional[str]
Model description
input_fields : typing.Optional[typing.Any]
organization : typing.Optional[int]
output_classes : typing.Optional[typing.Any]
skill_name : typing.Optional[SkillNameEnum]
title : typing.Optional[str]
Model name
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ModelInterface]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/prompts/{jsonable_encoder(id)}/",
method="PATCH",
json={
"associated_projects": associated_projects,
"created_by": convert_and_respect_annotation_metadata(
object_=created_by, annotation=UserSimpleRequest, direction="write"
),
"description": description,
"input_fields": input_fields,
"organization": organization,
"output_classes": output_classes,
"skill_name": skill_name,
"title": title,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelInterface,
construct_type(
type_=ModelInterface, # 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)