# 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.unprocessable_entity_error import UnprocessableEntityError from ...types.add_workspace_invite_response_model import AddWorkspaceInviteResponseModel from ...types.delete_workspace_invite_response_model import DeleteWorkspaceInviteResponseModel from ...types.http_validation_error import HttpValidationError from .types.body_invite_user_v_1_workspace_invites_add_post_workspace_permission import ( BodyInviteUserV1WorkspaceInvitesAddPostWorkspacePermission, ) # 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 create( self, *, email: str, group_ids: typing.Optional[typing.Sequence[str]] = OMIT, workspace_permission: typing.Optional[BodyInviteUserV1WorkspaceInvitesAddPostWorkspacePermission] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AddWorkspaceInviteResponseModel]: """ Sends an email invitation to join your workspace to the provided email. If the user doesn't have an account they will be prompted to create one. If the user accepts this invite they will be added as a user to your workspace and your subscription using one of your seats. This endpoint may only be called by workspace administrators. If the user is already in the workspace a 400 error will be returned. Parameters ---------- email : str The email of the customer group_ids : typing.Optional[typing.Sequence[str]] The group ids of the user workspace_permission : typing.Optional[BodyInviteUserV1WorkspaceInvitesAddPostWorkspacePermission] The workspace permission of the user request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AddWorkspaceInviteResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/workspace/invites/add", method="POST", json={ "email": email, "group_ids": group_ids, "workspace_permission": workspace_permission, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AddWorkspaceInviteResponseModel, construct_type( type_=AddWorkspaceInviteResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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 create_batch( self, *, emails: typing.Sequence[str], group_ids: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AddWorkspaceInviteResponseModel]: """ Sends email invitations to join your workspace to the provided emails. Requires all email addresses to be part of a verified domain. If the users don't have an account they will be prompted to create one. If the users accept these invites they will be added as users to your workspace and your subscription using one of your seats. This endpoint may only be called by workspace administrators. Parameters ---------- emails : typing.Sequence[str] The email of the customer group_ids : typing.Optional[typing.Sequence[str]] The group ids of the user request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AddWorkspaceInviteResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/workspace/invites/add-bulk", method="POST", json={ "emails": emails, "group_ids": group_ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AddWorkspaceInviteResponseModel, construct_type( type_=AddWorkspaceInviteResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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 delete( self, *, email: str, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteWorkspaceInviteResponseModel]: """ Invalidates an existing email invitation. The invitation will still show up in the inbox it has been delivered to, but activating it to join the workspace won't work. This endpoint may only be called by workspace administrators. Parameters ---------- email : str The email of the customer request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteWorkspaceInviteResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/workspace/invites", method="DELETE", json={ "email": email, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteWorkspaceInviteResponseModel, construct_type( type_=DeleteWorkspaceInviteResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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 create( self, *, email: str, group_ids: typing.Optional[typing.Sequence[str]] = OMIT, workspace_permission: typing.Optional[BodyInviteUserV1WorkspaceInvitesAddPostWorkspacePermission] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AddWorkspaceInviteResponseModel]: """ Sends an email invitation to join your workspace to the provided email. If the user doesn't have an account they will be prompted to create one. If the user accepts this invite they will be added as a user to your workspace and your subscription using one of your seats. This endpoint may only be called by workspace administrators. If the user is already in the workspace a 400 error will be returned. Parameters ---------- email : str The email of the customer group_ids : typing.Optional[typing.Sequence[str]] The group ids of the user workspace_permission : typing.Optional[BodyInviteUserV1WorkspaceInvitesAddPostWorkspacePermission] The workspace permission of the user request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AddWorkspaceInviteResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/workspace/invites/add", method="POST", json={ "email": email, "group_ids": group_ids, "workspace_permission": workspace_permission, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AddWorkspaceInviteResponseModel, construct_type( type_=AddWorkspaceInviteResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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 create_batch( self, *, emails: typing.Sequence[str], group_ids: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AddWorkspaceInviteResponseModel]: """ Sends email invitations to join your workspace to the provided emails. Requires all email addresses to be part of a verified domain. If the users don't have an account they will be prompted to create one. If the users accept these invites they will be added as users to your workspace and your subscription using one of your seats. This endpoint may only be called by workspace administrators. Parameters ---------- emails : typing.Sequence[str] The email of the customer group_ids : typing.Optional[typing.Sequence[str]] The group ids of the user request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AddWorkspaceInviteResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/workspace/invites/add-bulk", method="POST", json={ "emails": emails, "group_ids": group_ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AddWorkspaceInviteResponseModel, construct_type( type_=AddWorkspaceInviteResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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 delete( self, *, email: str, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteWorkspaceInviteResponseModel]: """ Invalidates an existing email invitation. The invitation will still show up in the inbox it has been delivered to, but activating it to join the workspace won't work. This endpoint may only be called by workspace administrators. Parameters ---------- email : str The email of the customer request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteWorkspaceInviteResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/workspace/invites", method="DELETE", json={ "email": email, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteWorkspaceInviteResponseModel, construct_type( type_=DeleteWorkspaceInviteResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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)