# 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.request_options import RequestOptions
from ...types.converted_format_request import ConvertedFormatRequest
from ...types.export import Export
from ...types.lse_annotation_filter_options_request import LseAnnotationFilterOptionsRequest
from ...types.lse_export_create import LseExportCreate
from ...types.lse_task_filter_options_request import LseTaskFilterOptionsRequest
from ...types.serialization_options_request import SerializationOptionsRequest
from ...types.status7bf_enum import Status7BfEnum
from ...types.user_simple_request import UserSimpleRequest
from .raw_client import AsyncRawExportsClient, RawExportsClient
from .types.convert_exports_response import ConvertExportsResponse
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class ExportsClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._raw_client = RawExportsClient(client_wrapper=client_wrapper)
@property
def with_raw_response(self) -> RawExportsClient:
"""
Retrieves a raw implementation of this client that returns raw responses.
Returns
-------
RawExportsClient
"""
return self._raw_client
def download_sync(
self,
id: int,
*,
download_all_tasks: typing.Optional[bool] = None,
download_resources: typing.Optional[bool] = None,
export_type: typing.Optional[str] = None,
ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.Iterator[bytes]:
"""
This endpoint is deprecated in Enterprise. Use the async export API instead:
POST /api/projects/{id}/exports/ (see [Create new export](/api#operation/api_projects_exports_create)).
In Label Studio Enterprise, this endpoint will always return a 404 Not Found response with instructions to use the async export API.
Note: if you have a large project it's recommended to use
export snapshots, this easy export endpoint might have timeouts.
Export annotated tasks as a file in a specific format.
For example, to export JSON annotations for a project to a file called `annotations.json`,
run the following from the command line:
```bash
curl -X GET http://localhost:8000/api/projects/{id}/export?exportType=JSON -H 'Authorization: Token abc123' --output 'annotations.json'
```
To export all tasks, including skipped tasks and others without annotations, run the following from the command line:
```bash
curl -X GET http://localhost:8000/api/projects/{id}/export?exportType=JSON&download_all_tasks=true -H 'Authorization: Token abc123' --output 'annotations.json'
```
To export specific tasks with IDs of 123 and 345, run the following from the command line:
```bash
curl -X GET 'http://localhost:8000/api/projects/{id}/export?ids[]=123&ids[]=345' -H 'Authorization: Token abc123' --output 'annotations.json'
```
Parameters
----------
id : int
A unique integer value identifying this project.
download_all_tasks : typing.Optional[bool]
If true, download all tasks regardless of status. If false, download only annotated tasks.
download_resources : typing.Optional[bool]
If true, download all resource files such as images, audio, and others relevant to the tasks.
export_type : typing.Optional[str]
Selected export format (JSON by default)
ids : typing.Optional[typing.Union[str, typing.Sequence[str]]]
Specify a list of task IDs to retrieve only the details for those tasks.
request_options : typing.Optional[RequestOptions]
Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response.
Returns
-------
typing.Iterator[bytes]
Exported data
"""
with self._raw_client.download_sync(
id,
download_all_tasks=download_all_tasks,
download_resources=download_resources,
export_type=export_type,
ids=ids,
request_options=request_options,
) as r:
yield from r.data
def list_formats(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> typing.List[str]:
"""
This endpoint is deprecated in Enterprise. Use the async export API instead:
POST /api/projects/{{id}}/exports/ (see [Create new export](/api#operation/api_projects_exports_create)).
In Label Studio Enterprise, this endpoint will always return a 404 Not Found response with instructions to use the async export API.
Retrieve the available export formats for the current project by ID.
Parameters
----------
id : int
A unique integer value identifying this project.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[str]
Export formats
Examples
--------
from label_studio_sdk import LabelStudio
client = LabelStudio(
api_key="YOUR_API_KEY",
)
client.projects.exports.list_formats(
id=1,
)
"""
_response = self._raw_client.list_formats(id, request_options=request_options)
return _response.data
def list(
self, id: int, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[Export]:
"""
Returns a list of exported files for a specific project by ID.
Parameters
----------
id : int
A unique integer value identifying this project.
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Export]
Examples
--------
from label_studio_sdk import LabelStudio
client = LabelStudio(
api_key="YOUR_API_KEY",
)
client.projects.exports.list(
id=1,
)
"""
_response = self._raw_client.list(id, ordering=ordering, request_options=request_options)
return _response.data
def create(
self,
id: int,
*,
annotation_filter_options: typing.Optional[LseAnnotationFilterOptionsRequest] = OMIT,
converted_formats: typing.Optional[typing.Sequence[ConvertedFormatRequest]] = OMIT,
counters: typing.Optional[typing.Any] = OMIT,
created_by: typing.Optional[UserSimpleRequest] = OMIT,
finished_at: typing.Optional[dt.datetime] = OMIT,
md5: typing.Optional[str] = OMIT,
serialization_options: typing.Optional[SerializationOptionsRequest] = OMIT,
status: typing.Optional[Status7BfEnum] = OMIT,
task_filter_options: typing.Optional[LseTaskFilterOptionsRequest] = OMIT,
title: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> LseExportCreate:
"""
Create a new export request to start a background task and generate an export file for a specific project by ID.
Parameters
----------
id : int
A unique integer value identifying this project.
annotation_filter_options : typing.Optional[LseAnnotationFilterOptionsRequest]
converted_formats : typing.Optional[typing.Sequence[ConvertedFormatRequest]]
counters : typing.Optional[typing.Any]
created_by : typing.Optional[UserSimpleRequest]
finished_at : typing.Optional[dt.datetime]
Complete or fail time
md5 : typing.Optional[str]
serialization_options : typing.Optional[SerializationOptionsRequest]
status : typing.Optional[Status7BfEnum]
task_filter_options : typing.Optional[LseTaskFilterOptionsRequest]
title : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
LseExportCreate
Examples
--------
from label_studio_sdk import LabelStudio
client = LabelStudio(
api_key="YOUR_API_KEY",
)
client.projects.exports.create(
id=1,
)
"""
_response = self._raw_client.create(
id,
annotation_filter_options=annotation_filter_options,
converted_formats=converted_formats,
counters=counters,
created_by=created_by,
finished_at=finished_at,
md5=md5,
serialization_options=serialization_options,
status=status,
task_filter_options=task_filter_options,
title=title,
request_options=request_options,
)
return _response.data
def get(self, id: int, export_pk: int, *, request_options: typing.Optional[RequestOptions] = None) -> Export:
"""
Retrieve information about an export file by export ID for a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Export
Examples
--------
from label_studio_sdk import LabelStudio
client = LabelStudio(
api_key="YOUR_API_KEY",
)
client.projects.exports.get(
id=1,
export_pk=1,
)
"""
_response = self._raw_client.get(id, export_pk, request_options=request_options)
return _response.data
def delete(self, id: int, export_pk: int, *, request_options: typing.Optional[RequestOptions] = None) -> None:
"""
Delete an export file by specified export ID.
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
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.projects.exports.delete(
id=1,
export_pk=1,
)
"""
_response = self._raw_client.delete(id, export_pk, request_options=request_options)
return _response.data
def convert(
self,
id: int,
export_pk: int,
*,
export_type: str,
download_resources: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> ConvertExportsResponse:
"""
Convert export snapshot to selected format
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
export_type : str
Export file format.
download_resources : typing.Optional[bool]
Download resources in converter.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
ConvertExportsResponse
Examples
--------
from label_studio_sdk import LabelStudio
client = LabelStudio(
api_key="YOUR_API_KEY",
)
client.projects.exports.convert(
id=1,
export_pk=1,
export_type="export_type",
)
"""
_response = self._raw_client.convert(
id,
export_pk,
export_type=export_type,
download_resources=download_resources,
request_options=request_options,
)
return _response.data
def download(
self,
id: int,
export_pk: int,
*,
export_type: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.Iterator[bytes]:
"""
Download an export file in the specified format for a specific project. Specify the project ID with the `id`
parameter in the path and the ID of the export file you want to download using the `export_pk` parameter
in the path.
Get the `export_pk` from the response of the request to [Create new export](/api#operation/api_projects_exports_create)
or after [listing export files](/api#operation/api_projects_exports_list).
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
export_type : typing.Optional[str]
Selected export format
request_options : typing.Optional[RequestOptions]
Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response.
Returns
-------
typing.Iterator[bytes]
Export file
"""
with self._raw_client.download(id, export_pk, export_type=export_type, request_options=request_options) as r:
yield from r.data
class AsyncExportsClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._raw_client = AsyncRawExportsClient(client_wrapper=client_wrapper)
@property
def with_raw_response(self) -> AsyncRawExportsClient:
"""
Retrieves a raw implementation of this client that returns raw responses.
Returns
-------
AsyncRawExportsClient
"""
return self._raw_client
async def download_sync(
self,
id: int,
*,
download_all_tasks: typing.Optional[bool] = None,
download_resources: typing.Optional[bool] = None,
export_type: typing.Optional[str] = None,
ids: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.AsyncIterator[bytes]:
"""
This endpoint is deprecated in Enterprise. Use the async export API instead:
POST /api/projects/{id}/exports/ (see [Create new export](/api#operation/api_projects_exports_create)).
In Label Studio Enterprise, this endpoint will always return a 404 Not Found response with instructions to use the async export API.
Note: if you have a large project it's recommended to use
export snapshots, this easy export endpoint might have timeouts.
Export annotated tasks as a file in a specific format.
For example, to export JSON annotations for a project to a file called `annotations.json`,
run the following from the command line:
```bash
curl -X GET http://localhost:8000/api/projects/{id}/export?exportType=JSON -H 'Authorization: Token abc123' --output 'annotations.json'
```
To export all tasks, including skipped tasks and others without annotations, run the following from the command line:
```bash
curl -X GET http://localhost:8000/api/projects/{id}/export?exportType=JSON&download_all_tasks=true -H 'Authorization: Token abc123' --output 'annotations.json'
```
To export specific tasks with IDs of 123 and 345, run the following from the command line:
```bash
curl -X GET 'http://localhost:8000/api/projects/{id}/export?ids[]=123&ids[]=345' -H 'Authorization: Token abc123' --output 'annotations.json'
```
Parameters
----------
id : int
A unique integer value identifying this project.
download_all_tasks : typing.Optional[bool]
If true, download all tasks regardless of status. If false, download only annotated tasks.
download_resources : typing.Optional[bool]
If true, download all resource files such as images, audio, and others relevant to the tasks.
export_type : typing.Optional[str]
Selected export format (JSON by default)
ids : typing.Optional[typing.Union[str, typing.Sequence[str]]]
Specify a list of task IDs to retrieve only the details for those tasks.
request_options : typing.Optional[RequestOptions]
Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response.
Returns
-------
typing.AsyncIterator[bytes]
Exported data
"""
async with self._raw_client.download_sync(
id,
download_all_tasks=download_all_tasks,
download_resources=download_resources,
export_type=export_type,
ids=ids,
request_options=request_options,
) as r:
async for _chunk in r.data:
yield _chunk
async def list_formats(
self, id: int, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[str]:
"""
This endpoint is deprecated in Enterprise. Use the async export API instead:
POST /api/projects/{{id}}/exports/ (see [Create new export](/api#operation/api_projects_exports_create)).
In Label Studio Enterprise, this endpoint will always return a 404 Not Found response with instructions to use the async export API.
Retrieve the available export formats for the current project by ID.
Parameters
----------
id : int
A unique integer value identifying this project.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[str]
Export formats
Examples
--------
import asyncio
from label_studio_sdk import AsyncLabelStudio
client = AsyncLabelStudio(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.projects.exports.list_formats(
id=1,
)
asyncio.run(main())
"""
_response = await self._raw_client.list_formats(id, request_options=request_options)
return _response.data
async def list(
self, id: int, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[Export]:
"""
Returns a list of exported files for a specific project by ID.
Parameters
----------
id : int
A unique integer value identifying this project.
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Export]
Examples
--------
import asyncio
from label_studio_sdk import AsyncLabelStudio
client = AsyncLabelStudio(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.projects.exports.list(
id=1,
)
asyncio.run(main())
"""
_response = await self._raw_client.list(id, ordering=ordering, request_options=request_options)
return _response.data
async def create(
self,
id: int,
*,
annotation_filter_options: typing.Optional[LseAnnotationFilterOptionsRequest] = OMIT,
converted_formats: typing.Optional[typing.Sequence[ConvertedFormatRequest]] = OMIT,
counters: typing.Optional[typing.Any] = OMIT,
created_by: typing.Optional[UserSimpleRequest] = OMIT,
finished_at: typing.Optional[dt.datetime] = OMIT,
md5: typing.Optional[str] = OMIT,
serialization_options: typing.Optional[SerializationOptionsRequest] = OMIT,
status: typing.Optional[Status7BfEnum] = OMIT,
task_filter_options: typing.Optional[LseTaskFilterOptionsRequest] = OMIT,
title: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> LseExportCreate:
"""
Create a new export request to start a background task and generate an export file for a specific project by ID.
Parameters
----------
id : int
A unique integer value identifying this project.
annotation_filter_options : typing.Optional[LseAnnotationFilterOptionsRequest]
converted_formats : typing.Optional[typing.Sequence[ConvertedFormatRequest]]
counters : typing.Optional[typing.Any]
created_by : typing.Optional[UserSimpleRequest]
finished_at : typing.Optional[dt.datetime]
Complete or fail time
md5 : typing.Optional[str]
serialization_options : typing.Optional[SerializationOptionsRequest]
status : typing.Optional[Status7BfEnum]
task_filter_options : typing.Optional[LseTaskFilterOptionsRequest]
title : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
LseExportCreate
Examples
--------
import asyncio
from label_studio_sdk import AsyncLabelStudio
client = AsyncLabelStudio(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.projects.exports.create(
id=1,
)
asyncio.run(main())
"""
_response = await self._raw_client.create(
id,
annotation_filter_options=annotation_filter_options,
converted_formats=converted_formats,
counters=counters,
created_by=created_by,
finished_at=finished_at,
md5=md5,
serialization_options=serialization_options,
status=status,
task_filter_options=task_filter_options,
title=title,
request_options=request_options,
)
return _response.data
async def get(self, id: int, export_pk: int, *, request_options: typing.Optional[RequestOptions] = None) -> Export:
"""
Retrieve information about an export file by export ID for a specific project.
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Export
Examples
--------
import asyncio
from label_studio_sdk import AsyncLabelStudio
client = AsyncLabelStudio(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.projects.exports.get(
id=1,
export_pk=1,
)
asyncio.run(main())
"""
_response = await self._raw_client.get(id, export_pk, request_options=request_options)
return _response.data
async def delete(self, id: int, export_pk: int, *, request_options: typing.Optional[RequestOptions] = None) -> None:
"""
Delete an export file by specified export ID.
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
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.projects.exports.delete(
id=1,
export_pk=1,
)
asyncio.run(main())
"""
_response = await self._raw_client.delete(id, export_pk, request_options=request_options)
return _response.data
async def convert(
self,
id: int,
export_pk: int,
*,
export_type: str,
download_resources: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> ConvertExportsResponse:
"""
Convert export snapshot to selected format
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
export_type : str
Export file format.
download_resources : typing.Optional[bool]
Download resources in converter.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
ConvertExportsResponse
Examples
--------
import asyncio
from label_studio_sdk import AsyncLabelStudio
client = AsyncLabelStudio(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.projects.exports.convert(
id=1,
export_pk=1,
export_type="export_type",
)
asyncio.run(main())
"""
_response = await self._raw_client.convert(
id,
export_pk,
export_type=export_type,
download_resources=download_resources,
request_options=request_options,
)
return _response.data
async def download(
self,
id: int,
export_pk: int,
*,
export_type: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.AsyncIterator[bytes]:
"""
Download an export file in the specified format for a specific project. Specify the project ID with the `id`
parameter in the path and the ID of the export file you want to download using the `export_pk` parameter
in the path.
Get the `export_pk` from the response of the request to [Create new export](/api#operation/api_projects_exports_create)
or after [listing export files](/api#operation/api_projects_exports_list).
Parameters
----------
id : int
A unique integer value identifying this project.
export_pk : int
Primary key identifying the export file.
export_type : typing.Optional[str]
Selected export format
request_options : typing.Optional[RequestOptions]
Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response.
Returns
-------
typing.AsyncIterator[bytes]
Export file
"""
async with self._raw_client.download(
id, export_pk, export_type=export_type, request_options=request_options
) as r:
async for _chunk in r.data:
yield _chunk