# 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 ..errors.bad_request_error import BadRequestError
from ..errors.forbidden_error import ForbiddenError
from ..errors.not_found_error import NotFoundError
from ..types.default_role import DefaultRole
from ..types.lse_organization import LseOrganization
from ..types.organization_id import OrganizationId
from ..types.organization_invite import OrganizationInvite
from ..types.role9e7enum import Role9E7Enum
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class RawOrganizationsClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def reset_token(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[OrganizationInvite]:
"""
Reset the token used in the invitation link to invite someone to an organization.
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[OrganizationInvite]
"""
_response = self._client_wrapper.httpx_client.request(
"api/invite/reset-token",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
OrganizationInvite,
construct_type(
type_=OrganizationInvite, # 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[OrganizationId]]:
"""
Return a list of the organizations you've created or that you have access to.
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[OrganizationId]]
"""
_response = self._client_wrapper.httpx_client.request(
"api/organizations/",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[OrganizationId],
construct_type(
type_=typing.List[OrganizationId], # 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: int, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[LseOrganization]:
"""
Retrieve the settings for a specific organization by ID.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[LseOrganization]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/organizations/{jsonable_encoder(id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseOrganization,
construct_type(
type_=LseOrganization, # 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 update(
self,
id: int,
*,
contact_info: typing.Optional[str] = OMIT,
created_by: typing.Optional[int] = OMIT,
custom_scripts_enabled: typing.Optional[bool] = OMIT,
email_notification_settings: typing.Optional[typing.Any] = OMIT,
embed_domains: typing.Optional[typing.Sequence[typing.Dict[str, str]]] = OMIT,
embed_settings: typing.Optional[typing.Any] = OMIT,
title: typing.Optional[str] = OMIT,
token: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[LseOrganization]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update organization details including title, embed domains, and Plugins settings.
Parameters
----------
id : int
contact_info : typing.Optional[str]
created_by : typing.Optional[int]
custom_scripts_enabled : typing.Optional[bool]
Plugins
email_notification_settings : typing.Optional[typing.Any]
Email Notification Settings
embed_domains : typing.Optional[typing.Sequence[typing.Dict[str, str]]]
Supported domains
embed_settings : typing.Optional[typing.Any]
Public Verification Key and Public Verification Algorithms configuration
title : typing.Optional[str]
Organization name
token : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[LseOrganization]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/organizations/{jsonable_encoder(id)}",
method="PATCH",
json={
"contact_info": contact_info,
"created_by": created_by,
"custom_scripts_enabled": custom_scripts_enabled,
"email_notification_settings": email_notification_settings,
"embed_domains": embed_domains,
"embed_settings": embed_settings,
"title": title,
"token": token,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseOrganization,
construct_type(
type_=LseOrganization, # 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(),
),
),
)
if _response.status_code == 403:
raise ForbiddenError(
headers=dict(_response.headers),
body=typing.cast(
typing.Any,
construct_type(
type_=typing.Any, # type: ignore
object_=_response.json(),
),
),
)
if _response.status_code == 404:
raise NotFoundError(
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 update_default_role(
self,
id: int,
*,
annotator_reviewer_firewall_enabled_at: typing.Optional[dt.datetime] = OMIT,
custom_scripts_enabled_at: typing.Optional[dt.datetime] = OMIT,
default_role: typing.Optional[Role9E7Enum] = OMIT,
email_notification_settings: typing.Optional[typing.Any] = OMIT,
embed_domains: typing.Optional[typing.Any] = OMIT,
embed_settings: typing.Optional[typing.Any] = OMIT,
external_id: typing.Optional[str] = OMIT,
extra_data_on_activity_logs: typing.Optional[bool] = OMIT,
label_stream_navigation_disabled_at: typing.Optional[dt.datetime] = OMIT,
organization: typing.Optional[int] = OMIT,
read_only_quick_view_enabled_at: typing.Optional[dt.datetime] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[DefaultRole]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update the default role for members of a specific organization.
Parameters
----------
id : int
annotator_reviewer_firewall_enabled_at : typing.Optional[dt.datetime]
Set to current time to restrict data sharing between annotators and reviewers in the label stream, review stream, and notifications (which will be disabled). In these settings, information about annotator and reviewer identity is suppressed in the UI.
custom_scripts_enabled_at : typing.Optional[dt.datetime]
Set to current time to enable custom scripts (Plugins) for this organization. Can only be enabled if no organization members are active members of any other organizations; otherwise an error will be raised. If this occurs, contact the LEAP team for assistance with enabling custom scripts (Plugins).
default_role : typing.Optional[Role9E7Enum]
Default membership role for invited users
* `OW` - Owner
* `AD` - Administrator
* `MA` - Manager
* `RE` - Reviewer
* `AN` - Annotator
* `DI` - Deactivated
* `NO` - Not Activated
email_notification_settings : typing.Optional[typing.Any]
Email notification settings for this organization. Controls which email notifications users can receive. Structure: {"notifications_allowed": {"notification_type": bool}}
embed_domains : typing.Optional[typing.Any]
List of objects: {"domain": "example.com"}. Used for CSP header on /embed routes.
embed_settings : typing.Optional[typing.Any]
Embed settings for this organization
external_id : typing.Optional[str]
External ID to uniquely identify this organization
extra_data_on_activity_logs : typing.Optional[bool]
label_stream_navigation_disabled_at : typing.Optional[dt.datetime]
Set to current time to disable the label stream navigation for this organization. This will prevent users from going back in the label stream to view previous labels.
organization : typing.Optional[int]
A unique integer value identifying this organization.
read_only_quick_view_enabled_at : typing.Optional[dt.datetime]
Set to current time to prevent creating or editing annotations in quick view.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[DefaultRole]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/organizations/{jsonable_encoder(id)}/set-default-role",
method="PATCH",
json={
"annotator_reviewer_firewall_enabled_at": annotator_reviewer_firewall_enabled_at,
"custom_scripts_enabled_at": custom_scripts_enabled_at,
"default_role": default_role,
"email_notification_settings": email_notification_settings,
"embed_domains": embed_domains,
"embed_settings": embed_settings,
"external_id": external_id,
"extra_data_on_activity_logs": extra_data_on_activity_logs,
"label_stream_navigation_disabled_at": label_stream_navigation_disabled_at,
"organization": organization,
"read_only_quick_view_enabled_at": read_only_quick_view_enabled_at,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
DefaultRole,
construct_type(
type_=DefaultRole, # 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 AsyncRawOrganizationsClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def reset_token(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[OrganizationInvite]:
"""
Reset the token used in the invitation link to invite someone to an organization.
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[OrganizationInvite]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/invite/reset-token",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
OrganizationInvite,
construct_type(
type_=OrganizationInvite, # 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[OrganizationId]]:
"""
Return a list of the organizations you've created or that you have access to.
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[OrganizationId]]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/organizations/",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[OrganizationId],
construct_type(
type_=typing.List[OrganizationId], # 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: int, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[LseOrganization]:
"""
Retrieve the settings for a specific organization by ID.
Parameters
----------
id : int
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[LseOrganization]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/organizations/{jsonable_encoder(id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseOrganization,
construct_type(
type_=LseOrganization, # 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 update(
self,
id: int,
*,
contact_info: typing.Optional[str] = OMIT,
created_by: typing.Optional[int] = OMIT,
custom_scripts_enabled: typing.Optional[bool] = OMIT,
email_notification_settings: typing.Optional[typing.Any] = OMIT,
embed_domains: typing.Optional[typing.Sequence[typing.Dict[str, str]]] = OMIT,
embed_settings: typing.Optional[typing.Any] = OMIT,
title: typing.Optional[str] = OMIT,
token: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[LseOrganization]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update organization details including title, embed domains, and Plugins settings.
Parameters
----------
id : int
contact_info : typing.Optional[str]
created_by : typing.Optional[int]
custom_scripts_enabled : typing.Optional[bool]
Plugins
email_notification_settings : typing.Optional[typing.Any]
Email Notification Settings
embed_domains : typing.Optional[typing.Sequence[typing.Dict[str, str]]]
Supported domains
embed_settings : typing.Optional[typing.Any]
Public Verification Key and Public Verification Algorithms configuration
title : typing.Optional[str]
Organization name
token : typing.Optional[str]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[LseOrganization]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/organizations/{jsonable_encoder(id)}",
method="PATCH",
json={
"contact_info": contact_info,
"created_by": created_by,
"custom_scripts_enabled": custom_scripts_enabled,
"email_notification_settings": email_notification_settings,
"embed_domains": embed_domains,
"embed_settings": embed_settings,
"title": title,
"token": token,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
LseOrganization,
construct_type(
type_=LseOrganization, # 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(),
),
),
)
if _response.status_code == 403:
raise ForbiddenError(
headers=dict(_response.headers),
body=typing.cast(
typing.Any,
construct_type(
type_=typing.Any, # type: ignore
object_=_response.json(),
),
),
)
if _response.status_code == 404:
raise NotFoundError(
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 update_default_role(
self,
id: int,
*,
annotator_reviewer_firewall_enabled_at: typing.Optional[dt.datetime] = OMIT,
custom_scripts_enabled_at: typing.Optional[dt.datetime] = OMIT,
default_role: typing.Optional[Role9E7Enum] = OMIT,
email_notification_settings: typing.Optional[typing.Any] = OMIT,
embed_domains: typing.Optional[typing.Any] = OMIT,
embed_settings: typing.Optional[typing.Any] = OMIT,
external_id: typing.Optional[str] = OMIT,
extra_data_on_activity_logs: typing.Optional[bool] = OMIT,
label_stream_navigation_disabled_at: typing.Optional[dt.datetime] = OMIT,
organization: typing.Optional[int] = OMIT,
read_only_quick_view_enabled_at: typing.Optional[dt.datetime] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[DefaultRole]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update the default role for members of a specific organization.
Parameters
----------
id : int
annotator_reviewer_firewall_enabled_at : typing.Optional[dt.datetime]
Set to current time to restrict data sharing between annotators and reviewers in the label stream, review stream, and notifications (which will be disabled). In these settings, information about annotator and reviewer identity is suppressed in the UI.
custom_scripts_enabled_at : typing.Optional[dt.datetime]
Set to current time to enable custom scripts (Plugins) for this organization. Can only be enabled if no organization members are active members of any other organizations; otherwise an error will be raised. If this occurs, contact the LEAP team for assistance with enabling custom scripts (Plugins).
default_role : typing.Optional[Role9E7Enum]
Default membership role for invited users
* `OW` - Owner
* `AD` - Administrator
* `MA` - Manager
* `RE` - Reviewer
* `AN` - Annotator
* `DI` - Deactivated
* `NO` - Not Activated
email_notification_settings : typing.Optional[typing.Any]
Email notification settings for this organization. Controls which email notifications users can receive. Structure: {"notifications_allowed": {"notification_type": bool}}
embed_domains : typing.Optional[typing.Any]
List of objects: {"domain": "example.com"}. Used for CSP header on /embed routes.
embed_settings : typing.Optional[typing.Any]
Embed settings for this organization
external_id : typing.Optional[str]
External ID to uniquely identify this organization
extra_data_on_activity_logs : typing.Optional[bool]
label_stream_navigation_disabled_at : typing.Optional[dt.datetime]
Set to current time to disable the label stream navigation for this organization. This will prevent users from going back in the label stream to view previous labels.
organization : typing.Optional[int]
A unique integer value identifying this organization.
read_only_quick_view_enabled_at : typing.Optional[dt.datetime]
Set to current time to prevent creating or editing annotations in quick view.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[DefaultRole]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/organizations/{jsonable_encoder(id)}/set-default-role",
method="PATCH",
json={
"annotator_reviewer_firewall_enabled_at": annotator_reviewer_firewall_enabled_at,
"custom_scripts_enabled_at": custom_scripts_enabled_at,
"default_role": default_role,
"email_notification_settings": email_notification_settings,
"embed_domains": embed_domains,
"embed_settings": embed_settings,
"external_id": external_id,
"extra_data_on_activity_logs": extra_data_on_activity_logs,
"label_stream_navigation_disabled_at": label_stream_navigation_disabled_at,
"organization": organization,
"read_only_quick_view_enabled_at": read_only_quick_view_enabled_at,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
DefaultRole,
construct_type(
type_=DefaultRole, # 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)