This repository has been archived on 2026-03-20. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files

105 lines
4.3 KiB
Python

import logging
from pathlib import Path
import sys
import yaml
class ConfigDict(dict):
def __getattr__(self, item):
return super().__getitem__(item)
def __setattr__(self, item, value):
return super().__setitem__(item, value)
def __dir__(self):
return super().__dir__() + [str(k) for k in self.keys()]
def __init__(self, initialvalues: dict = {}):
self.update(initialvalues)
class Config(ConfigDict):
def __dir__(self):
return super().__dir__() + [str(k) for k in self.keys()]
def __init__(self):
self['model'] = ConfigDict()
self['paths'] = ConfigDict()
self['datavault'] = ConfigDict()
self['generator'] = ConfigDict()
self['layer'] = ConfigDict()
self['vars'] = ConfigDict()
self['pre_hooks'] = ConfigDict()
self['post_hooks'] = ConfigDict()
self['commonattributes'] = ConfigDict()
self['entitydefaults'] = ConfigDict()
self['basetemplates'] = ConfigDict()
self['jinja'] = ConfigDict()
self['sys_specification'] = ConfigDict()
def load(self, filename, schema, validation_handler):
self.filename = filename
self.path = Path(filename).absolute().parent
try:
with open(filename, 'r') as file:
self.content = yaml.safe_load(file)
validation_handler('Configuration: ' + filename, schema, self.content)
except FileNotFoundError as e:
print("")
print(e)
logging.error(e)
sys.exit(2)
except yaml.scanner.ScannerError as e:
print("")
logging.error(e)
sys.exit(2)
except yaml.parser.ParserError as e:
print("")
logging.error(e)
sys.exit(2)
self.model.update(dict(name = self.content.get('model').get('name')
, ignore_file_prefix = self.content.get('model').get('ignore_file_prefix','_')))
self.paths.update(dict(log = self.content.get('model').get('paths').get('log')
, entities = self.content.get('model').get('paths').get('entities')
, mappings = self.content.get('model').get('paths').get('mappings')
, templates = self.content.get('model').get('paths').get('templates')
, output = self.content.get('model').get('paths').get('output')
))
self.vars.update(self.content.get('vars', {}))
self.pre_hooks.update(self.content.get('pre_hooks', {}))
self.post_hooks.update(self.content.get('post_hooks', {}))
self.datavault.update(dict(keyattribute = self.content.get('keyattribute')
, zerokey = self.content.get('zerokey','')
, constraints = ConfigDict(self.content.get('constraints', {}))
, hash = ConfigDict(dict(algorithm = self.content.get('hash_algorithm')
,separator = self.content.get('hash_separator')
,case = self.content.get('hash_case')))
, business_key_treatment = ConfigDict(self.content.get('business_key_treatment'))
, hashdiff_attribute_treatment = ConfigDict(self.content.get('hashdiff_attribute_treatment'))
, ghostrecord = ConfigDict(self.content.get('ghostrecord'))
))
self.commonattributes.update(self.content.get('commonattributes'))
self.basetemplates.update(self.content.get('templates'))
self.jinja.update(dict(environment = ConfigDict(self.content.get('jinja', {}).get('environment', {}))))
if self.content.get('sys_specification'):
self.sys_specification.update(self.content.get('sys_specification'))
for k,v in self.content.get('layer').items():
self.layer[k] = ConfigDict(v)
for k,v in self.content.get('entitydefaults').items():
self.entitydefaults[k] = ConfigDict(v)
for k,v in self.content.get('generator').items():
self.generator[k] = ConfigDict(v)