# 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.request_options import RequestOptions
from ...core.unchecked_base_model import construct_type
from ...errors.forbidden_error import ForbiddenError
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 RawInvitesClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def get_invite_link(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[OrganizationInvite]:
"""
Get invite link for organization
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[OrganizationInvite]
"""
_response = self._client_wrapper.httpx_client.request(
"api/invite",
method="GET",
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)
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(),
),
),
)
_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 revoke_invite(
self, *, email: 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)
Revoke invite to organization
Parameters
----------
email : str
Email address
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
"api/invite/revoke",
method="POST",
json={
"email": email,
},
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 send_email(
self,
*,
emails: typing.Sequence[str],
role: Role9E7Enum,
projects: typing.Optional[typing.Sequence[int]] = OMIT,
workspaces: typing.Optional[typing.Sequence[int]] = 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)
Send email with invite to organization
Parameters
----------
emails : typing.Sequence[str]
Email addresses
role : Role9E7Enum
Organization role
* `OW` - Owner
* `AD` - Administrator
* `MA` - Manager
* `RE` - Reviewer
* `AN` - Annotator
* `DI` - Deactivated
* `NO` - Not Activated
projects : typing.Optional[typing.Sequence[int]]
Project IDs to grant access to
workspaces : typing.Optional[typing.Sequence[int]]
Workspace IDs to grant access to
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
"api/invite/send-email",
method="POST",
json={
"emails": emails,
"projects": projects,
"role": role,
"workspaces": workspaces,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return HttpResponse(response=_response, data=None)
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(),
),
),
)
_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 AsyncRawInvitesClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def get_invite_link(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[OrganizationInvite]:
"""
Get invite link for organization
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[OrganizationInvite]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/invite",
method="GET",
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)
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(),
),
),
)
_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 revoke_invite(
self, *, email: 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)
Revoke invite to organization
Parameters
----------
email : str
Email address
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/invite/revoke",
method="POST",
json={
"email": email,
},
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 send_email(
self,
*,
emails: typing.Sequence[str],
role: Role9E7Enum,
projects: typing.Optional[typing.Sequence[int]] = OMIT,
workspaces: typing.Optional[typing.Sequence[int]] = 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)
Send email with invite to organization
Parameters
----------
emails : typing.Sequence[str]
Email addresses
role : Role9E7Enum
Organization role
* `OW` - Owner
* `AD` - Administrator
* `MA` - Manager
* `RE` - Reviewer
* `AN` - Annotator
* `DI` - Deactivated
* `NO` - Not Activated
projects : typing.Optional[typing.Sequence[int]]
Project IDs to grant access to
workspaces : typing.Optional[typing.Sequence[int]]
Workspace IDs to grant access to
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/invite/send-email",
method="POST",
json={
"emails": emails,
"projects": projects,
"role": role,
"workspaces": workspaces,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return AsyncHttpResponse(response=_response, data=None)
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(),
),
),
)
_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)