# 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.jsonable_encoder import jsonable_encoder from ...core.request_options import RequestOptions from ...core.serialization import convert_and_respect_annotation_metadata from ...core.unchecked_base_model import construct_type from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.http_validation_error import HttpValidationError from ...types.workspace_api_key_list_response_model import WorkspaceApiKeyListResponseModel from ...types.workspace_create_api_key_response_model import WorkspaceCreateApiKeyResponseModel from .types.body_create_service_account_api_key_v_1_service_accounts_service_account_user_id_api_keys_post_permissions import ( BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions, ) from .types.body_edit_service_account_api_key_v_1_service_accounts_service_account_user_id_api_keys_api_key_id_patch_permissions import ( BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions, ) # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawApiKeysClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, service_account_user_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[WorkspaceApiKeyListResponseModel]: """ Get all API keys for a service account Parameters ---------- service_account_user_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceApiKeyListResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceApiKeyListResponseModel, construct_type( type_=WorkspaceApiKeyListResponseModel, # 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( self, service_account_user_id: str, *, name: str, permissions: BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions, character_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[WorkspaceCreateApiKeyResponseModel]: """ Create a new API key for a service account Parameters ---------- service_account_user_id : str name : str permissions : BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions The permissions of the XI API. character_limit : typing.Optional[int] The character limit of the XI API key. If provided this will limit the usage of this api key to n characters per month where n is the chosen value. Requests that incur charges will fail after reaching this monthly limit. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[WorkspaceCreateApiKeyResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys", method="POST", json={ "name": name, "permissions": convert_and_respect_annotation_metadata( object_=permissions, annotation=BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions, direction="write", ), "character_limit": character_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceCreateApiKeyResponseModel, construct_type( type_=WorkspaceCreateApiKeyResponseModel, # 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, service_account_user_id: str, api_key_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.Any]: """ Delete an existing API key for a service account Parameters ---------- service_account_user_id : str api_key_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys/{jsonable_encoder(api_key_id)}", method="DELETE", request_options=request_options, ) try: if _response is None or not _response.text.strip(): return HttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # 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 update( self, service_account_user_id: str, api_key_id: str, *, is_enabled: bool, name: str, permissions: BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions, character_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.Any]: """ Update an existing API key for a service account Parameters ---------- service_account_user_id : str api_key_id : str is_enabled : bool Whether to enable or disable the API key. name : str The name of the XI API key to use (used for identification purposes only). permissions : BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions The permissions of the XI API. character_limit : typing.Optional[int] The character limit of the XI API key. If provided this will limit the usage of this api key to n characters per month where n is the chosen value. Requests that incur charges will fail after reaching this monthly limit. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys/{jsonable_encoder(api_key_id)}", method="PATCH", json={ "is_enabled": is_enabled, "name": name, "permissions": convert_and_respect_annotation_metadata( object_=permissions, annotation=BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions, direction="write", ), "character_limit": character_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if _response is None or not _response.text.strip(): return HttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # 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 AsyncRawApiKeysClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, service_account_user_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[WorkspaceApiKeyListResponseModel]: """ Get all API keys for a service account Parameters ---------- service_account_user_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceApiKeyListResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceApiKeyListResponseModel, construct_type( type_=WorkspaceApiKeyListResponseModel, # 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( self, service_account_user_id: str, *, name: str, permissions: BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions, character_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[WorkspaceCreateApiKeyResponseModel]: """ Create a new API key for a service account Parameters ---------- service_account_user_id : str name : str permissions : BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions The permissions of the XI API. character_limit : typing.Optional[int] The character limit of the XI API key. If provided this will limit the usage of this api key to n characters per month where n is the chosen value. Requests that incur charges will fail after reaching this monthly limit. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[WorkspaceCreateApiKeyResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys", method="POST", json={ "name": name, "permissions": convert_and_respect_annotation_metadata( object_=permissions, annotation=BodyCreateServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysPostPermissions, direction="write", ), "character_limit": character_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( WorkspaceCreateApiKeyResponseModel, construct_type( type_=WorkspaceCreateApiKeyResponseModel, # 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, service_account_user_id: str, api_key_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.Any]: """ Delete an existing API key for a service account Parameters ---------- service_account_user_id : str api_key_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.Any] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys/{jsonable_encoder(api_key_id)}", method="DELETE", request_options=request_options, ) try: if _response is None or not _response.text.strip(): return AsyncHttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # 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 update( self, service_account_user_id: str, api_key_id: str, *, is_enabled: bool, name: str, permissions: BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions, character_limit: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.Any]: """ Update an existing API key for a service account Parameters ---------- service_account_user_id : str api_key_id : str is_enabled : bool Whether to enable or disable the API key. name : str The name of the XI API key to use (used for identification purposes only). permissions : BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions The permissions of the XI API. character_limit : typing.Optional[int] The character limit of the XI API key. If provided this will limit the usage of this api key to n characters per month where n is the chosen value. Requests that incur charges will fail after reaching this monthly limit. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.Any] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/service-accounts/{jsonable_encoder(service_account_user_id)}/api-keys/{jsonable_encoder(api_key_id)}", method="PATCH", json={ "is_enabled": is_enabled, "name": name, "permissions": convert_and_respect_annotation_metadata( object_=permissions, annotation=BodyEditServiceAccountApiKeyV1ServiceAccountsServiceAccountUserIdApiKeysApiKeyIdPatchPermissions, direction="write", ), "character_limit": character_limit, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if _response is None or not _response.text.strip(): return AsyncHttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # 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)