# coding: utf-8
"""
Asset Management API
Service for configuring, reading and managing assets, asset ~ and aspect types. # noqa: E501
"""
from __future__ import absolute_import
import json
import time
from mindsphere_core.mindsphere_core import logger
from mindsphere_core import mindsphere_core, exceptions, token_service
from mindsphere_core.token_service import init_credentials
from assetmanagement.models import ListAssetsRequest, DeleteAssetRequest, GetAssetRequest
from assetmanagement.models.field_type_enum import FieldTypeEnum
from mindsphere_core.exceptions import MindsphereClientError
[docs]class AssetsClient:
__base_path__ = '/api/assetmanagement/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 add_asset(self, request_object):
"""Create an asset
Creates a new asset with the provided content. Only instantiable types could be used.
:param AddAssetRequest request_object: It contains the below parameters --> |br| ( asset* - asset )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.add_asset() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `add_asset`')
if request_object.asset is None:
raise exceptions.MindsphereClientError('The required parameter `asset` is missing from `request_object`, when calling `add_asset`')
end_point_url = '/assets'
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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, request_object.asset
logger.info('AssetsClient.add_asset() --> 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, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def delete_asset(self, request_object):
"""Delete an asset
Deletes the given asset. After deletion only users with admin role can read it, but modification is not possible anymore. It's not possible to delete an asset if it has children.
:param DeleteAssetRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - Unique identifier )
:return: None
"""
logger.info('AssetsClient.delete_asset() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `delete_asset`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `delete_asset`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `delete_asset`')
end_point_url = '/assets/{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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-Match': request_object.if_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssetsClient.delete_asset() --> 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 delete_asset_file_assigment(self, request_object):
"""Deletes a file assignment from an asset
Deletes a file assignment from an asset. If the asset's parent type has defined a file with the same key, the key will be displayed with the inherited value.
:param DeleteAssetFileAssigmentRequest request_object: It contains the below parameters --> |br| ( id* - Unique identifier ), |br| ( key* - Keyword for the file to be assigned to an asset or asset type. ), |br| ( If-Match* - Last known version to facilitate optimistic locking )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.delete_asset_file_assigment() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `delete_asset_file_assigment`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `delete_asset_file_assigment`')
if request_object.key is None:
raise exceptions.MindsphereClientError('The required parameter `key` is missing from `request_object`, when calling `delete_asset_file_assigment`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `delete_asset_file_assigment`')
end_point_url = '/assets/{id}/fileAssignments/{key}'
end_point_url = end_point_url.format(id=request_object.id, key=request_object.key)
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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-Match': request_object.if_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssetsClient.delete_asset_file_assigment() --> 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, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def get_asset(self, request_object):
"""Returns an asset.
Read a single asset. All static properties of asset are returned.
:param GetAssetRequest request_object: It contains the below parameters --> |br| ( id* - Unique identifier ), |br| ( If-None-Match - ETag hash of previous request to allow caching )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.get_asset() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `get_asset`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `get_asset`')
end_point_url = '/assets/{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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-None-Match': request_object.if_none_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssetsClient.get_asset() --> 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, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def get_root_asset(self, request_object):
"""Returns the root asset of the user.
Read the root asset of the user, from which the whole asset hierarchy can be rebuilt.
:param GetRootAssetRequest request_object: It contains the below parameters --> |br| ( If-None-Match - ETag hash of previous request to allow caching )
:return: RootAssetResource
"""
logger.info('AssetsClient.get_root_asset() invoked.')
end_point_url = '/assets/root'
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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-None-Match': request_object.if_none_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssetsClient.get_root_asset() --> 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, 'RootAssetResource', self.__model_package__)
[docs] def list_assets(self, request_object):
"""List all available assets
List all assets available for the authenticated user. ## Filter Supports all basic fields and the 'hasType' filter which search for the assets with type that originated from the given asset type. # Examples Example: This filter searches for assets which type is exactly the given type: ``` { \"typeId\": \"mandal.ship\" } ``` Beyond the basic fields we can search for assets which type is inherited from the given type: ``` { \"hasType\": \"mandal.ship\" } ```
:param ListAssetsRequest request_object: It contains the below parameters --> |br| ( page - Specifies the requested page index ), |br| ( size - Specifies the number of elements in a page ), |br| ( sort - Specifies the ordering of returned elements ), |br| ( filter - Specifies the additional filtering criteria ), |br| ( If-None-Match - ETag hash of previous request to allow caching )
:return: AssetListResource
"""
logger.info('AssetsClient.list_assets() invoked.')
end_point_url = '/assets'
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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-None-Match': request_object.if_none_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {'page': request_object.page, 'size': request_object.size, 'sort': request_object.sort, 'filter': request_object.filter}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssetsClient.list_assets() --> 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, 'AssetListResource', self.__model_package__)
[docs] def move_asset(self, request_object):
"""Move an asset
Moves an asset (and all it's children) in the instance hierarchy
:param MoveAssetRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - Unique identifier ), |br| ( moveParameters* - parameters for the asset move )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.move_asset() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `move_asset`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `move_asset`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `move_asset`')
if request_object.move_parameters is None:
raise exceptions.MindsphereClientError('The required parameter `moveParameters` is missing from `request_object`, when calling `move_asset`')
end_point_url = '/assets/{id}/move'
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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-Match': request_object.if_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, request_object.move_parameters
logger.info('AssetsClient.move_asset() --> 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, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def replace_asset(self, request_object):
"""Update an asset
Updates an asset with the provided content. Only values can be modified, asset's structure have to be modified in asset's type
:param ReplaceAssetRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - Unique identifier ), |br| ( asset* - asset )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.replace_asset() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `replace_asset`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `replace_asset`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `replace_asset`')
if request_object.asset is None:
raise exceptions.MindsphereClientError('The required parameter `asset` is missing from `request_object`, when calling `replace_asset`')
end_point_url = '/assets/{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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-Match': request_object.if_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, request_object.asset
logger.info('AssetsClient.replace_asset() --> Proceeding for API Invoker.')
return mindsphere_core.invoke_service(self.rest_client_config, api_url, headers, 'PUT', query_params, form_params, body_params, local_var_files, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def save_asset_file_assignment(self, request_object):
"""Save an file assignment to an asset
Save a file assignment to a given asset
:param SaveAssetFileAssignmentRequest request_object: It contains the below parameters --> |br| ( id* - Unique identifier ), |br| ( key* - Keyword for the file to be assigned to an asset or asset type. ), |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( assignment* - Data for file assignment )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.save_asset_file_assignment() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `save_asset_file_assignment`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `save_asset_file_assignment`')
if request_object.key is None:
raise exceptions.MindsphereClientError('The required parameter `key` is missing from `request_object`, when calling `save_asset_file_assignment`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `save_asset_file_assignment`')
if request_object.assignment is None:
raise exceptions.MindsphereClientError('The required parameter `assignment` is missing from `request_object`, when calling `save_asset_file_assignment`')
end_point_url = '/assets/{id}/fileAssignments/{key}'
end_point_url = end_point_url.format(id=request_object.id, key=request_object.key)
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/hal+json, application/vnd.error+json', 'Content-Type': 'application/json', 'If-Match': request_object.if_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, request_object.assignment
logger.info('AssetsClient.save_asset_file_assignment() --> Proceeding for API Invoker.')
return mindsphere_core.invoke_service(self.rest_client_config, api_url, headers, 'PUT', query_params, form_params, body_params, local_var_files, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def update_asset(self, request_object):
"""Patch an asset
Patch an asset with the provided content. Only values can be modified, asset's structure have to be modified in asset's type. Conforms to RFC 7396 - JSON merge Patch.
:param UpdateAssetRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - Unique identifier ), |br| ( asset* - asset )
:return: AssetResourceWithHierarchyPath
"""
logger.info('AssetsClient.update_asset() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `update_asset`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `update_asset`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `update_asset`')
if request_object.asset is None:
raise exceptions.MindsphereClientError('The required parameter `asset` is missing from `request_object`, when calling `update_asset`')
end_point_url = '/assets/{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/hal+json, application/vnd.error+json', 'Content-Type': 'application/merge-patch+json', 'If-Match': request_object.if_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {}
form_params, local_var_files, body_params = {}, {}, request_object.asset
logger.info('AssetsClient.update_asset() --> Proceeding for API Invoker.')
return mindsphere_core.invoke_service(self.rest_client_config, api_url, headers, 'PATCH', query_params, form_params, body_params, local_var_files, 'AssetResourceWithHierarchyPath', self.__model_package__)
[docs] def delete_asset_with_confirmation(self, request_object):
"""delete an asset confiramtion
:param DeleteAssetConfirmationRequest request_object: Deletes an asset and returns a boolean which confirms deletion. It internally checks existence of the asset after we receive a confirmation from server. It retries 3 times for getting the confirmation from server. After deletion only users with admin role can read it, but modification is not possible anymore.It's not possible to delete an asset if it has children.
:return: bool
"""
__delay_time__ = 0.750
retry_counter = 0
is_asset_deletion_confirmed = False
request_object = DeleteAssetRequest(
if_match=request_object.if_match, id=request_object.id
)
self.delete_asset(request_object)
while retry_counter < 3 and is_asset_deletion_confirmed is not True:
try:
time.sleep(__delay_time__)
request_object = GetAssetRequest(id=request_object.id)
response = self.get_asset(request_object)
id_retrieved = response.asset_id
if id_retrieved == request_object.id:
logger.info("Failed to confirm deletion on trial: ")
retry_counter += 1
except Exception as error:
if (
hasattr(error, "message") and error.args[0].
message["errors"][0]["errorcode"] == "assetmngmnt.asset.notfound"):
logger.info(
"Deletion confirmed on trial: " + str(retry_counter + 1)
)
is_asset_deletion_confirmed = True
break
else:
break
return is_asset_deletion_confirmed
[docs] def get_assets_equals_to(self, field_type, filter_value):
"""Filter assets by name or assetId or tenantId or externalId or subTenant or parentId or typeId.
Returns assets whose name or tenantId equals to filterValue. Only one field type and value can be passed at a time.
:param FieldTypeEnum field_type: Field type on which assets need to be filtered.
:param str filter_value: Value based on which assets need to be filtered.
:return: AssetListResource of type JSON
"""
field_type = self.check_supported_field_types(field_type)
filter_input = {field_type: {"eq": filter_value}}
data = json.dumps(filter_input)
request_object = ListAssetsRequest(filter=[data])
return self.list_assets(request_object)
[docs] def get_assets_like(self, field_type, *filter_value):
"""Filter assets by name or assetId or tenantId or externalId or subTenant or parentId or typeId.
Returns assets whose name or tenantId matches an array of filter values. One field type and multiple filter values can be passed.
:param FieldTypeEnum field_type: Field type on which assets need to be filtered.
:param str filter_value: Variable array of values based on which assets need to be filtered.
:return: AssetListResource of type JSON
"""
field_type = self.check_supported_field_types(field_type)
filter_input = {field_type: {"in": {"value": filter_value}}}
data = json.dumps(filter_input)
request_object = ListAssetsRequest(filter=[data])
return self.list_assets(request_object)
[docs] def get_assets_starts_with(self, field_type, filter_value):
""" Filter assets by name or assetId or tenantId or externalId or subTenant or parentId or typeId.
Returns assets whose name or tenantId starts with the filter value. Only one field type and filter value can be passed.
:param FieldTypeEnum field_type: Field type on which assets need to be filtered.
:param str filter_value: Value based on which assets need to be filtered.
:return: AssetListResource of type JSON
"""
field_type = self.check_supported_field_types(field_type)
filter_input = {field_type: {"startsWith": filter_value}}
data = json.dumps(filter_input)
request_object = ListAssetsRequest(filter=[data])
return self.list_assets(request_object)
[docs] def get_assets_ends_with(self, field_type, filter_value):
""" Filter assets by name or assetId or tenantId or externalId or subTenant or parentId or typeId.
Returns assets whose name or tenantId ends with the filter value. Only one field type and filter value can be passed.
:param FieldTypeEnum field_type: Field type on which assets need to be filtered.
:param str filter_value: Value based on which assets need to be filtered.
:return: AssetListResource of type JSON
"""
field_type = self.check_supported_field_types(field_type)
filter_input = {field_type: {"endsWith": filter_value}}
data = json.dumps(filter_input)
request_object = ListAssetsRequest(filter=[data])
return self.list_assets(request_object)
[docs] def get_assets_contains(self, field_type, filter_value):
""" Filter assets by name or assetId or tenantId or externalId or subTenant or parentId or typeId.
Returns assets whose name or tenantId contains the filter value. Only one field type and filter value can be passed.
:param FieldTypeEnum field_type: Field type on which assets need to be filtered.
:param str filter_value: Value based on which assets need to be filtered.
:return: AssetListResource of type JSON
"""
field_type = self.check_supported_field_types(field_type)
filter_input = {field_type: {"contains": filter_value}}
data = json.dumps(filter_input)
request_object = ListAssetsRequest(filter=[data])
return self.list_assets(request_object)
[docs] def get_assets_of_type(self, filter):
"""Filter all Assets by asset type id.
Returns all assets of a particular asset type. Only one field type and filter value can be passed.
:param str filter: filter assets of a particular asset type.
:return: AssetListResource of type JSON
"""
filter_input = {"hasType": filter}
data = json.dumps(filter_input)
request_object = ListAssetsRequest(filter=[data])
return self.list_assets(request_object)
[docs] @staticmethod
def check_supported_field_types(field_type):
if field_type == FieldTypeEnum.NAME:
return FieldTypeEnum.NAME.value
elif field_type == FieldTypeEnum.TENANT_ID:
return FieldTypeEnum.TENANT_ID.value
elif field_type == FieldTypeEnum.ASSET_ID:
return FieldTypeEnum.ASSET_ID.value
elif field_type == FieldTypeEnum.DELETED:
return FieldTypeEnum.DELETED.value
elif field_type == FieldTypeEnum.EXTERNAL_ID:
return FieldTypeEnum.EXTERNAL_ID.value
elif field_type == FieldTypeEnum.PARENT_ID:
return FieldTypeEnum.PARENT_ID.value
elif field_type == FieldTypeEnum.SUBTENANT:
return FieldTypeEnum.SUBTENANT.value
elif field_type == FieldTypeEnum.TYPE_ID:
return FieldTypeEnum.TYPE_ID.value
else:
raise MindsphereClientError(
str("Unsupported filter field type: {0}").format(field_type),
http_status=400,
)