# 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
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 ListAssetTypesRequest
from assetmanagement.models.field_type_enum import FieldTypeEnum
from mindsphere_core.exceptions import MindsphereClientError
[docs]class AssettypeClient:
__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 delete_asset_type(self, request_object):
"""Delete an asset type
Deletes an asset type. Deletion only possible when the type has no child-type and there is no asset that instantiate it.
:param DeleteAssetTypeRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - The type's id is a unique identifier. The id's length must be between 1 and 128 characters and matches the following symbols \"A-Z\", \"a-z\", \"0-9\", \"_\" and \".\" beginning with the tenant prefix what has a maximum of 8 characters. (e.g . ten_pref.type_id) )
:return: None
"""
logger.info('AssettypeClient.delete_asset_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `delete_asset_type`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `delete_asset_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `delete_asset_type`')
end_point_url = '/assettypes/{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('AssettypeClient.delete_asset_type() --> 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_type_file_assignment(self, request_object):
"""Deletes a file assignment from an asset type
Deletes a file assignment from an asset type. If the type's parent has defined a file with the same key, the key will be displayed with the inherited value.
:param DeleteAssetTypeFileAssignmentRequest request_object: It contains the below parameters --> |br| ( id* - The type's id is a unique identifier. The id's length must be between 1 and 128 characters and matches the following symbols \"A-Z\", \"a-z\", \"0-9\", \"_\" and \".\" beginning with the tenant prefix what has a maximum of 8 characters. (e.g . ten_pref.type_id) ), |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: AssetTypeResource
"""
logger.info('AssettypeClient.delete_asset_type_file_assignment() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `delete_asset_type_file_assignment`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `delete_asset_type_file_assignment`')
if request_object.key is None:
raise exceptions.MindsphereClientError('The required parameter `key` is missing from `request_object`, when calling `delete_asset_type_file_assignment`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `delete_asset_type_file_assignment`')
end_point_url = '/assettypes/{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('AssettypeClient.delete_asset_type_file_assignment() --> 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, 'AssetTypeResource', self.__model_package__)
[docs] def get_asset_type(self, request_object):
"""Read an asset type
Read an asset type
:param GetAssetTypeRequest request_object: It contains the below parameters --> |br| ( id* - The type's id is a unique identifier. The id's length must be between 1 and 128 characters and matches the following symbols \"A-Z\", \"a-z\", \"0-9\", \"_\" and \".\" beginning with the tenant prefix what has a maximum of 8 characters. (e.g . ten_pref.type_id) ), |br| ( If-None-Match - ETag hash of previous request to allow caching ), |br| ( exploded - Specifies if the asset type should include all of it's inherited variables and aspects. Default is false. )
:return: AssetTypeResource
"""
logger.info('AssettypeClient.get_asset_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `get_asset_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `get_asset_type`')
end_point_url = '/assettypes/{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 = {'exploded': request_object.exploded}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssettypeClient.get_asset_type() --> 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, 'AssetTypeResource', self.__model_package__)
[docs] def list_asset_types(self, request_object):
"""List all asset types
List all asset types
:param ListAssetTypesRequest 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 ), |br| ( exploded - Specifies if the asset type should include all of it's inherited variables and aspects. Default is false. )
:return: AssetTypeListResource
"""
logger.info('AssettypeClient.list_asset_types() invoked.')
end_point_url = '/assettypes'
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, 'exploded': request_object.exploded}
form_params, local_var_files, body_params = {}, {}, None
logger.info('AssettypeClient.list_asset_types() --> 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, 'AssetTypeListResource', self.__model_package__)
[docs] def save_asset_type(self, request_object):
"""Create or Update an asset type
Create or Update an asset type User can increase the length of a STRING variable. The length cannot be decreased.
:param SaveAssetTypeRequest request_object: It contains the below parameters --> |br| ( id* - The type's id is a unique identifier. The id's length must be between 1 and 128 characters and matches the following symbols \"A-Z\", \"a-z\", \"0-9\", \"_\" and \".\" beginning with the tenant prefix what has a maximum of 8 characters. (e.g . ten_pref.type_id) ), |br| ( assettype* - asset type ), |br| ( If-Match - Last known version to facilitate optimistic locking. *Required for modification. ), |br| ( If-None-Match - Set ifNoneMatch header to \"*\" for ensuring create request ), |br| ( exploded - Specifies if the asset type should include all of it's inherited variables and aspects. Default is false. )
:return: AssetTypeResource
"""
logger.info('AssettypeClient.save_asset_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `save_asset_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `save_asset_type`')
if request_object.assettype is None:
raise exceptions.MindsphereClientError('The required parameter `assettype` is missing from `request_object`, when calling `save_asset_type`')
end_point_url = '/assettypes/{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, 'If-None-Match': request_object.if_none_match, 'Authorization': 'Bearer ' + str(token)}
query_params = {'exploded': request_object.exploded}
form_params, local_var_files, body_params = {}, {}, request_object.assettype
logger.info('AssettypeClient.save_asset_type() --> 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, 'AssetTypeResource', self.__model_package__)
[docs] def save_asset_type_file_assignment(self, request_object):
"""Add a new file assignment to an asset type
Add a new file assignment to a given asset type. All asset which extends these types will have its file by default.
:param SaveAssetTypeFileAssignmentRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - The type's id is a unique identifier. The id's length must be between 1 and 128 characters and matches the following symbols \"A-Z\", \"a-z\", \"0-9\", \"_\" and \".\" beginning with the tenant prefix what has a maximum of 8 characters. (e.g . ten_pref.type_id) ), |br| ( key* - Keyword for the file to be assigned to an asset or asset type. ), |br| ( assignment* - Data for file assignment )
:return: AssetTypeResource
"""
logger.info('AssettypeClient.save_asset_type_file_assignment() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `save_asset_type_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_type_file_assignment`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `save_asset_type_file_assignment`')
if request_object.key is None:
raise exceptions.MindsphereClientError('The required parameter `key` is missing from `request_object`, when calling `save_asset_type_file_assignment`')
if request_object.assignment is None:
raise exceptions.MindsphereClientError('The required parameter `assignment` is missing from `request_object`, when calling `save_asset_type_file_assignment`')
end_point_url = '/assettypes/{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('AssettypeClient.save_asset_type_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, 'AssetTypeResource', self.__model_package__)
[docs] def update_asset_type(self, request_object):
"""Patch an asset type
Patch an asset type. Patching requires the inclusion of all existing variables and aspects. Missing file assignments will be deleted. Other fields may be omitted. Conforms to RFC 7396 - JSON merge Patch.
:param UpdateAssetTypeRequest request_object: It contains the below parameters --> |br| ( If-Match* - Last known version to facilitate optimistic locking ), |br| ( id* - The type's id is a unique identifier. The id's length must be between 1 and 128 characters and matches the following symbols \"A-Z\", \"a-z\", \"0-9\", \"_\" and \".\" beginning with the tenant prefix what has a maximum of 8 characters. (e.g . ten_pref.type_id) ), |br| ( assettype* - asset type ), |br| ( exploded - Specifies if the asset type should include all of it's inherited variables and aspects. Default is false. )
:return: AssetTypeResource
"""
logger.info('AssettypeClient.update_asset_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `update_asset_type`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `update_asset_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `update_asset_type`')
if request_object.assettype is None:
raise exceptions.MindsphereClientError('The required parameter `assettype` is missing from `request_object`, when calling `update_asset_type`')
end_point_url = '/assettypes/{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 = {'exploded': request_object.exploded}
form_params, local_var_files, body_params = {}, {}, request_object.assettype
logger.info('AssettypeClient.update_asset_type() --> 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, 'AssetTypeResource', self.__model_package__)
[docs] def get_asset_types_equals_to(self, field_type, filter_value):
"""Filter asset types by name or tenantId or parentTypeId.
Returns asset types whose name or tenantId or parentTypeId equals to filterValue. Only one field type and value can be passed at a time.
:param FieldTypeEnum field_type: Field type on which asset types need to be filtered.
:param str filter_value: Value based on which asset types need to be filtered.
:return: AssetTypeListResource 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 = ListAssetTypesRequest(filter=[data])
return self.list_asset_types(request_object)
[docs] def get_asset_types_like(self, field_type, *filter_value):
"""Filter asset types by name or tenantId or parentTypeId.
Returns asset types whose name or tenantId or parentTypeId matches an array of filter values. One field type and multiple filter values can be passed.
:param FieldTypeEnum field_type: Field type on which asset types need to be filtered.
:param str filter_value: Variable array of values based on which asset types need to be filtered.
:return: AssetTypeListResource 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 = ListAssetTypesRequest(filter=[data])
return self.list_asset_types(request_object)
[docs] def get_asset_types_starts_with(self, field_type, filter_value):
""" Filter asset types by name or tenantId or parentTypeId.
Returns asset types whose name or tenantId or parentTypeId starts with the filter value. Only one field type and filter value can be passed.
:param FieldTypeEnum field_type: Field type on which asset types need to be filtered.
:param stt filter_value: Value based on which asset types need to be filtered.
:return: AssetTypeListResource 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 = ListAssetTypesRequest(filter=[data])
return self.list_asset_types(request_object)
[docs] def get_asset_types_ends_with(self, field_type, filter_value):
"""Filter asset types by name or tenantId or parentTypeId.
Returns asset types whose name or tenantId or parentTypeId ends with the filter value. Only one field type and filter value can be passed.
:param FieldTypeEnum field_type: Field type on which asset types need to be filtered.
:param str filter_value: Value based on which asset types need to be filtered.
:return: AssetTypeListResource 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 = ListAssetTypesRequest(filter=[data])
return self.list_asset_types(request_object)
[docs] def get_asset_types_contains(self, field_type, filter_value):
"""Filter asset types by name or tenantId or parentTypeId.
Returns asset types whose name or tenantId or parentTypeId contains the filter value. Only one field type and filter value can be passed.
:param FieldTypeEnum field_type: Field type on which asset types need to be filtered.
:param str filter_value: Value based on which asset types need to be filtered.
:return: AssetTypeListResource 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 = ListAssetTypesRequest(filter=[data])
return self.list_asset_types(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.PARENT_TYPE_ID:
return FieldTypeEnum.PARENT_TYPE_ID.value
else:
raise MindsphereClientError(
str("Unsupported filter field type: {0}").format(field_type),
http_status=400,
)