Source code for integrateddatalake.clients.object_operations_with_access_token_client

# coding: utf-8

"""
    Data Lake API

    Service for storing Objects, download objects, add extended metadata tags,  subscribe for notifications, import tenant specific MindSphere Time Series data, and enable data access using cross account access and STS in Integrated MindSphere Data Lake. - Objects reside in user specified path like below example. Extension in object name is optional. Root path can be denoted as \"/\"       /basefolder/subfolder/objectname.objectext   - The following generic error codes might occur at any of the specified operations. Generic errors are prefixed with 'mdsp.core.generic.'.    - missingParameter   - invalidParameter   - missingRequestBodyProperty   - invalidRequestBodyProperty   - unauthorized   - forbidden   - noMatch   - unsupportedMediaType   - tooManyRequests   - internalServerError  # noqa: E501
"""


from __future__ import absolute_import

from mindsphere_core.mindsphere_core import logger
from mindsphere_core import mindsphere_core, exceptions, token_service
from mindsphere_core.token_service import init_credentials


[docs]class ObjectOperationsWithAccessTokenClient: __base_path__ = "/api/datalake/v3" __model_package__ = __name__.split(".")[0] def __init__(self, rest_client_config=None, mindsphere_credentials=None): self.rest_client_config = rest_client_config self.mindsphere_credentials = init_credentials(mindsphere_credentials)
[docs] def access_token_permissions(self, request_object): """Allows to give write premission on folder/path Allows users (tenant) to give write permission on folder/path. This API can only be accessed by tenant admin. Tenant admin can also provide write permission on subtenant folder. Write permission on root folder can also be given, except TSI folder :param AccessTokenPermissionsRequest request_object: It contains the below parameters --> |br| ( writePathPayload - Optional subtenantId (if path for write permission belongs to the tenant) ) :return: AccessTokenPermissionResource """ logger.info( "ObjectOperationsWithAccessTokenClient.access_token_permissions() invoked." ) end_point_url = "/accessTokenPermissions" end_point_url = end_point_url.format() token = token_service.fetch_token( self.rest_client_config, self.mindsphere_credentials ) api_url = mindsphere_core.build_url( self.__base_path__, end_point_url, self.rest_client_config ) headers = { "Accept": "application/json", "Authorization": "Bearer " + str(token), } query_params = {} form_params, local_var_files, body_params = ( {}, {}, request_object.write_path_payload, ) logger.info( "ObjectOperationsWithAccessTokenClient.access_token_permissions() --> Proceeding for API Invoker." ) return mindsphere_core.invoke_service( self.rest_client_config, api_url, headers, "POST", query_params, form_params, body_params, local_var_files, "AccessTokenPermissionResource", self.__model_package__, )
[docs] def delete_access_token_permissions(self, request_object): """Delete write permission on folder for the given id Delete write permission on folder for the given id This API can be accessed by tenant admin only :param DeleteAccessTokenPermissionsRequest request_object: It contains the below parameters --> |br| ( id* - Unique identifier of the write enabled folders ) :return: None """ logger.info( "ObjectOperationsWithAccessTokenClient.delete_access_token_permissions() invoked." ) if request_object is None: raise exceptions.MindsphereClientError( "`request_object` is not passed when calling `delete_access_token_permissions`" ) if request_object.id is None: raise exceptions.MindsphereClientError( "The required parameter `id` is missing from `request_object`, when calling `delete_access_token_permissions`" ) end_point_url = "/accessTokenPermissions/{id}" end_point_url = end_point_url.format(id=request_object.id) token = token_service.fetch_token( self.rest_client_config, self.mindsphere_credentials ) api_url = mindsphere_core.build_url( self.__base_path__, end_point_url, self.rest_client_config ) headers = { "Accept": "application/json", "Authorization": "Bearer " + str(token), } query_params = {} form_params, local_var_files, body_params = {}, {}, None logger.info( "ObjectOperationsWithAccessTokenClient.delete_access_token_permissions() --> Proceeding for API Invoker." ) return mindsphere_core.invoke_service( self.rest_client_config, api_url, headers, "DELETE", query_params, form_params, body_params, local_var_files, None, self.__model_package__, )
[docs] def generate_access_token(self, request_object): """Generate AWS STS token Allows users to request temporary, limited-privilege AWS credentials to get read-only or write-only access on the URI returned in the response. * Read permission will always be on the root level. * Path field is optional for READ permission - If value for path is not provided then it will be considered on root level (\"/\"). * Ensure to enable write access on the path before requesting token with write permission. * Write access can be enabled using POST /accessTokenPermissions endpoint. * An access token requested for a given path also automatically gives access to all subpaths of the path. For example, if an access token is requested for path /a and there are subpaths /a/b and /a/b/c, the token allows to access those too. * An access token with write permissions can only be requested for the paths defined by resource accessTokenPermissions. An acecss token with read permissions can only be requested for the root path /. :param GenerateAccessTokenRequest request_object: It contains the below parameters --> |br| ( stsPayload - Optional subtenant ID (if tenant is performing action on behalf of subtenant) ) :return: AccessTokens """ logger.info( "ObjectOperationsWithAccessTokenClient.generate_access_token() invoked." ) end_point_url = "/generateAccessToken" end_point_url = end_point_url.format() token = token_service.fetch_token( self.rest_client_config, self.mindsphere_credentials ) api_url = mindsphere_core.build_url( self.__base_path__, end_point_url, self.rest_client_config ) headers = { "Accept": "application/json", "Authorization": "Bearer " + str(token), } query_params = {} form_params, local_var_files, body_params = {}, {}, request_object.sts_payload logger.info( "ObjectOperationsWithAccessTokenClient.generate_access_token() --> Proceeding for API Invoker." ) return mindsphere_core.invoke_service( self.rest_client_config, api_url, headers, "POST", query_params, form_params, body_params, local_var_files, "AccessTokens", self.__model_package__, )
[docs] def get_access_token_permissions(self, request_object): """Details of the write folder request for the given id Details of the write folder request for the given id This API can be accessed by tenant admin, to get details of the request including for subtenants. Subtenant can access this API, to get details of the request belongs to their write folder. :param GetAccessTokenPermissionsRequest request_object: It contains the below parameters --> |br| ( id* - Unique identifier of the write enabled folders ) :return: AccessTokenPermissionResource """ logger.info( "ObjectOperationsWithAccessTokenClient.get_access_token_permissions() invoked." ) if request_object is None: raise exceptions.MindsphereClientError( "`request_object` is not passed when calling `get_access_token_permissions`" ) if request_object.id is None: raise exceptions.MindsphereClientError( "The required parameter `id` is missing from `request_object`, when calling `get_access_token_permissions`" ) end_point_url = "/accessTokenPermissions/{id}" end_point_url = end_point_url.format(id=request_object.id) token = token_service.fetch_token( self.rest_client_config, self.mindsphere_credentials ) api_url = mindsphere_core.build_url( self.__base_path__, end_point_url, self.rest_client_config ) headers = { "Accept": "application/json", "Authorization": "Bearer " + str(token), } query_params = {} form_params, local_var_files, body_params = {}, {}, None logger.info( "ObjectOperationsWithAccessTokenClient.get_access_token_permissions() --> Proceeding for API Invoker." ) return mindsphere_core.invoke_service( self.rest_client_config, api_url, headers, "GET", query_params, form_params, body_params, local_var_files, "AccessTokenPermissionResource", self.__model_package__, )
[docs] def list_access_token_permissions(self, request_object): """List all folders having write premission List all folders having write permission. This API can be accessed by tenant admin, to list all write permission folders including of subtenants. Subtenant can access this API, to get list write permission folders owned by subtenant. <h3>Limitations</h3> * <h4>Size parameter value should not be more than 1000.</h4> :param ListAccessTokenPermissionsRequest request_object: It contains the below parameters --> |br| ( page - Specifies the requested page index ), |br| ( size - Specifies the number of elements in a page ) :return: AccessTokenPermissionResources """ logger.info( "ObjectOperationsWithAccessTokenClient.list_access_token_permissions() invoked." ) end_point_url = "/accessTokenPermissions" end_point_url = end_point_url.format() token = token_service.fetch_token( self.rest_client_config, self.mindsphere_credentials ) api_url = mindsphere_core.build_url( self.__base_path__, end_point_url, self.rest_client_config ) headers = { "Accept": "application/json", "Authorization": "Bearer " + str(token), } query_params = {"page": request_object.page, "size": request_object.size} form_params, local_var_files, body_params = {}, {}, None logger.info( "ObjectOperationsWithAccessTokenClient.list_access_token_permissions() --> Proceeding for API Invoker." ) return mindsphere_core.invoke_service( self.rest_client_config, api_url, headers, "GET", query_params, form_params, body_params, local_var_files, "AccessTokenPermissionResources", self.__model_package__, )