# 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.unchecked_base_model import construct_type from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.http_validation_error import HttpValidationError from ...types.resource_metadata_response_model import ResourceMetadataResponseModel from ...types.workspace_resource_type import WorkspaceResourceType from .types.body_share_workspace_resource_v_1_workspace_resources_resource_id_share_post_role import ( BodyShareWorkspaceResourceV1WorkspaceResourcesResourceIdSharePostRole, ) # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawResourcesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get( self, resource_id: str, *, resource_type: WorkspaceResourceType, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ResourceMetadataResponseModel]: """ Gets the metadata of a resource by ID. Parameters ---------- resource_id : str The ID of the target resource. resource_type : WorkspaceResourceType Resource type of the target resource. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ResourceMetadataResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/workspace/resources/{jsonable_encoder(resource_id)}", method="GET", params={ "resource_type": resource_type, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ResourceMetadataResponseModel, construct_type( type_=ResourceMetadataResponseModel, # 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 share( self, resource_id: str, *, role: BodyShareWorkspaceResourceV1WorkspaceResourcesResourceIdSharePostRole, resource_type: WorkspaceResourceType, user_email: typing.Optional[str] = OMIT, group_id: typing.Optional[str] = OMIT, workspace_api_key_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.Any]: """ Grants a role on a workspace resource to a user or a group. It overrides any existing role this user/service account/group/workspace api key has on the resource. To target a user or service account, pass only the user email. The user must be in your workspace. To target a group, pass only the group id. To target a workspace api key, pass the api key id. The resource will be shared with the service account associated with the api key. You must have admin access to the resource to share it. Parameters ---------- resource_id : str The ID of the target resource. role : BodyShareWorkspaceResourceV1WorkspaceResourcesResourceIdSharePostRole Role to update the target principal with. resource_type : WorkspaceResourceType Resource type of the target resource. user_email : typing.Optional[str] The email of the user or service account. group_id : typing.Optional[str] The ID of the target group. To target the permissions principals have by default on this resource, use the value 'default'. workspace_api_key_id : typing.Optional[str] The ID of the target workspace API key. This isn't the same as the key itself that would you pass in the header for authentication. Workspace admins can find this in the workspace settings UI. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/workspace/resources/{jsonable_encoder(resource_id)}/share", method="POST", json={ "role": role, "resource_type": resource_type, "user_email": user_email, "group_id": group_id, "workspace_api_key_id": workspace_api_key_id, }, 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) def unshare( self, resource_id: str, *, resource_type: WorkspaceResourceType, user_email: typing.Optional[str] = OMIT, group_id: typing.Optional[str] = OMIT, workspace_api_key_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[typing.Any]: """ Removes any existing role on a workspace resource from a user, service account, group or workspace api key. To target a user or service account, pass only the user email. The user must be in your workspace. To target a group, pass only the group id. To target a workspace api key, pass the api key id. The resource will be unshared from the service account associated with the api key. You must have admin access to the resource to unshare it. You cannot remove permissions from the user who created the resource. Parameters ---------- resource_id : str The ID of the target resource. resource_type : WorkspaceResourceType Resource type of the target resource. user_email : typing.Optional[str] The email of the user or service account. group_id : typing.Optional[str] The ID of the target group. To target the permissions principals have by default on this resource, use the value 'default'. workspace_api_key_id : typing.Optional[str] The ID of the target workspace API key. This isn't the same as the key itself that would you pass in the header for authentication. Workspace admins can find this in the workspace settings UI. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/workspace/resources/{jsonable_encoder(resource_id)}/unshare", method="POST", json={ "resource_type": resource_type, "user_email": user_email, "group_id": group_id, "workspace_api_key_id": workspace_api_key_id, }, 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 AsyncRawResourcesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get( self, resource_id: str, *, resource_type: WorkspaceResourceType, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ResourceMetadataResponseModel]: """ Gets the metadata of a resource by ID. Parameters ---------- resource_id : str The ID of the target resource. resource_type : WorkspaceResourceType Resource type of the target resource. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ResourceMetadataResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/workspace/resources/{jsonable_encoder(resource_id)}", method="GET", params={ "resource_type": resource_type, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ResourceMetadataResponseModel, construct_type( type_=ResourceMetadataResponseModel, # 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 share( self, resource_id: str, *, role: BodyShareWorkspaceResourceV1WorkspaceResourcesResourceIdSharePostRole, resource_type: WorkspaceResourceType, user_email: typing.Optional[str] = OMIT, group_id: typing.Optional[str] = OMIT, workspace_api_key_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.Any]: """ Grants a role on a workspace resource to a user or a group. It overrides any existing role this user/service account/group/workspace api key has on the resource. To target a user or service account, pass only the user email. The user must be in your workspace. To target a group, pass only the group id. To target a workspace api key, pass the api key id. The resource will be shared with the service account associated with the api key. You must have admin access to the resource to share it. Parameters ---------- resource_id : str The ID of the target resource. role : BodyShareWorkspaceResourceV1WorkspaceResourcesResourceIdSharePostRole Role to update the target principal with. resource_type : WorkspaceResourceType Resource type of the target resource. user_email : typing.Optional[str] The email of the user or service account. group_id : typing.Optional[str] The ID of the target group. To target the permissions principals have by default on this resource, use the value 'default'. workspace_api_key_id : typing.Optional[str] The ID of the target workspace API key. This isn't the same as the key itself that would you pass in the header for authentication. Workspace admins can find this in the workspace settings UI. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.Any] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/workspace/resources/{jsonable_encoder(resource_id)}/share", method="POST", json={ "role": role, "resource_type": resource_type, "user_email": user_email, "group_id": group_id, "workspace_api_key_id": workspace_api_key_id, }, 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) async def unshare( self, resource_id: str, *, resource_type: WorkspaceResourceType, user_email: typing.Optional[str] = OMIT, group_id: typing.Optional[str] = OMIT, workspace_api_key_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[typing.Any]: """ Removes any existing role on a workspace resource from a user, service account, group or workspace api key. To target a user or service account, pass only the user email. The user must be in your workspace. To target a group, pass only the group id. To target a workspace api key, pass the api key id. The resource will be unshared from the service account associated with the api key. You must have admin access to the resource to unshare it. You cannot remove permissions from the user who created the resource. Parameters ---------- resource_id : str The ID of the target resource. resource_type : WorkspaceResourceType Resource type of the target resource. user_email : typing.Optional[str] The email of the user or service account. group_id : typing.Optional[str] The ID of the target group. To target the permissions principals have by default on this resource, use the value 'default'. workspace_api_key_id : typing.Optional[str] The ID of the target workspace API key. This isn't the same as the key itself that would you pass in the header for authentication. Workspace admins can find this in the workspace settings UI. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.Any] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/workspace/resources/{jsonable_encoder(resource_id)}/unshare", method="POST", json={ "resource_type": resource_type, "user_email": user_email, "group_id": group_id, "workspace_api_key_id": workspace_api_key_id, }, 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)