Source code for sdi.models.input_mapping
# coding: utf-8
"""
SDI - Semantic Data Interconnect APIs
The Semantic Data Interconnect (SDI) is a collection of APIs that allows the user to unlock the potential of disparate big data by connecting external data. The SDI can infer the schemas of data based on schema-on-read, allow creating a semantic model and perform big data semantic queries. It seamlessly connects to MindSphere's Integrated Data Lake (IDL), but it can work independently as well. There are two mechanisms that can be used to upload files so that SDI can generate schemas and make data ready for query. The SDI operations are divided into the following groups: **Data Registration for SDI** This set of APIs is used to organize the incoming data. When configuring a Data Registry, you have the option to update your data based on a replace or append strategy. If you consider a use case where schema may change and incoming data files are completely different every time then replace is a good strategy. The replace strategy will replace the existing schema and data during each data ingest operation whereas the append strategy will update the existing schema and data during each data ingest operation. **Custom Data Type for SDI** The SDI by default identifies basic data types for each property, such as String, Integer, Float, Date, etc. The user can use this set of APIs to create their own custom data type. The SDI also provides an API under this category to suggest data type based on user-provided sample test values. **Data Lake for SDI** The SDI can process files uploaded provides endpoints to manage customer's data lake registration based on tenant id, cloud provider and data lake type. The set of REST endpoint allows to create, update and retrieve base path for their data lake. The IDL customer needs to create an SDI folder that is under the root folder. Any file uploaded in this folder is automatically picked up by SDI to process via IDL notification. **Data Ingest for SDI** This set of APIs allows user to upload files, start an ingest job for uploaded files, find job status for ingested jobs or retrieve all job statuses. **Schema Registry for SDI** The SDI provides a way to find the generated schema in this category. Users can find an SDI generated schema for uploaded files based on source name, data tag or schema name. **Data Query for SDI** allows querying based on the extracted schemas. Important supported APIs are: * Query interface for querying semantically correlated and transformed data. * Stores and executes data queries. * Uses a semantic model to translate model-based query to physical queries. **Semantic Model for SDI** allows user to create semantic model ontologies based on the extracted one or more schemas. The important functionalities achieved with APIs are: * Contextual correlation of data from different systems. * Infers & Recommends mappings between different schemas. * Import and store Semantic model. # noqa: E501
"""
import pprint
import re
import six
from sdi.models.input_mapping_class_property import InputMappingClassProperty
from sdi.models.input_mapping_schema_property import InputMappingSchemaProperty
from sdi.models.mapping_function import MappingFunction
from mindsphere_core.exceptions import MindsphereClientError
[docs]class InputMapping(object):
"""
Attributes:
attribute_types (dict): The key is attribute name
and the value is attribute type.
attribute_map (dict): The key is attribute name
and the value is json key in definition.
"""
attribute_types = {
'class_property': 'InputMappingClassProperty',
'description': 'str',
'functional_mapping': 'bool',
'key_mapping': 'bool',
'mapping_function': 'MappingFunction',
'name': 'str',
'schema_properties': 'list[InputMappingSchemaProperty]'
}
attribute_map = {
'class_property': 'classProperty',
'description': 'description',
'functional_mapping': 'functionalMapping',
'key_mapping': 'keyMapping',
'mapping_function': 'mappingFunction',
'name': 'name',
'schema_properties': 'schemaProperties'
}
def __init__(self, class_property=None, description=None, functional_mapping=None, key_mapping=None, mapping_function=None, name=None, schema_properties=None):
self._class_property = class_property
self._description = description
self._functional_mapping = functional_mapping
self._key_mapping = key_mapping
self._mapping_function = mapping_function
self._name = name
self._schema_properties = schema_properties
self.discriminator = None
@property
def class_property(self):
"""Gets the class_property of this InputMapping.
:return: The class_property of this InputMapping.
:rtype: InputMappingClassProperty
"""
return self._class_property
@class_property.setter
def class_property(self, class_property):
"""Sets the class_property of this InputMapping.
:param class_property: The class_property of this InputMapping.
:type: InputMappingClassProperty
"""
self._class_property = class_property
@property
def description(self):
"""Gets the description of this InputMapping.
:return: The description of this InputMapping.
:rtype: str
"""
return self._description
@description.setter
def description(self, description):
"""Sets the description of this InputMapping.
:param description: The description of this InputMapping.
:type: str
"""
self._description = description
@property
def functional_mapping(self):
"""Gets the functional_mapping of this InputMapping.
:return: The functional_mapping of this InputMapping.
:rtype: bool
"""
return self._functional_mapping
@functional_mapping.setter
def functional_mapping(self, functional_mapping):
"""Sets the functional_mapping of this InputMapping.
:param functional_mapping: The functional_mapping of this InputMapping.
:type: bool
"""
self._functional_mapping = functional_mapping
@property
def key_mapping(self):
"""Gets the key_mapping of this InputMapping.
:return: The key_mapping of this InputMapping.
:rtype: bool
"""
return self._key_mapping
@key_mapping.setter
def key_mapping(self, key_mapping):
"""Sets the key_mapping of this InputMapping.
:param key_mapping: The key_mapping of this InputMapping.
:type: bool
"""
self._key_mapping = key_mapping
@property
def mapping_function(self):
"""Gets the mapping_function of this InputMapping.
:return: The mapping_function of this InputMapping.
:rtype: MappingFunction
"""
return self._mapping_function
@mapping_function.setter
def mapping_function(self, mapping_function):
"""Sets the mapping_function of this InputMapping.
:param mapping_function: The mapping_function of this InputMapping.
:type: MappingFunction
"""
self._mapping_function = mapping_function
@property
def name(self):
"""Gets the name of this InputMapping.
:return: The name of this InputMapping.
:rtype: str
"""
return self._name
@name.setter
def name(self, name):
"""Sets the name of this InputMapping.
:param name: The name of this InputMapping.
:type: str
"""
self._name = name
@property
def schema_properties(self):
"""Gets the schema_properties of this InputMapping.
:return: The schema_properties of this InputMapping.
:rtype: list[InputMappingSchemaProperty]
"""
return self._schema_properties
@schema_properties.setter
def schema_properties(self, schema_properties):
"""Sets the schema_properties of this InputMapping.
:param schema_properties: The schema_properties of this InputMapping.
:type: list[InputMappingSchemaProperty]
"""
self._schema_properties = schema_properties
[docs] def to_dict(self):
"""Returns the model properties as a dict"""
result = {}
for attr, _ in six.iteritems(self.attribute_types):
value = getattr(self, attr)
if isinstance(value, list):
result[attr] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
value
))
elif hasattr(value, "to_dict"):
result[attr] = value.to_dict()
elif isinstance(value, dict):
result[attr] = dict(map(
lambda item: (item[0], item[1].to_dict())
if hasattr(item[1], "to_dict") else item,
value.items()
))
else:
result[attr] = value
if issubclass(InputMapping, dict):
for key, value in self.items():
result[key] = value
return result
[docs] def to_str(self):
"""Returns the string representation of the model"""
return pprint.pformat(self.to_dict())
def __repr__(self):
"""For `print` and `pprint`"""
return self.to_str()
def __eq__(self, other):
"""Returns true if both objects are equal"""
if not isinstance(other, InputMapping):
return False
return self.__dict__ == other.__dict__
def __ne__(self, other):
"""Returns true if both objects are not equal"""
return not self == other