77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
from DataVaultGenerator.Components import ErrorCollection, GeneratorEntity, DBEntity
|
|
|
|
|
|
class GenericTransformation(GeneratorEntity):
|
|
def __init__(self, model, filename, definition: dict = None):
|
|
GeneratorEntity.__init__(self, model, filename, definition)
|
|
|
|
self._layername = definition.get('layer', self.model.config.entitydefaults[self.type]['layer'])
|
|
|
|
@property
|
|
def dbentity(self):
|
|
return DBEntity(self.name,
|
|
self,
|
|
self.model.config.layer.get(self._layername).get('defaultdatabaseobject'),
|
|
self._definition.get('dbentity'))
|
|
|
|
@property
|
|
def layer(self):
|
|
"""Returns the entity layer."""
|
|
return self.model.get_layer(self._layername)
|
|
|
|
def validate(self):
|
|
|
|
errors = ErrorCollection()
|
|
|
|
# Validating entity references:
|
|
|
|
for e in self._definition.get('sources'):
|
|
if self.model.get_entity(e) is None:
|
|
errors.add("VALIDATION ERROR",
|
|
(self.filename, "Generic Transformation", "<" + self.name + ">"),
|
|
f'source <{e}> not found')
|
|
|
|
for e in self._definition.get('targets'):
|
|
if self.model.get_entity(e) is None:
|
|
errors.add("VALIDATION ERROR",
|
|
(self.filename, "Generic Transformation", "<" + self.name + ">"),
|
|
f'target <{e}> not found')
|
|
|
|
for name, e in self.get_query_entities().items():
|
|
if e is None:
|
|
errors.add("VALIDATION ERROR",
|
|
(self.filename, "Generic Transformation", "<" + self.name + ">"),
|
|
f'query-entity <{name}> not found.')
|
|
|
|
for name, e in self.get_query_entities().items():
|
|
if name not in self._definition.get('sources') and name not in self._definition.get('targets'):
|
|
errors.add("VALIDATION ERROR",
|
|
(self.filename, "Generic Transformation", "<" + self.name + ">"),
|
|
f'query-entity <{name}> not specified as source or target.')
|
|
|
|
return errors
|
|
|
|
def get_source_entities(self):
|
|
"""returns a list of linked entities."""
|
|
return [self.model.get_entity(h) for h in self._definition['sources']]
|
|
|
|
def get_target_entities(self):
|
|
"""returns a list of linked entities."""
|
|
return [self.model.get_entity(h) for h in self._definition['targets']]
|
|
|
|
@property
|
|
def query(self):
|
|
return self.model.get_parsed_query(self, self.rawquery)
|
|
|
|
def get_query_entities(self):
|
|
return self.model.get_query_entities(self.rawquery)
|
|
|
|
@property
|
|
def rawquery(self):
|
|
return self._definition.get('query', '')
|
|
|
|
def get_component_entities(self):
|
|
return [{'entity': self, 'component': c, 'type': c.type} for c in self.get_source_entities()]
|
|
|
|
def get_attributes(self, roles=(), exclude=()):
|
|
return [] |