# This file was auto-generated by Fern from our API Definition.
import datetime as dt
import typing
from json.decoder import JSONDecodeError
from ...core.api_error import ApiError
from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ...core.http_response import AsyncHttpResponse, HttpResponse
from ...core.jsonable_encoder import jsonable_encoder
from ...core.request_options import RequestOptions
from ...core.unchecked_base_model import construct_type
from ...types.lse_s3export_storage import LseS3ExportStorage
from ...types.status_c5a_enum import StatusC5AEnum
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class RawS3SClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def list(
self,
*,
project: int,
ordering: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[typing.List[LseS3ExportStorage]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Get a list of all S3 export storage connections that were set up with IAM role access.
Parameters
----------
project : int
Project ID
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[typing.List[LseS3ExportStorage]]
"""
_response = self._client_wrapper.httpx_client.request(
"api/storages/export/s3s",
method="GET",
params={
"ordering": ordering,
"project": project,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[LseS3ExportStorage],
construct_type(
type_=typing.List[LseS3ExportStorage], # 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,
*,
project: int,
role_arn: str,
aws_access_key_id: typing.Optional[str] = OMIT,
aws_secret_access_key: typing.Optional[str] = OMIT,
aws_session_token: typing.Optional[str] = OMIT,
aws_sse_kms_key_id: typing.Optional[str] = OMIT,
bucket: typing.Optional[str] = OMIT,
can_delete_objects: typing.Optional[bool] = OMIT,
description: typing.Optional[str] = OMIT,
external_id: typing.Optional[str] = OMIT,
last_sync: typing.Optional[dt.datetime] = OMIT,
last_sync_count: typing.Optional[int] = OMIT,
last_sync_job: typing.Optional[str] = OMIT,
legacy_auth: typing.Optional[bool] = OMIT,
meta: typing.Optional[typing.Any] = OMIT,
prefix: typing.Optional[str] = OMIT,
regex_filter: typing.Optional[str] = OMIT,
region_name: typing.Optional[str] = OMIT,
s3endpoint: typing.Optional[str] = OMIT,
status: typing.Optional[StatusC5AEnum] = OMIT,
synchronizable: typing.Optional[bool] = OMIT,
title: typing.Optional[str] = OMIT,
traceback: typing.Optional[str] = OMIT,
use_blob_urls: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create an S3 export storage connection with IAM role access to store annotations.
Parameters
----------
project : int
A unique integer value identifying this project.
role_arn : str
AWS RoleArn
aws_access_key_id : typing.Optional[str]
AWS_ACCESS_KEY_ID
aws_secret_access_key : typing.Optional[str]
AWS_SECRET_ACCESS_KEY
aws_session_token : typing.Optional[str]
AWS_SESSION_TOKEN
aws_sse_kms_key_id : typing.Optional[str]
AWS SSE KMS Key ID
bucket : typing.Optional[str]
S3 bucket name
can_delete_objects : typing.Optional[bool]
Deletion from storage enabled
description : typing.Optional[str]
Cloud storage description
external_id : typing.Optional[str]
AWS ExternalId
last_sync : typing.Optional[dt.datetime]
Last sync finished time
last_sync_count : typing.Optional[int]
Count of tasks synced last time
last_sync_job : typing.Optional[str]
Last sync job ID
legacy_auth : typing.Optional[bool]
meta : typing.Optional[typing.Any]
Meta and debug information about storage processes
prefix : typing.Optional[str]
S3 bucket prefix
regex_filter : typing.Optional[str]
Cloud storage regex for filtering objects
region_name : typing.Optional[str]
AWS Region
s3endpoint : typing.Optional[str]
S3 Endpoint
status : typing.Optional[StatusC5AEnum]
synchronizable : typing.Optional[bool]
title : typing.Optional[str]
Cloud storage title
traceback : typing.Optional[str]
Traceback report for the last failed sync
use_blob_urls : typing.Optional[bool]
Interpret objects as BLOBs and generate URLs
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[LseS3ExportStorage]
"""
_response = self._client_wrapper.httpx_client.request(
"api/storages/export/s3s",
method="POST",
json={
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_sse_kms_key_id": aws_sse_kms_key_id,
"bucket": bucket,
"can_delete_objects": can_delete_objects,
"description": description,
"external_id": external_id,
"last_sync": last_sync,
"last_sync_count": last_sync_count,
"last_sync_job": last_sync_job,
"legacy_auth": legacy_auth,
"meta": meta,
"prefix": prefix,
"project": project,
"regex_filter": regex_filter,
"region_name": region_name,
"role_arn": role_arn,
"s3_endpoint": s3endpoint,
"status": status,
"synchronizable": synchronizable,
"title": title,
"traceback": traceback,
"use_blob_urls": use_blob_urls,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # 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 validate(
self,
*,
project: int,
role_arn: str,
aws_access_key_id: typing.Optional[str] = OMIT,
aws_secret_access_key: typing.Optional[str] = OMIT,
aws_session_token: typing.Optional[str] = OMIT,
aws_sse_kms_key_id: typing.Optional[str] = OMIT,
bucket: typing.Optional[str] = OMIT,
can_delete_objects: typing.Optional[bool] = OMIT,
description: typing.Optional[str] = OMIT,
external_id: typing.Optional[str] = OMIT,
last_sync: typing.Optional[dt.datetime] = OMIT,
last_sync_count: typing.Optional[int] = OMIT,
last_sync_job: typing.Optional[str] = OMIT,
legacy_auth: typing.Optional[bool] = OMIT,
meta: typing.Optional[typing.Any] = OMIT,
prefix: typing.Optional[str] = OMIT,
regex_filter: typing.Optional[str] = OMIT,
region_name: typing.Optional[str] = OMIT,
s3endpoint: typing.Optional[str] = OMIT,
status: typing.Optional[StatusC5AEnum] = OMIT,
synchronizable: typing.Optional[bool] = OMIT,
title: typing.Optional[str] = OMIT,
traceback: typing.Optional[str] = OMIT,
use_blob_urls: typing.Optional[bool] = OMIT,
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)
Validate a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
project : int
A unique integer value identifying this project.
role_arn : str
AWS RoleArn
aws_access_key_id : typing.Optional[str]
AWS_ACCESS_KEY_ID
aws_secret_access_key : typing.Optional[str]
AWS_SECRET_ACCESS_KEY
aws_session_token : typing.Optional[str]
AWS_SESSION_TOKEN
aws_sse_kms_key_id : typing.Optional[str]
AWS SSE KMS Key ID
bucket : typing.Optional[str]
S3 bucket name
can_delete_objects : typing.Optional[bool]
Deletion from storage enabled
description : typing.Optional[str]
Cloud storage description
external_id : typing.Optional[str]
AWS ExternalId
last_sync : typing.Optional[dt.datetime]
Last sync finished time
last_sync_count : typing.Optional[int]
Count of tasks synced last time
last_sync_job : typing.Optional[str]
Last sync job ID
legacy_auth : typing.Optional[bool]
meta : typing.Optional[typing.Any]
Meta and debug information about storage processes
prefix : typing.Optional[str]
S3 bucket prefix
regex_filter : typing.Optional[str]
Cloud storage regex for filtering objects
region_name : typing.Optional[str]
AWS Region
s3endpoint : typing.Optional[str]
S3 Endpoint
status : typing.Optional[StatusC5AEnum]
synchronizable : typing.Optional[bool]
title : typing.Optional[str]
Cloud storage title
traceback : typing.Optional[str]
Traceback report for the last failed sync
use_blob_urls : typing.Optional[bool]
Interpret objects as BLOBs and generate URLs
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
"api/storages/export/s3s/validate",
method="POST",
json={
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_sse_kms_key_id": aws_sse_kms_key_id,
"bucket": bucket,
"can_delete_objects": can_delete_objects,
"description": description,
"external_id": external_id,
"last_sync": last_sync,
"last_sync_count": last_sync_count,
"last_sync_job": last_sync_job,
"legacy_auth": legacy_auth,
"meta": meta,
"prefix": prefix,
"project": project,
"regex_filter": regex_filter,
"region_name": region_name,
"role_arn": role_arn,
"s3_endpoint": s3endpoint,
"status": status,
"synchronizable": synchronizable,
"title": title,
"traceback": traceback,
"use_blob_urls": use_blob_urls,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
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 get(
self, id: int, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Get a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[LseS3ExportStorage]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # type: ignore
object_=_response.json(),
),
)
return HttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def delete(self, id: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Delete a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}",
method="DELETE",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return HttpResponse(response=_response, data=None)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def update(
self,
id: int,
*,
aws_access_key_id: typing.Optional[str] = OMIT,
aws_secret_access_key: typing.Optional[str] = OMIT,
aws_session_token: typing.Optional[str] = OMIT,
aws_sse_kms_key_id: typing.Optional[str] = OMIT,
bucket: typing.Optional[str] = OMIT,
can_delete_objects: typing.Optional[bool] = OMIT,
description: typing.Optional[str] = OMIT,
external_id: typing.Optional[str] = OMIT,
last_sync: typing.Optional[dt.datetime] = OMIT,
last_sync_count: typing.Optional[int] = OMIT,
last_sync_job: typing.Optional[str] = OMIT,
legacy_auth: typing.Optional[bool] = OMIT,
meta: typing.Optional[typing.Any] = OMIT,
prefix: typing.Optional[str] = OMIT,
project: typing.Optional[int] = OMIT,
regex_filter: typing.Optional[str] = OMIT,
region_name: typing.Optional[str] = OMIT,
role_arn: typing.Optional[str] = OMIT,
s3endpoint: typing.Optional[str] = OMIT,
status: typing.Optional[StatusC5AEnum] = OMIT,
synchronizable: typing.Optional[bool] = OMIT,
title: typing.Optional[str] = OMIT,
traceback: typing.Optional[str] = OMIT,
use_blob_urls: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
id : int
aws_access_key_id : typing.Optional[str]
AWS_ACCESS_KEY_ID
aws_secret_access_key : typing.Optional[str]
AWS_SECRET_ACCESS_KEY
aws_session_token : typing.Optional[str]
AWS_SESSION_TOKEN
aws_sse_kms_key_id : typing.Optional[str]
AWS SSE KMS Key ID
bucket : typing.Optional[str]
S3 bucket name
can_delete_objects : typing.Optional[bool]
Deletion from storage enabled
description : typing.Optional[str]
Cloud storage description
external_id : typing.Optional[str]
AWS ExternalId
last_sync : typing.Optional[dt.datetime]
Last sync finished time
last_sync_count : typing.Optional[int]
Count of tasks synced last time
last_sync_job : typing.Optional[str]
Last sync job ID
legacy_auth : typing.Optional[bool]
meta : typing.Optional[typing.Any]
Meta and debug information about storage processes
prefix : typing.Optional[str]
S3 bucket prefix
project : typing.Optional[int]
A unique integer value identifying this project.
regex_filter : typing.Optional[str]
Cloud storage regex for filtering objects
region_name : typing.Optional[str]
AWS Region
role_arn : typing.Optional[str]
AWS RoleArn
s3endpoint : typing.Optional[str]
S3 Endpoint
status : typing.Optional[StatusC5AEnum]
synchronizable : typing.Optional[bool]
title : typing.Optional[str]
Cloud storage title
traceback : typing.Optional[str]
Traceback report for the last failed sync
use_blob_urls : typing.Optional[bool]
Interpret objects as BLOBs and generate URLs
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[LseS3ExportStorage]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}",
method="PATCH",
json={
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_sse_kms_key_id": aws_sse_kms_key_id,
"bucket": bucket,
"can_delete_objects": can_delete_objects,
"description": description,
"external_id": external_id,
"last_sync": last_sync,
"last_sync_count": last_sync_count,
"last_sync_job": last_sync_job,
"legacy_auth": legacy_auth,
"meta": meta,
"prefix": prefix,
"project": project,
"regex_filter": regex_filter,
"region_name": region_name,
"role_arn": role_arn,
"s3_endpoint": s3endpoint,
"status": status,
"synchronizable": synchronizable,
"title": title,
"traceback": traceback,
"use_blob_urls": use_blob_urls,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # 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 sync(
self, id: int, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Sync tasks from an S3 export storage.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[LseS3ExportStorage]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}/sync",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # 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 AsyncRawS3SClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def list(
self,
*,
project: int,
ordering: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[typing.List[LseS3ExportStorage]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Get a list of all S3 export storage connections that were set up with IAM role access.
Parameters
----------
project : int
Project ID
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[typing.List[LseS3ExportStorage]]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/storages/export/s3s",
method="GET",
params={
"ordering": ordering,
"project": project,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[LseS3ExportStorage],
construct_type(
type_=typing.List[LseS3ExportStorage], # 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,
*,
project: int,
role_arn: str,
aws_access_key_id: typing.Optional[str] = OMIT,
aws_secret_access_key: typing.Optional[str] = OMIT,
aws_session_token: typing.Optional[str] = OMIT,
aws_sse_kms_key_id: typing.Optional[str] = OMIT,
bucket: typing.Optional[str] = OMIT,
can_delete_objects: typing.Optional[bool] = OMIT,
description: typing.Optional[str] = OMIT,
external_id: typing.Optional[str] = OMIT,
last_sync: typing.Optional[dt.datetime] = OMIT,
last_sync_count: typing.Optional[int] = OMIT,
last_sync_job: typing.Optional[str] = OMIT,
legacy_auth: typing.Optional[bool] = OMIT,
meta: typing.Optional[typing.Any] = OMIT,
prefix: typing.Optional[str] = OMIT,
regex_filter: typing.Optional[str] = OMIT,
region_name: typing.Optional[str] = OMIT,
s3endpoint: typing.Optional[str] = OMIT,
status: typing.Optional[StatusC5AEnum] = OMIT,
synchronizable: typing.Optional[bool] = OMIT,
title: typing.Optional[str] = OMIT,
traceback: typing.Optional[str] = OMIT,
use_blob_urls: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create an S3 export storage connection with IAM role access to store annotations.
Parameters
----------
project : int
A unique integer value identifying this project.
role_arn : str
AWS RoleArn
aws_access_key_id : typing.Optional[str]
AWS_ACCESS_KEY_ID
aws_secret_access_key : typing.Optional[str]
AWS_SECRET_ACCESS_KEY
aws_session_token : typing.Optional[str]
AWS_SESSION_TOKEN
aws_sse_kms_key_id : typing.Optional[str]
AWS SSE KMS Key ID
bucket : typing.Optional[str]
S3 bucket name
can_delete_objects : typing.Optional[bool]
Deletion from storage enabled
description : typing.Optional[str]
Cloud storage description
external_id : typing.Optional[str]
AWS ExternalId
last_sync : typing.Optional[dt.datetime]
Last sync finished time
last_sync_count : typing.Optional[int]
Count of tasks synced last time
last_sync_job : typing.Optional[str]
Last sync job ID
legacy_auth : typing.Optional[bool]
meta : typing.Optional[typing.Any]
Meta and debug information about storage processes
prefix : typing.Optional[str]
S3 bucket prefix
regex_filter : typing.Optional[str]
Cloud storage regex for filtering objects
region_name : typing.Optional[str]
AWS Region
s3endpoint : typing.Optional[str]
S3 Endpoint
status : typing.Optional[StatusC5AEnum]
synchronizable : typing.Optional[bool]
title : typing.Optional[str]
Cloud storage title
traceback : typing.Optional[str]
Traceback report for the last failed sync
use_blob_urls : typing.Optional[bool]
Interpret objects as BLOBs and generate URLs
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[LseS3ExportStorage]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/storages/export/s3s",
method="POST",
json={
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_sse_kms_key_id": aws_sse_kms_key_id,
"bucket": bucket,
"can_delete_objects": can_delete_objects,
"description": description,
"external_id": external_id,
"last_sync": last_sync,
"last_sync_count": last_sync_count,
"last_sync_job": last_sync_job,
"legacy_auth": legacy_auth,
"meta": meta,
"prefix": prefix,
"project": project,
"regex_filter": regex_filter,
"region_name": region_name,
"role_arn": role_arn,
"s3_endpoint": s3endpoint,
"status": status,
"synchronizable": synchronizable,
"title": title,
"traceback": traceback,
"use_blob_urls": use_blob_urls,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # 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 validate(
self,
*,
project: int,
role_arn: str,
aws_access_key_id: typing.Optional[str] = OMIT,
aws_secret_access_key: typing.Optional[str] = OMIT,
aws_session_token: typing.Optional[str] = OMIT,
aws_sse_kms_key_id: typing.Optional[str] = OMIT,
bucket: typing.Optional[str] = OMIT,
can_delete_objects: typing.Optional[bool] = OMIT,
description: typing.Optional[str] = OMIT,
external_id: typing.Optional[str] = OMIT,
last_sync: typing.Optional[dt.datetime] = OMIT,
last_sync_count: typing.Optional[int] = OMIT,
last_sync_job: typing.Optional[str] = OMIT,
legacy_auth: typing.Optional[bool] = OMIT,
meta: typing.Optional[typing.Any] = OMIT,
prefix: typing.Optional[str] = OMIT,
regex_filter: typing.Optional[str] = OMIT,
region_name: typing.Optional[str] = OMIT,
s3endpoint: typing.Optional[str] = OMIT,
status: typing.Optional[StatusC5AEnum] = OMIT,
synchronizable: typing.Optional[bool] = OMIT,
title: typing.Optional[str] = OMIT,
traceback: typing.Optional[str] = OMIT,
use_blob_urls: typing.Optional[bool] = OMIT,
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)
Validate a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
project : int
A unique integer value identifying this project.
role_arn : str
AWS RoleArn
aws_access_key_id : typing.Optional[str]
AWS_ACCESS_KEY_ID
aws_secret_access_key : typing.Optional[str]
AWS_SECRET_ACCESS_KEY
aws_session_token : typing.Optional[str]
AWS_SESSION_TOKEN
aws_sse_kms_key_id : typing.Optional[str]
AWS SSE KMS Key ID
bucket : typing.Optional[str]
S3 bucket name
can_delete_objects : typing.Optional[bool]
Deletion from storage enabled
description : typing.Optional[str]
Cloud storage description
external_id : typing.Optional[str]
AWS ExternalId
last_sync : typing.Optional[dt.datetime]
Last sync finished time
last_sync_count : typing.Optional[int]
Count of tasks synced last time
last_sync_job : typing.Optional[str]
Last sync job ID
legacy_auth : typing.Optional[bool]
meta : typing.Optional[typing.Any]
Meta and debug information about storage processes
prefix : typing.Optional[str]
S3 bucket prefix
regex_filter : typing.Optional[str]
Cloud storage regex for filtering objects
region_name : typing.Optional[str]
AWS Region
s3endpoint : typing.Optional[str]
S3 Endpoint
status : typing.Optional[StatusC5AEnum]
synchronizable : typing.Optional[bool]
title : typing.Optional[str]
Cloud storage title
traceback : typing.Optional[str]
Traceback report for the last failed sync
use_blob_urls : typing.Optional[bool]
Interpret objects as BLOBs and generate URLs
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/storages/export/s3s/validate",
method="POST",
json={
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_sse_kms_key_id": aws_sse_kms_key_id,
"bucket": bucket,
"can_delete_objects": can_delete_objects,
"description": description,
"external_id": external_id,
"last_sync": last_sync,
"last_sync_count": last_sync_count,
"last_sync_job": last_sync_job,
"legacy_auth": legacy_auth,
"meta": meta,
"prefix": prefix,
"project": project,
"regex_filter": regex_filter,
"region_name": region_name,
"role_arn": role_arn,
"s3_endpoint": s3endpoint,
"status": status,
"synchronizable": synchronizable,
"title": title,
"traceback": traceback,
"use_blob_urls": use_blob_urls,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
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 get(
self, id: int, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Get a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[LseS3ExportStorage]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # type: ignore
object_=_response.json(),
),
)
return AsyncHttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def delete(
self, id: int, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[None]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Delete a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}",
method="DELETE",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return AsyncHttpResponse(response=_response, data=None)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def update(
self,
id: int,
*,
aws_access_key_id: typing.Optional[str] = OMIT,
aws_secret_access_key: typing.Optional[str] = OMIT,
aws_session_token: typing.Optional[str] = OMIT,
aws_sse_kms_key_id: typing.Optional[str] = OMIT,
bucket: typing.Optional[str] = OMIT,
can_delete_objects: typing.Optional[bool] = OMIT,
description: typing.Optional[str] = OMIT,
external_id: typing.Optional[str] = OMIT,
last_sync: typing.Optional[dt.datetime] = OMIT,
last_sync_count: typing.Optional[int] = OMIT,
last_sync_job: typing.Optional[str] = OMIT,
legacy_auth: typing.Optional[bool] = OMIT,
meta: typing.Optional[typing.Any] = OMIT,
prefix: typing.Optional[str] = OMIT,
project: typing.Optional[int] = OMIT,
regex_filter: typing.Optional[str] = OMIT,
region_name: typing.Optional[str] = OMIT,
role_arn: typing.Optional[str] = OMIT,
s3endpoint: typing.Optional[str] = OMIT,
status: typing.Optional[StatusC5AEnum] = OMIT,
synchronizable: typing.Optional[bool] = OMIT,
title: typing.Optional[str] = OMIT,
traceback: typing.Optional[str] = OMIT,
use_blob_urls: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update a specific S3 export storage connection that was set up with IAM role access.
Parameters
----------
id : int
aws_access_key_id : typing.Optional[str]
AWS_ACCESS_KEY_ID
aws_secret_access_key : typing.Optional[str]
AWS_SECRET_ACCESS_KEY
aws_session_token : typing.Optional[str]
AWS_SESSION_TOKEN
aws_sse_kms_key_id : typing.Optional[str]
AWS SSE KMS Key ID
bucket : typing.Optional[str]
S3 bucket name
can_delete_objects : typing.Optional[bool]
Deletion from storage enabled
description : typing.Optional[str]
Cloud storage description
external_id : typing.Optional[str]
AWS ExternalId
last_sync : typing.Optional[dt.datetime]
Last sync finished time
last_sync_count : typing.Optional[int]
Count of tasks synced last time
last_sync_job : typing.Optional[str]
Last sync job ID
legacy_auth : typing.Optional[bool]
meta : typing.Optional[typing.Any]
Meta and debug information about storage processes
prefix : typing.Optional[str]
S3 bucket prefix
project : typing.Optional[int]
A unique integer value identifying this project.
regex_filter : typing.Optional[str]
Cloud storage regex for filtering objects
region_name : typing.Optional[str]
AWS Region
role_arn : typing.Optional[str]
AWS RoleArn
s3endpoint : typing.Optional[str]
S3 Endpoint
status : typing.Optional[StatusC5AEnum]
synchronizable : typing.Optional[bool]
title : typing.Optional[str]
Cloud storage title
traceback : typing.Optional[str]
Traceback report for the last failed sync
use_blob_urls : typing.Optional[bool]
Interpret objects as BLOBs and generate URLs
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[LseS3ExportStorage]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}",
method="PATCH",
json={
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_sse_kms_key_id": aws_sse_kms_key_id,
"bucket": bucket,
"can_delete_objects": can_delete_objects,
"description": description,
"external_id": external_id,
"last_sync": last_sync,
"last_sync_count": last_sync_count,
"last_sync_job": last_sync_job,
"legacy_auth": legacy_auth,
"meta": meta,
"prefix": prefix,
"project": project,
"regex_filter": regex_filter,
"region_name": region_name,
"role_arn": role_arn,
"s3_endpoint": s3endpoint,
"status": status,
"synchronizable": synchronizable,
"title": title,
"traceback": traceback,
"use_blob_urls": use_blob_urls,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # 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 sync(
self, id: int, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[LseS3ExportStorage]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Sync tasks from an S3 export storage.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[LseS3ExportStorage]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/storages/export/s3s/{jsonable_encoder(id)}/sync",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseS3ExportStorage,
construct_type(
type_=LseS3ExportStorage, # 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)