# 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 ListAspectTypesRequest
from assetmanagement.models.field_type_enum import FieldTypeEnum
from mindsphere_core.exceptions import MindsphereClientError
[docs]class AspecttypeClient:
__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_aspect_type(self, request_object):
"""Delete aspect type
Delete an aspect type. Aspect type can only be deleted if there is no asset type using it.
:param DeleteAspectTypeRequest 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('AspecttypeClient.delete_aspect_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `delete_aspect_type`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `delete_aspect_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `delete_aspect_type`')
end_point_url = '/aspecttypes/{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('AspecttypeClient.delete_aspect_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 get_aspect_type(self, request_object):
"""Read an aspect type
Read an aspect type.
:param GetAspectTypeRequest 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 )
:return: AspectTypeResource
"""
logger.info('AspecttypeClient.get_aspect_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `get_aspect_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `get_aspect_type`')
end_point_url = '/aspecttypes/{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('AspecttypeClient.get_aspect_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, 'AspectTypeResource', self.__model_package__)
[docs] def list_aspect_types(self, request_object):
"""List all aspect types
List all aspect types
:param ListAspectTypesRequest 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: AspectTypeListResource
"""
logger.info('AspecttypeClient.list_aspect_types() invoked.')
end_point_url = '/aspecttypes'
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('AspecttypeClient.list_aspect_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, 'AspectTypeListResource', self.__model_package__)
[docs] def save_aspect_type(self, request_object):
"""Create or Update an aspect type
Create or Update an aspect type. Only adding variables supported. User can increase the length of a static STRING variable. The length cannot be decreased. The length of a dynamic STRING variable cannot be changed.
:param SaveAspectTypeRequest 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| ( aspecttype* - aspect 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 )
:return: AspectTypeResource
"""
logger.info('AspecttypeClient.save_aspect_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `save_aspect_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `save_aspect_type`')
if request_object.aspecttype is None:
raise exceptions.MindsphereClientError('The required parameter `aspecttype` is missing from `request_object`, when calling `save_aspect_type`')
end_point_url = '/aspecttypes/{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 = {}
form_params, local_var_files, body_params = {}, {}, request_object.aspecttype
logger.info('AspecttypeClient.save_aspect_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, 'AspectTypeResource', self.__model_package__)
[docs] def update_aspect_type(self, request_object):
"""Patch an aspect type
Patch an aspect type. Only adding variables supported. Patching requires the inclusion of already existing variables. Other fields may be omitted. Conforms to RFC 7396 - JSON merge Patch.
:param UpdateAspectTypeRequest 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| ( aspecttype* - aspect type )
:return: AspectTypeResource
"""
logger.info('AspecttypeClient.update_aspect_type() invoked.')
if request_object is None:
raise exceptions.MindsphereClientError('`request_object` is not passed when calling `update_aspect_type`')
if request_object.if_match is None:
raise exceptions.MindsphereClientError('The required parameter `If-Match` is missing from `request_object`, when calling `update_aspect_type`')
if request_object.id is None:
raise exceptions.MindsphereClientError('The required parameter `id` is missing from `request_object`, when calling `update_aspect_type`')
if request_object.aspecttype is None:
raise exceptions.MindsphereClientError('The required parameter `aspecttype` is missing from `request_object`, when calling `update_aspect_type`')
end_point_url = '/aspecttypes/{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.aspecttype
logger.info('AspecttypeClient.update_aspect_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, 'AspectTypeResource', self.__model_package__)
[docs] def get_aspect_types_equals_to(self, field_type, filter_value):
"""Filter aspect types by name or tenantId.
Returns aspect types 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 aspect types need to be filtered.
:param str filter_value: Value based on which aspect types need to be filtered.
:return: AspectTypeListResource 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 = ListAspectTypesRequest(filter=[data])
return self.list_aspect_types(request_object)
[docs] def get_aspect_types_like(self, field_type, *filter_value):
"""Filter aspect types by name or tenantId.
Returns aspect types 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 aspect types need to be filtered.
:param str filter_value: Value based on which aspect types need to be filtered.
:return: AspectTypeListResource 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 = ListAspectTypesRequest(filter=[data])
return self.list_aspect_types(request_object)
[docs] def get_aspect_types_starts_with(self, field_type, filter_value):
""" Filter aspect types by name or tenantId.
Returns aspect types 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 aspect types need to be filtered.
:param stt filter_value: Value based on which aspect types need to be filtered.
:return: AspectTypeListResource 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 = ListAspectTypesRequest(filter=[data])
return self.list_aspect_types(request_object)
[docs] def get_aspect_types_ends_with(self, field_type, filter_value):
"""Filter aspect types by name or tenantId.
Returns aspect types 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 aspect types need to be filtered.
:param str filter_value: Value based on which aspect types need to be filtered.
:return: AspectTypeListResource 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 = ListAspectTypesRequest(filter=[data])
return self.list_aspect_types(request_object)
[docs] def get_aspect_types_contains(self, field_type, filter_value):
"""Filter aspect types by name or tenantId.
Returns aspect types 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 aspect types need to be filtered.
:param str filter_value: Value based on which aspect types need to be filtered.
:return: AspectTypeListResource 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 = ListAspectTypesRequest(filter=[data])
return self.list_aspect_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
else:
raise MindsphereClientError(
str("Unsupported filter field type: {0}").format(field_type),
http_status=400,
)