class Dwh(Database):
    """De datawarehouse database. Binnen deze database meerdere schema's."""
    def __init__(self, config: Dict[str, str] = {}) -> None:
        self.config = config  # type: Dict[str, str]
        if 'conn_dwh' in self.config:
            conn_string = self.config['conn_dwh']  # type: str
            super().__init__(conn_string)
        self.schemas = {}
        self.schemas['dv'] = Schema('dv', self, DwhLayerTypes.DV)  # type: Schema
        self.schemas['valset'] = Schema('valset', self, DwhLayerTypes.VALSET)  # type: Schema
        self.schemas['sys'] = Schema('sys', self, DwhLayerTypes.SYS)  # type: Schema

    def get_or_create_sor_schema(self, name) -> 'Schema':
        if name in  self.schemas:
            return self.schemas[name]
        else:
            sor = Schema(name, self, DwhLayerTypes.SOR)
            self.schemas[name] = sor
        return sor

    def get_schema(self, name=''):
        found = None
        if name in self.schemas:
            found = self.schemas[name]
        return found

    def set_schema(self, name, layer_type = DwhLayerTypes.DV):
        schema = Schema(name, self, layer_type)
        self.schemas[name] = schema

    @property
    def sors(self):
        return {k:v for k,v in self.schemas.items() if v.schema_type == DwhLayerTypes.SOR}

    @property
    def sys(self):
        return self.get_schema('sys')

    @property
    def dv(self):
        return self.get_schema('dv')

    @property
    def rdv(self):
        return self.get_schema('rdv')

    @property
    def valset(self):
        return self.get_schema('valset')

    def get_or_create_datamart_schema(self, dm_name: str) -> 'Schema':
        if dm_name in self.schemas:
            return self.schemas[dm_name]
        else:
            dm = Schema(dm_name, self, DwhLayerTypes.DM)
            self.schemas[dm_name] = dm
            return dm

    def create_schemas_if_not_exists(self, schema_name: str ='') -> None:
        self.reflect_schemas()
        for schema_name in self.schemas.keys():
            if not schema_name in self.reflected_schemas:
                sql = """CREATE SCHEMA {};""".format(schema_name)
                self.confirm_execute(sql, 'nieuw ' + schema_name + ' schema aanmaken')
        return
        if schema_name and not schema_name in self.reflected_schemas:
            sql = """CREATE SCHEMA {};""".format(schema_name)
            self.confirm_execute(sql, 'nieuw schema aanmaken')
            sor = Sor(schema_name, self)
            self.schemas[schema_name] = sor
        dv_schemas = {k:v for k,v in self.schemas.items() if v.schema_type == DwhLayerTypes.DV}
        for dv_schema_name in dv_schemas.keys():
            if not dv_schema_name in self.reflected_schemas:
                sql = """CREATE SCHEMA {};""".format(dv_schema_name)
                self.confirm_execute(sql, 'nieuw ' + dv_schema_name + ' schema aanmaken')
        if not self.sys.name in self.reflected_schemas:
            sql = """CREATE SCHEMA {};""".format(self.sys.name)
            self.confirm_execute(sql, 'nieuw sys schema aanmaken')
            sql = self.get_create_sys_ddl()
            self.confirm_execute(sql, 'nieuwe sys tabellen aanmaken')

    def get_create_sys_ddl(self) -> str:
        # ivm cross import reference conficts staat deze sql hier en niet in ddl.py
        sql = """
        CREATE TABLE sys.runs
        (
          runid numeric(8,2) NOT NULL,
          rundate timestamp without time zone NOT NULL DEFAULT now(),
          finish_date timestamp without time zone,
          exceptions boolean,
          sor_versions text,
          dv_version numeric(8,2),
          pyelt_version text,
          CONSTRAINT "PK" PRIMARY KEY (runid)
        )
        WITH (
          OIDS=FALSE
        );

        CREATE TABLE sys.currentversion
        (
          schemaname text NOT NULL,
          version numeric(8,2) NOT NULL,
          date timestamp without time zone,
          CONSTRAINT currentversion_pkey PRIMARY KEY (schemaname, version),
          CONSTRAINT currentversion_schemaname_key UNIQUE (schemaname)
        )
        WITH (
          OIDS=FALSE
        );
        INSERT INTO sys.runs (runid, rundate) VALUES (0, now());
        """ #type: str

        return sql

    def get_layer_version(self, layer_name) -> None:
        sql = """SELECT version FROM sys.currentversion WHERE schemaname = '{}'""".format(layer_name)
        rows = self.execute_read(sql, 'get version')
        if len(rows) > 0:
            return float(rows[0][0])
        else:
            return None

    def increase_layer_version(self, layer_name) -> None:
        old_version_number = self.get_layer_version(layer_name)
        if old_version_number:
            new_version_number = old_version_number + 0.01
            sql = """UPDATE sys.currentversion SET version = {} WHERE schemaname = '{}'""".format(new_version_number, layer_name)
            self.execute(sql, 'update version')
        else:
            new_version_number = 1.0
            sql = """INSERT INTO sys.currentversion (schemaname, version, date) VALUES ('{}', 1, now());""".format(layer_name)
            self.execute(sql, 'insert version')
        return new_version_number

    def confirm_execute(self, sql: str, log_message: str='') -> None:
        ask_confirm = False
        if 'ask_confirm_on_db_changes' in self.config:
            ask_confirm = self.config['ask_confirm_on_db_changes']
        else:
            ask_confirm = 'debug' in self.config and self.config['debug']
        super().confirm_execute(sql, log_message, ask_confirm) class DwhLayerTypes():
    """We onderscheiden 5 soorten database schemas:

    * Sor: laag met historische staging tabellen
    * Dv: laag met hubs, sats en links
    * Valset: laag met referentie data (code, omschrijving combinaties)
    * Dm: laag met dims en facts
    * Sys: laag met logging info over de runs

    De dwh kan meerdere sor lagen bevatten. aan te raden per import-bron 1 sor laag
    De dwh kan ook meerdere dv-lagen bevatten, zoals de raw-dv, de dv en de valuesets
    De dwh kan meerdere dm lagen bevatten: per bedrijfsonderdeel kan er een eigen dm laag worden gemaakt.
    Er is altijd maar 1 sys laag
    """
    SOR = 'SOR' #type: str
    DV = 'DV' #type: str
    VALSET = 'VALSET'  # type: str
    DM = 'DM' #type: str
    SYS = 'SYS' #type: str class DvValueset(AbstractOrderderTable):
    __dbschema__ = 'valset'
    _id = Columns.SerialColumn()
    _runid = Columns.FloatColumn()
    _source_system = Columns.TextColumn()
    _valid = Columns.BoolColumn()
    _validation_msg = Columns.TextColumn()
    _insert_date = Columns.DateTimeColumn()
    _finish_date = Columns.TextColumn()
    _active = Columns.BoolColumn(default_value=True, indexed=True)
    _revision = Columns.IntColumn()
    valueset_naam = Columns.TextColumn(unique=True)
    code = Columns.TextColumn(unique=True, indexed=True)
    omschrijving = Columns.TextColumn() class Logger:
    def __init__(self):
        self.logger = None #type: logging.logger
        self.start_time = datetime.now()  # type: datetime.datetime
        self.last_start_time = datetime.now()  # type: datetime.datetime
        self.errors = [] #type: List[str]
        self.to_console = True  # type: bool
        self.filename = ''  # type: str

    @staticmethod
    def create_logger(logger_type: str = LoggerTypes.MAIN, runid: float = 0, configs={}, to_console: bool = True, filename_args = '') -> 'Logger':
        logger = logging.getLogger(logger_type)
        # create file
        path, filename = Logger.__create_path_and_filename_by_type(logger_type, runid, configs, filename_args)
        if len(logger.handlers) == 0:
            # create formatter
            if logger_type == LoggerTypes.MAIN:
                formatter = logging.Formatter('%(asctime)s - %(message)s')
            else:
                formatter = logging.Formatter('%(message)s')

            fileHandler = logging.FileHandler(path + filename)
            fileHandler.setFormatter(formatter)
            logger.addHandler(fileHandler)
            # if to_console:
            #     consoleHandler = logging.StreamHandler()
            #     consoleHandler.setFormatter(formatter)
            #     logger.addHandler(consoleHandler)
            logger.setLevel(logging.INFO)
            # logger.errors = []
        log_obj = Logger()
        log_obj.logger = logger
        log_obj.start_time = datetime.now()
        log_obj.last_start_time = datetime.now()
        log_obj.errors = []  # logger.errors
        log_obj.to_console = to_console
        if 'log_to_console' in configs:
            log_obj.to_console = configs['log_to_console']
        log_obj.filename = filename
        log_obj.config = configs
        return log_obj

    @staticmethod
    def __create_path_and_filename_by_type(logger_type: str = LoggerTypes.MAIN, runid=0.00, configs={}, filename_args = ''):
        import os
        from main import get_root_path
        path = get_root_path() + configs['log_path']
        if logger_type == LoggerTypes.SQL and 'sql_log_path' in configs:
            path = get_root_path() + configs['sql_log_path']
        if logger_type == LoggerTypes.DDL and 'ddl_log_path' in configs:
            path = get_root_path() + configs['ddl_log_path']
        if not os.path.exists(path):
            os.makedirs(path)
        filename = 'LOG {0:%Y-%m-%d %H.%M.%S} {1}.log'.format(datetime.now(), logger_type)
        if runid:
            filename = 'LOG {1:%Y-%m-%d %H.%M.%S} RUN{0:07.2f} {2}.log'.format(runid, datetime.now(), logger_type)
        if filename_args:
            filename = filename.replace('.log', '_' + filename_args + '.log')

        return path, filename

    def log_simple(self, msg: str) -> None:
        self.logger.info(self.strip_formatting_tags(msg))
        if self.to_console:
            Logger.pprint(msg)

    def log(self, descr: str, last_start_time: datetime = None, rowcount: int = -1, newline: bool = False, indent_level=0) -> None:
        if not last_start_time:
            last_start_time = self.last_start_time
        newline_str = '' #type: str
        if newline:
            newline_str = '\r\n'

        end_time = datetime.now()
        global_time_descr = ''
        if last_start_time:
            global_elapsed_time = end_time - self.start_time
            elapsed_time_since_last_log = end_time - last_start_time
            global_time_descr = self.time_descr(global_elapsed_time)

        if descr:
            if indent_level >= 4:
                descr = '-' + descr
            for i in range(indent_level):
                descr = ' ' + descr
            for i in range(len(self.strip_formatting_tags(descr)), 60):
                # uitvullen tot 50 positities
                descr += ' '
                # divmod(elapsedTime.total_seconds(), 60)
        if descr and self.logger:
            if global_time_descr:
                if rowcount >= 0:
                    descr = '{0} <lightgray>executed on {1} rows in {2} ({3} since start) {4}</>'.format(descr, rowcount, self.time_descr(elapsed_time_since_last_log), global_time_descr, newline_str)
                else:
                    descr = '{0} <lightgray>executed in {1} ({2} since start) {3}</>'.format(descr, self.time_descr(elapsed_time_since_last_log), global_time_descr, newline_str)
            else:
                descr = '{}{}'.format(descr, newline_str)
        self.logger.info(self.strip_formatting_tags(descr))
        if self.to_console:
            Logger.pprint(descr)
        self.last_start_time = end_time

    def log_error(self, log_msg, sql='', err_msg= '', ex = None):
        if ex:
            for arg in ex.args:
                err_msg += str(arg)
        while err_msg.endswith('\n'):
            err_msg = err_msg[:-1]
        msg = """
<red>==========================================================================
ERROR tijdens : {}

SQL: {}

ERROR: {}
==========================================================================</>""".format(log_msg, sql, err_msg)
        if 'on_errors' in self.config and self.config['on_errors'] == 'throw':
            raise Exception(msg)
        else:

            self.logger.error(self.strip_formatting_tags(msg))
            self.errors.append(msg)
            if self.to_console:
                Logger.pprint(msg)
                # even wachten opdat de log-msg niet verward raakt door de foutmelding
                sleep(0.1)


    def time_descr(self, elapsed_time) -> str:
        sec = elapsed_time.total_seconds()
        time_descr = '{0:.3f} seconds'.format(sec)

        if sec > 60:
            import math
            min = math.floor(sec / 60)
            rest_sec = sec - min * 60
            time_descr = '{0} minutes and {1:.3f} seconds'.format(min, rest_sec)
        return time_descr

    @staticmethod
    def pprint(value):
        """Maakt het mogelijk in kleur te printen in de console
        Gebruik html-achtige tags zoals: <red>rood</>
        """
        if isinstance(value, list):
            for val in value:
                Logger.pprint(val)
        elif isinstance(value, str):
            value = value.replace('<br>', '\r\n')
            value = value.replace('<\br>', '\r\n')
            value = value.replace('<b>', ConsoleColors.BOLD)
            value = value.replace('<u>', ConsoleColors.UNDERLINE)
            value = value.replace('<red>', ConsoleColors.RED)
            value = value.replace('<gray>', ConsoleColors.GRAY)
            value = value.replace('<lightgray>', ConsoleColors.LIGHTGRAY)
            value = value.replace('<yellow>', ConsoleColors.YELLOW)
            value = value.replace('<blue>', ConsoleColors.BLUE)
            value = value.replace('<green>', ConsoleColors.GREEN)
            value = value.replace('<purle>', ConsoleColors.PURPLE)
            value = value.replace('<cyan>', ConsoleColors.CYAN)
            value = value.replace('<darkcyan>', ConsoleColors.DARKCYAN)
            value = value.replace('<filledred>', ConsoleColors.FILLED_RED)
            value = value.replace('</>', ConsoleColors.END)
            print(value)
        else:
            print(value)

    def strip_formatting_tags(self, value):
        value = value.replace('<br>', '\r\n')
        value = value.replace('<\br>', '\r\n')
        value = value.replace('<b>', '')
        value = value.replace('<u>', '')
        value = value.replace('<red>', '')
        value = value.replace('<gray>', '')
        value = value.replace('<lightgray>', '')
        value = value.replace('<yellow>', '')
        value = value.replace('<blue>', '')
        value = value.replace('<green>', '')
        value = value.replace('<purle>', '')
        value = value.replace('<cyan>', '')
        value = value.replace('<darkcyan>', '')
        value = value.replace('<filledred>', '')
        value = value.replace('</>', '')
        value = value.replace('\ufeff', '')
        return value

    def test_show_all_colors(self):
        for i in range(120):
            s = '\033[{}mDIT IS FORMAT. \033[0mDit niet.'.format(i)
            print(i, s) class LoggerTypes:
    MAIN = 'MAIN'
    SQL = 'SQL'
    DDL = 'DDL'
    ERROR = 'ERROR'
    PULL_DATA = 'PULL_DATA' class DomainValidator:
    wrong_field_name_chars = [' ', '*', '(', ')', '!', '@', '|', '{', '}', "'", '"', ':', ';', '#', '$', '%', '^', '&']
    """Helper om domein classes te valideren (entities, hubs, sats en links). Wordt aangeroepen voordat run start in pipeline.pipe"""
    def validate(self, module):
        validation_msg = ''
        for name, cls in inspect.getmembers(module,  inspect.isclass):
            if HubEntity in cls.__mro__ and cls is not HubEntity:
                # cls.cls_init()
                validation_msg += self.validate_entity(cls)
            elif LinkEntity in cls.__mro__ and cls is not LinkEntity:
                validation_msg += self.validate_linkentity(cls)
            # elif Link in cls.__mro__ and cls is not Link and cls is not HybridLink:
            #     validation_msg += self.validate_linkentity(cls)
            elif DvValueset in cls.__mro__ and cls is not DvValueset:
                validation_msg += self.validate_valueset(cls)
        return validation_msg

    def validate_entity(self, entity_cls):
        validation_msg = ''
        sat_classes = entity_cls.cls_get_sats()
        for sat_cls in sat_classes.values():
            columns = sat_cls.cls_get_columns()
            if not columns:
                validation_msg += 'Domainclass <red>{}.{}</> is niet geldig. Sat <red>{}</> zonder velden.\r\n'.format(entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name())
            for col in columns:
                if col.name[0].isnumeric():
                    validation_msg += 'Domainclass <red>{}.{}.{}</> is niet geldig. Columnname <red>{}</> is ongeldig.\r\n'.format(entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name(),
                                                                                                                                   col.name)

                for char in DomainValidator.wrong_field_name_chars:
                    if char in col.name:
                        validation_msg += 'Domainclass <red>{}.{}.{}</> is niet geldig. Columnname <red>{}</> is ongeldig.\r\n'.format(entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name(),
                                                                                                                                       col.name)

                if isinstance(col, Columns.RefColumn):
                    if not col.valueset_name:
                        validation_msg += 'Domainclass <red>{}.{}.{}</> is niet geldig. RefColumn <red>{}</> zonder type.\r\n'.format(entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name(),
                                                                                                                                      col.name)
            if sat_cls.__base__ == HybridSat:
                if not 'Types' in sat_cls.__dict__:
                    validation_msg += 'Hybrid Sat <red>{}.{}.{}</> is niet geldig. Definieer in de HybridSat een inner-class Types(HybridSat.Types) met een aantal string constanten. Dit is nodig om de databse view aan te maken.\r\n'.format(
                        entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name())
                elif sat_cls.__dict__['Types'] and not sat_cls.__dict__['Types'].__base__ is HybridSat.Types:
                    validation_msg += 'Hybrid Sat <red>{}.{}.{}</> is niet geldig. Inner-class Types moet erven van HybridSat.Types zoals "class Types(HybridSat.Types):" .\r\n'.format(
                        entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name())

        # for sat_cls in entity_cls.__sats__.values():
        #     if sat_cls.__name__.lower() == 'default' and entity_cls.__base__ is not HubEntity:
        #         validation_msg += 'Domainclass <red>{}.{}</> is niet geldig. Sat <red>Default</> mag alleen hangen onder de class die rechtstreeks erft van HubEntity (degene die de hub wordt).\r\n'.format(
        #             entity_cls.__module__, entity_cls.__name__, sat_cls.cls_get_name())


        return validation_msg

    def validate_linkentity(self, link_entity_cls: Link) -> str:
        validation_msg = ''
        if 'Link' not in link_entity_cls.__dict__:
            validation_msg += """LinkEntity <red>{}.{}</> is niet geldig. Een linkentity moet tenminste een subclass Link bevatten
             Een link entity moet er zo uit zien:
        <blue>
        class MedewerkerOrganisatieLinkEntity(LinkEntity):
            class Link(Link):
                medewerker = LinkReference(Medewerker)
                werkgever = LinkReference(Organisatie)"""
            return validation_msg
        else:
            link_cls = link_entity_cls.Link
            link_refs = link_cls.cls_get_link_refs()
            if len(link_refs) < 2:
                validation_msg += """Link <red>{}.{}</> is niet geldig. Een link moet ten minste twee LinkReferences bevatten (klasse variabelen van type LinkReference(HubEntity)).
                Een link moet er zo uit zien:
        <blue>
        class MedewerkerOrganisatieLinkEntity(LinkEntity):
            class Link(Link):
                medewerker = LinkReference(Medewerker)
                werkgever = LinkReference(Organisatie)
            </>""".format(
                    link_cls.__module__, link_cls.__name__)
            for link_ref in link_refs.values():
                if '_hub' not in link_ref.hub.__dbname__:
                    validation_msg += 'Link <red>{}.{}</> is niet geldig. LinkReference <red>{}</> moet verwijzen naar een Hub.\r\n'.format(
                        link_cls.__module__, link_cls.__name__, link_ref.name)
        return validation_msg

    def validate_valueset(self, valueset_cls: DvValueset) -> str:
        validation_msg = ''
        contains_code_field = False
        contains_valueset_name_field = False
        contains_ref_column = False
        for col in valueset_cls.cls_get_columns():
            if isinstance(col, Column):
                if col.name == 'code':
                    contains_code_field = True
                elif col.name == 'valueset_naam':
                    contains_valueset_name_field = True
                if isinstance(col, Columns.RefColumn):
                    contains_ref_column = True
        if not contains_code_field:
            validation_msg += """ValueSet <red>{}.{}</> is niet geldig. Een ValueSet moet ten minste een code veld bevatten.""".format(
                valueset_cls.__module__, valueset_cls.__name__)
        if not contains_code_field:
            validation_msg += """ValueSet <red>{}.{}</> is niet geldig. Een ValueSet moet ten minste een valueset_naam veld bevatten.""".format(
                valueset_cls.__module__, valueset_cls.__name__)
        # if contains_ref_column:
        #     validation_msg += """ValueSet <red>{}.{}</> is niet geldig. Een ValueSet mag geen refcolumn bevatten.""".format(
        #         valueset_cls.__module__, valueset_cls.__name__)
        return validation_msg class MappingsValidator:
    """Helper om domein classes te valideren (entities, hubs, sats en links). Wordt aangeroepen voordat run start in pipeline.pipe"""

    def validate_before_ddl(self, mappings_list: List[BaseTableMapping]) -> str:
        validation_msg = ''
        for mappings in mappings_list:
            if isinstance(mappings, SourceToSorMapping):
                validation_msg += self.validate_source_to_sor_mappings_before_ddl(mappings)

        # mapping_target_tables = [map.target for map in mappings_list]
        # if len(mapping_target_tables) != len(set(mapping_target_tables)):
        #     validation_msg += 'SorMappings zijn niet geldig. Er wordt meer keer gemapt op dezelfde doel tabel.\r\n'.format()

        return validation_msg

    def validate_after_ddl(self, mappings_list: List[BaseTableMapping]) -> str:
        validation_msg = ''
        for mappings in mappings_list:
            if isinstance(mappings, SourceToSorMapping):
                validation_msg += self.validate_source_to_sor_mappings_after_ddl(mappings)
            elif isinstance(mappings, SorToValueSetMapping):
                validation_msg += self.validate_sor_to_ref_mappings(mappings)
            elif isinstance(mappings, SorToEntityMapping):
                validation_msg += self.validate_sor_to_entity_mappings(mappings)
            elif isinstance(mappings, SorToLinkMapping):
                validation_msg += self.validate_sor_to_link_mappings(mappings)

        return validation_msg

    def validate_source_to_sor_mappings_before_ddl(self, mappings):
        validation_msg = ''
        source = mappings.source
        for key_name in source.key_names:
            if not key_name:
                validation_msg += 'Mapping <red>{}</> is niet geldig. Geen geldige key opgegeven.\r\n'.format(mappings.name)

        col_names = ''
        for field_map in mappings.field_mappings:
            col_names += """{}""".format(field_map.target)
        if not col_names:
            validation_msg += 'Mapping <red>{}</> is niet geldig. Geen geldige kolommen opgegeven. TIP: gebruik auto_map=True.\r\n'.format(mappings.name)

        for col in source.columns:
            if col.name.startswith('?'):
                validation_msg += 'Mapping <red>{}</> is niet geldig. Kolom is niet geldig. Controleer of alle velden een geldige alias hebben in de SourceQuery("""{}""")\r\n'.format(mappings.name,
                                                                                                                                                                                       source.sql)

        source_col_names = [col.name for col in mappings.source.columns]
        for ignore_col_name in mappings.ignore_fields:
            if ignore_col_name not in source_col_names:
                validation_msg += 'Mapping <red>{}</> is niet geldig. Ignore_field  <red>{}</> komt niet voor in bron.\r\n'.format(mappings.name, ignore_col_name)

        for key_name in mappings.source.key_names:
            if key_name not in source_col_names:
                validation_msg += 'Mapping <red>{}</> is niet geldig. Key <red>{}</> komt niet voor in bron.\r\n'.format(key_name, key_name)


        return validation_msg

    def validate_source_to_sor_mappings_after_ddl(self, mappings):
        validation_msg = ''
        if not mappings.auto_map:
            validation_msg += 'Mapping <red>{}</> is niet geldig. Auto_map moet aan staan bij SourceToSorMapping.\r\n'.format(mappings.name)

        source_col_names = [col.name for col in mappings.source.columns]
        if not mappings.field_mappings:
            validation_msg += 'Mapping <red>{}</> is niet geldig. Geen velden gemapt. Gebruik auto_map = True of map_field() om velden te mappen.\r\n'.format(mappings.name)
        for fld_mapping in mappings.field_mappings:
            if isinstance(fld_mapping.source, str) and fld_mapping.source.name not in source_col_names:
                validation_msg += 'Mapping <red>{}</> is niet geldig. Veld <red>{}</> komt niet voor in bron. TIP: gebruik haakjes () voor berekeningen of om constanten te definieren als bronvelden. Je kunt ook ConstantValue() gebruiken (zie bron header voor geldige veldnamen).\r\n'.format(
                    mappings.name,
                    fld_mapping.source.name)

        fld_mappings_names = [fld_mapping.name for fld_mapping in mappings.field_mappings]
        if len(fld_mappings_names) != len(set(fld_mappings_names)):
            validation_msg += 'Mapping <red>{}</> is niet geldig. Mappings zijn niet uniek.\r\n'.format(mappings.name)

        keys = mappings.keys
        if not keys:
            validation_msg += 'Mapping <red>{}</> is niet geldig. Geen primary key gemapt.\r\n'.format(mappings.name)
        else:
            invalid_key_names = ''
            for key in keys:
                if key not in source_col_names:
                    invalid_key_names += key + ','
            if invalid_key_names:
                validation_msg += 'Mapping <red>{}</> is niet geldig. Key: [<red>{}</>] komt niet voor in bron. (zie bron header voor geldige veldnamen).\r\n'.format(mappings.name, invalid_key_names)
        if len(keys) != len(set(keys)):
            validation_msg += 'Mapping <red>{}</> is niet geldig. Key-velden niet uniek.\r\n'.format(mappings.name)

        fixed_field_names = ['_id', '_runid', '_active', '_source_system', '_insert_date', '_finish_date', '_revision', '_valid', '_validation_msg', '_hash']
        for fixed_field_name in fixed_field_names:
            if fixed_field_name in source_col_names:
                validation_msg += 'Mapping <red>{}</>, veld <red>{}</> is niet geldig. Mag de volgende veldnamen niet gebruiken: {}.\r\n'.format(mappings.name, fixed_field_name, fixed_field_names)

        # if isinstance(mappings.source, SourceQuery) and mappings.filter:
        #     validation_msg += 'Mapping <red>{}</> is niet geldig. Mag geen filter opgeven bij SouryQuery(). Gebruik een where clause.\r\n'.format(mappings.name)
        if isinstance(mappings.source, SourceQuery) and mappings.ignore_fields:
            validation_msg += 'Mapping <red>{}</> is niet geldig. Mag geen ignore_fields opgeven bij SouryQuery. Sluit de velden uit in de sql.\r\n'.format(mappings.name)

        return validation_msg

    def validate_sor_to_entity_mappings(self, mappings):
        validation_msg = ''
        if isinstance(mappings.source, SorTable) and not (mappings.bk_mapping or mappings.key_mappings):
            validation_msg += 'Mapping <red>{}</> is niet geldig. Bk mapping of external_key_mapping ontbreekt. Gebruik map_bk(str of list(str) of mapfield(str of list(str), bk)\r\n'.format(
                mappings.name)



        sor_table = mappings.source
        if not sor_table.is_reflected:
            sor_table.reflect()
        source_col_names = [col.name for col in sor_table.columns]
        if isinstance(mappings.source, SorQuery):
            # controleer of tenminste de vaste velden aanwezig zijn
            if '_runid' not in source_col_names or '_active' not in source_col_names or '_valid' not in source_col_names:
                validation_msg += 'Mapping <red>{}</> is niet geldig. Noodzakelijke vaste velden (_runid, _active en _valid) ontbreken in query. Ook fk naar hub moet reeds aanwezig zijn in hstage tabel. Je moet [tabel]_hstage.* gebruiken voor de sor-tabel die de fk naar de hub bevat. \r\n'.format(
                    mappings.name)
            # if not '.*' in mappings.source.sql:
            #     validation_msg += 'Mapping <red>{}</> is niet geldig. Geldige fk naar hub ontbreekt in sql. Je moet [tabel]_hstage.* gebruiken voor de sor-tabel die de fk naar de hub bevat. Deze moet reeds eerder zijn gemapt op de hub als SorTable().\r\n'.format(
            #         mappings.name)
        for sat_mappings in mappings.sat_mappings.values():
            fld_mappings_names = [fld_mapping.name for fld_mapping in sat_mappings.field_mappings]
            # testen op uniek
            if len(fld_mappings_names) != len(set(fld_mappings_names)):
                validation_msg += 'Mapping <red>{}</> is niet geldig. Mappings zijn niet uniek.\r\n'.format(sat_mappings.name)
            fld_mappings_target_names = [fld_mapping.target.name for fld_mapping in sat_mappings.field_mappings if fld_mapping.target.type != 'jsonb']
            if len(fld_mappings_target_names) != len(set(fld_mappings_target_names)):
                validation_msg += 'Mapping <red>{}</> is niet geldig. Doelvelden zijn niet uniek.\r\n'.format(sat_mappings.name)
            # testen of doel veld bestaat
            for fld_mapping in sat_mappings.field_mappings:
                if isinstance(fld_mapping.source, ConstantValue):
                    continue

                if not isinstance(fld_mapping.source, Column):
                    # overslaan
                    continue

                source_col_name = fld_mapping.source.name
                if '(' in source_col_name:
                    # sql functies overslaan
                    continue


                source_col_name = source_col_name.split('::')[0]
                if source_col_name not in source_col_names:
                    validation_msg += 'Mapping <red>{}</> is niet geldig. Veld <red>{}</> komt niet voor in bron. (zie bron header voor geldige veldnamen).\r\n'.format(mappings.name,
                                                                                                                                                                        source_col_name)


            if sat_mappings.target.__base__ == HybridSat:
                if not sat_mappings.type:
                    validation_msg += 'Mapping <red>{}</> is niet geldig. Bij mappings naar Hybdrid sats moet elke veldmapping van een type worden voorzien bijv. (type = mobiel)\r\n'.format(
                        mappings.name)
        return validation_msg

    def validate_sor_to_ref_mappings(self, mappings):
        validation_msg = ''

        # if not isinstance(mappings.source, dict) and not mappings.source_descr_field:
        #     validation_msg += 'Mapping <red>{}</> is niet geldig. Omschrijvingsveld ontbreekt.\r\n'.format(
        #                 mappings.name)

        return validation_msg

    def validate_sor_to_link_mappings(self, mappings):
        validation_msg = ''
        if len(mappings.field_mappings) < 2:
            validation_msg += 'Linkmapping <red>{}</> is niet geldig. Moet minstens naar 2 velden verwijzen.\r\n'.format(mappings.name)

        num_dynamic_refs = 0
        for fld_mapping in mappings.field_mappings:
            if isinstance(fld_mapping.target, DynamicLinkReference):
                num_dynamic_refs += 1
        if num_dynamic_refs > 1:
            validation_msg += 'Linkmapping <red>{}</> is niet geldig. Link mag maar 1 DynamicLinkReference veld bevatten.\r\n'.format(mappings.name)

        return validation_msg

        # todo: betere foutmeldingen
        # todo: controle of sat compleet is gemapt class SorToValueSetMapping(BaseTableMapping):
    def __init__(self, source: str, target: DvValueset, sor: Schema = None ) -> None:
        if isinstance(source, str) and 'SELECT' in source.upper() and ' ' in source:
            source = SorQuery(source, sor)
        elif isinstance(source, str):
            source = SorTable(source, sor)
        super().__init__(source, target)
        self.sor_table = source
        self.valueset = target

    def get_source_fields(self, alias: str = '')-> str:
        if not alias: alias = 'hstg'
        fields = ''  # type: List[str]
        target_json_dict = OrderedDict()
        for field_map in self.field_mappings:
            if field_map.target.type == 'jsonb':
                if field_map.target.name not in target_json_dict:
                    target_json_dict[field_map.target.name] = []
                target_json_dict[field_map.target.name].append(field_map)
            elif isinstance(field_map.source, ConstantValue):
                field = '{},'.format(field_map.source)
            elif isinstance(field_map.source, FieldTransformation):
                # field = '{},'.format(field_map.source.get_sql(alias))
                field = '{},'.format(field_map.source.get_sql())
            elif isinstance(field_map.source, DbFunction):
                # field = '{},'.format(field_map.source.get_sql(alias))
                field = '{},'.format(field_map.source.get_sql(alias))
            elif not alias:
                field = '{},'.format(field_map.source)
            else:
                field = '{}.{},'.format(alias, field_map.source)
            if '[]' in field_map.target.type:
                #array velden
                #eerst komma eraf halen
                field = field[:-1]
                field = "'{" + field + "}',"
            # voorkom eventuele dubbele veldnamen bij hybrid sats
            if field not in fields:
                fields += field

        for name, value in target_json_dict.items():
            sql_snippet = """json_build_object("""
            for field_map in value:
                sql_snippet += """'{0}', {0}, """.format(field_map.source.name)
            sql_snippet = sql_snippet[:-2] + ')::jsonb,'
            fields += sql_snippet
        fields = fields[:-1]
        return fields

    def get_target_fields(self) -> str:
        fields = []
        #json velden willen we helemaal als laatste omdat ze in get_source_fields ook als laatste komen te staan
        json_fields = []
        for field_map in self.field_mappings:
            field = '{}'.format(field_map.target)
            if field_map.target.type == 'jsonb':
                if field not in json_fields:
                    json_fields.append(field)
            elif field not in fields:
                fields.append(field)
        return_sql = ','.join(fields)
        if json_fields:
            return_sql += "," + ','.join(json_fields)
        return return_sql class EntityViewToEntityMapping(SorToEntityMapping):
    def __init__(self, source, target, filter: str = '') -> None:
        super().__init__(source, target, filter)
        self.view = source class EntityViewToLinkMapping(SorToLinkMapping):
    def __init__(self, source, target, filter=''):
        super().__init__(source, target, filter)
        self.view = source

    def add_mapping(self, *args, **kwargs):
        if len(args) == 1:
            hub_name = args[0]
            source_fk = '_id'
            target_fk = 'fk_' + hub_name
        elif len(args) == 2:
            source_fk = args[0]
            hub_name = args[1]
            target_fk = 'fk_' + hub_name

        db = Dwh()
        hub = Table(hub_name, db.dv)
        source_field = Column('_id',  tbl=hub)
        target_field = Column(target_fk, 'integer')
        field_mapping = LinkFieldMapping(source_field, target_field)
        if 'join' in kwargs:
           field_mapping.join = kwargs['join']
        self.field_mappings.append(field_mapping)

    def get_source_fks(self):
        fks = ''
        for field_mapping in self.field_mappings:
            alias = field_mapping.source.table.type
            if alias:
                fks += '{}._id, '.format(alias)
            else:
                fks += '{}._id, '.format(field_mapping.source.table)
        fks = fks[:-2]
        return fks

    def get_from(self, dv_name = 'dv'):
        from_sql = ''
        for field_mapping in self.field_mappings:
            join_tbl = field_mapping.get_source_table()
            alias = join_tbl.type
            if alias:
                from_sql += '{} AS {}, '.format(join_tbl, alias)
            else:
                from_sql += '{}, '.format(join_tbl)
        from_sql = from_sql[:-2]
        return from_sql

    def get_join_where(self, dv_name = 'dv'):

        join_sql = ''
        for field_mapping in self.field_mappings:
            join = field_mapping.join
            bk = field_mapping.bk
            if bk:
                join_tbl = field_mapping.get_source_table()
                alias = join_tbl.type
                if alias:
                    join_sql += ' {0}.bk = {1} AND '.format(alias, bk)
                else:
                    join_sql += ' {0}.bk = {1} AND '.format(join_tbl, bk)
        return join_sql

    def get_fks_compare(self, source_alias='', target_alias=''):

        fks_compare = ''

        for field_mapping in self.field_mappings:

            # source_alias = self.get_alias_of_source_tbl(field_mapping, fks_compare)
            if isinstance(field_mapping.source, ConstantValue):
                fks_compare += '{0} = {1}.{2} AND '.format(field_mapping.source, target_alias, field_mapping.target)
            else:
                fks_compare += '{0}.{1} = link.{2} AND '.format(field_mapping.source.table, field_mapping.source, field_mapping.target)
        fks_compare = fks_compare[:-4]
        return fks_compare class SorToEntityMapping(BaseTableMapping):
    def __init__(self, source: str, target: HubEntity, sor: Schema = None, filter: str = '', type: str = '') -> None:
        # target.cls_init_sats()

        # if not sor:
        #     sor_name = source.split('.')[0]
        #     from pyelt.pipeline import Pipeline
        #     p = Pipeline()
        #     db = p.dwh
        #     sor = db.get_sor_schema(sor_name)
        if isinstance(source, str) and 'SELECT' in source.upper() and ' ' in source:
            source = SorQuery(source, sor)
        elif isinstance(source, str):
            source = SorTable(source, sor)
        if not sor and source is SorTable:
            raise Exception('Je moet een sor laag opgeven bij aanmaken van de SorToEntityMapping')
        # target.init_cls()
        super().__init__(source, target, filter)
        self.entity = target
        self.type = type
        if not type and not target.__base__ == HubEntity:
            self.type = target.__subtype__
        self.sat_mappings = {} #type: Dict[str, SorSatMapping]
        self.bk_mapping = None #type: str
        self.key_mappings = []  # type: str

    def map_field(self, source: Union[str, 'ConstantValue', list, dict], target: 'Column' = None, transform_func: 'FieldTransformation'=None, ref: str = '', type: str = '') -> None:
        if not source:
            return
        if not target:
            return
        if target.name == 'bk':
            return self.map_bk(source)
        sat = target.get_table()

        sat_name = sat.cls_get_name()
        sat_mapping = SorSatMapping(self.source, sat, type)
        if not sat_mapping.name in self.sat_mappings:
            self.sat_mappings[sat_mapping.name] = sat_mapping
        else:
            sat_mapping = self.sat_mappings[sat_mapping.name]
        sat_mapping.map_field(source, target, transform_func, ref)
        sat_mapping.type = type
        # field_mapping = FieldMapping(source, target, transform_func, ref = ref)
        # self.field_mappings.append(field_mapping)

    def map_bk(self, source: Union[str,List[str]]) -> None:
        if isinstance(source, list):
            source = " || '.' ||".join(source)
        self.bk_mapping = source

    def map_keys_in_sat(self, source: Union[str, List[str]], target: Union['Column', List['Column']]):
        """In plaats van een bk_mapping kun je ook mappen op een veld combinatie in een reeds bestaande sat.

        Voorwaarde is dat er reeds een gevulde hub met een sat is, die vanuit een andere bron is gevuld.

        Denk hierbij ook aan een zogenaamde keys_sat: een sat met alleen sleutel info vanuit de verschillende bronnen"""
        if isinstance(source, list):
            for index in range(len(source)):
                self.key_mappings.append(KeyMapping(source[index], target[index]))



    def map_fixed_value(self, fixed_value: str, target_field: 'Column' = None) -> None:
        source_field = ConstantValue("'{}'".format(fixed_value))
        self.map_field(source_field, target_field) class SorToLinkMapping(BaseTableMapping):
    auto_generated_sor_fk = 'auto'

    def __init__(self, source: Union[str, 'Table', 'View'], target: 'LinkEntity', sor: Schema, filter: str = '', type: str = '') -> None:
        if isinstance(source, str):
            source = SorTable(source, sor)
        super().__init__(source, target, filter)
        self.sor_table_name = str(source)
        if isinstance(type, str):
            self.type = type
        elif type == HubEntity:
            self.type = type.__subtype__
        self.target = target
        # self.target.cls_init()
        self.hubs = {} #type: Dict[str, Table]
        self.sat_mappings = {}  # type: Dict[str, Sat]

    def get_fk_name(self, entity_cls):
        for prop_name, entity in self.target.__dict__.items():
            if entity == entity_cls or entity == entity_cls.__base__:
                fk = """fk_{}""".format(entity.cls_get_hub_name())
                if prop_name.lower() != entity.__name__.lower():
                    fk = """fk_{}_{}""".format(prop_name.lower(), entity.cls_get_hub_name())
                return fk
        raise Exception('Entity niet gevonden in Link')

    def map_bk(self, source_bk, target_link_ref: Union[LinkReference, DynamicLinkReference], join: str = '', type: Union[str, 'HubEntity'] = ''):
        if not (isinstance(target_link_ref, LinkReference) or isinstance(target_link_ref, DynamicLinkReference)):
            raise PyeltException('Link moet verwijzen naar veld van type LinkReference of DynamicLinkReference')

        if isinstance(target_link_ref, LinkReference):
            hub_name = target_link_ref.hub.__dbname__
            hub = target_link_ref.hub

        elif isinstance(target_link_ref, DynamicLinkReference):
            # type verwijst hier naar de entity die dynamisch gelinkt wordt in de fk
            if not type or isinstance(type, str):
                raise PyeltException('Bij DynamicLinkReference moet type verwijzen naar een afgeleide van HubEntity')
            hub_name = type.cls_get_hub_name()
            hub = type.cls_get_hub()

            self.type = type.__name__.lower()
        hub_alias = target_link_ref.fk.replace('fk_', '')

        source_field = '_id' #type: str
        source_col = Column(source_field, 'integer', tbl=hub) #type: Column
        target_fk = target_link_ref.fk
        target_col = Column(target_fk, 'integer', tbl=hub)
        field_mapping = LinkFieldMapping(source_col, target_col)
        field_mapping.join = join
        field_mapping.source_alias = hub_alias
        if isinstance(source_bk, list):
            source_bk = "|| '.' || ".join(source_bk)
        field_mapping.bk = source_bk
        self.hubs[target_fk] = hub
        self.field_mappings.append(field_mapping)

    def map_sor_fk(self, source_fk: str = '', target_link_ref: Union[LinkReference, DynamicLinkReference] = None, join: str = '', type: Union[str, 'HubEntity'] = ''):
        if not (isinstance(target_link_ref, LinkReference) or isinstance(target_link_ref, DynamicLinkReference)):
            raise PyeltException('Link moet verwijzen naar veld van type LinkReference of DynamicLinkReference')
        if isinstance(target_link_ref, LinkReference):
            hub_name = target_link_ref.hub.__dbname__
            hub = target_link_ref.hub
            # if not type and target_link_ref.sub_entity_type != hub_name.replace('_hub', ''):
            #     type = target_link_ref.sub_entity_type
            # if not type and not target_link_ref.hub.__base__ == HubEntity:
            #     type = target_link_ref.hub.__dbname__.replace('_hub', '')

            if not source_fk or source_fk == SorToLinkMapping.auto_generated_sor_fk:
                # source_fk = target_link_ref.fk
                source_fk = 'fk_{}{}'.format(target_link_ref.sub_entity_type, hub_name)
        elif isinstance(target_link_ref, DynamicLinkReference):
            # type verwijst hier naar de entity die dynamisch gelinkt wordt in de fk
            if not type or isinstance(type, str):
                raise PyeltException('Bij DynamicLinkReference moet type verwijzen naar een afgeleide van HubEntity')
            hub = type.cls_get_hub()
            self.type = type.__name__.lower()
        source_col = Column(source_fk, 'integer', tbl=self.sor_table_name)

        target_fk = target_link_ref.fk
        target_col = Column(target_fk, 'integer', tbl=hub)
        field_mapping = LinkFieldMapping(source_col, target_col)
        field_mapping.join = join
        self.hubs[target_fk] = hub
        self.field_mappings.append(field_mapping)

    def map_entity(self, target_link_ref: Union[LinkReference, DynamicLinkReference] = None, bk: Union[List[str], str] = '', join: str = '', type: Union[str, 'HubEntity'] = ''):
        if isinstance(target_link_ref, DynamicLinkReference) and not bk:
            hub_name = type.cls_get_hub_name()
            sor_fk = 'fk_{}'.format(hub_name)
            self.map_sor_fk(sor_fk, target_link_ref, join, type)
        elif bk:
            self.map_bk(bk, target_link_ref, join, type)
        else:
            self.map_sor_fk('', target_link_ref, join, type)

    # def map_dynamic_entity(self, source_entity: HubEntity, target_link_ref: DynamicLinkReference,  type: str = '', bk: str = '', join: str = ''):
    #     if bk:
    #         self.map_bk(bk, target_link_ref, join, type)
    #     else:
    #         hub_name = target_link_ref.entity_cls.get_hub_name()
    #         sor_fk = 'fk_{}{}'.format(type, hub_name)
    #         type = source_entity
    #         self.map_sor_fk(sor_fk, target_link_ref, join, type)

    def map_field(self, source: Union[str, 'ConstantValue'], target: 'Column' = None, transform_func: 'FieldTransformation' = None, ref: str = '', type: str = '') -> None:
        if not target:
            return
        sat = target.get_table()
        sat_name = sat.cls_get_name()
        sat_mapping = SorSatMapping(self.source, sat, type)
        if not sat_mapping.name in self.sat_mappings:
            self.sat_mappings[sat_mapping.name] = sat_mapping
        else:
            sat_mapping = self.sat_mappings[sat_mapping.name]
        sat_mapping.map_field(source, target, transform_func, ref)
        sat_mapping.type = type

    def map_entity_old(self, entity_cls: HubEntity,  bk: str = '', join: str = '', type: str = ''):
        # is_hybrid = False
        hub_name = entity_cls.cls_get_hub_name()
        hub = entity_cls.cls_get_hub()

        if bk:
            source_field = '_id' #type: str
            source_col = Column(source_field, 'integer', tbl=hub) #type: Column
        # elif is_hybrid:
        #     source_field = 'fk_{}{}'.format(type, hub_name)
        #     source_col = Column(source_field, 'integer', tbl=self.sor_table)
        else:
            # source_field = 'fk_' + hub_name
            source_field = 'fk_{}{}'.format(type, hub_name)
            source_col = Column(source_field, 'integer', tbl=self.sor_table_name)

        # target_field = '_fk_' + hub_name #type: str
        target_field = 'fk_{}'.format(hub_name)
        target_field = self.get_fk_name(entity_cls)
        # if target_field in self.hubs:
        #     target_field = '_fk_parent_' + self.type + hub_name
        target_col = Column(target_field, 'integer', tbl=hub)
        field_mapping = LinkFieldMapping(source_col, target_col)
        field_mapping.bk = bk
        field_mapping.join = join
        self.hubs[target_field] = hub
        self.field_mappings.append(field_mapping)

    # def map_source(self, source, target: 'HubEntity'):
    #     source_col = Column(source, 'integer', tbl=self.sor_table)
    #     target_field = """_fk_source_{}""".format(target.get_hub_name())
    #     target_col = Column(target_field, 'integer', tbl=target.hub)
    #     field_mapping = LinkFieldMapping(source_col, target_col)
    #     self.hubs[target_field] = target.hub
    #     self.field_mappings.append(field_mapping)
    #
    # def map_target(self, source, target: 'HubEntity'):
    #     source_col = Column(source, 'integer', tbl=self.sor_table)
    #     target_field = """_fk_target_{}""".format(target.get_hub_name())
    #     target_col = Column(target_field, 'integer', tbl=target.hub)
    #     field_mapping = LinkFieldMapping(source_col, target_col)
    #     # field_mapping.bk = bk
    #     # field_mapping.join = join
    #     self.hubs[target_field] = target.hub
    #     self.field_mappings.append(field_mapping)
    #
    #     # source_col = Column(source, 'integer', tbl=self.sor_table)
    #     # target_field = target.__qualname__
    #     # mro = target.__mro__
    #     # currentframe = inspect.currentframe()
    #     # fs = inspect.getouterframes(currentframe)
    #     # target_field = self.get_fk_name(target)
    #     # # if target_field in self.hubs:
    #     # #     target_field = '_fk_parent_' + self.type + hub_name
    #     # target_col = Column(target_field, 'integer', tbl=target.hub)
    #     # field_mapping = LinkFieldMapping(source_col, target_col)
    #     # # field_mapping.bk = bk
    #     # # field_mapping.join = join
    #     # self.hubs[target_field] = target.hub
    #     # self.field_mappings.append(field_mapping) class SourceToSorMapping(BaseTableMapping):
    def __init__(self, source: Union['SourceTable', 'SourceQuery', 'File'], target: Union[str, Table], auto_map: bool = True, filter='', ignore_fields: List[str] = []) -> None:
        #todo transformations
        if isinstance(source, File):
            self.file_name = source.file_name
        elif isinstance(source, SourceTable):
            self.source_table = source
        elif isinstance(source, SourceQuery):
            self.source_table = source

        if isinstance(target, str):
            self.sor_table = str(target) #type: str
        else:
            target_tbl = cast(Table, target)
            self.sor_table = target_tbl.name
        super().__init__(source, target, filter)
        self.temp_table = self.sor_table.replace('_hstage', '') + '_temp'
        #todo keys
        self.keys = [] #type: List[str]
        ignore_fields = [s.lower() for s in ignore_fields]
        self.ignore_fields = ignore_fields
        # self.field_mappings = [] #type: List[FieldMapping]
        self.auto_map = auto_map
        if auto_map: self.create_auto_mappings(source, ignore_fields)



    def map_field(self, source: str, target: str = '', transform_func: 'FieldTransformation'=None, ref: str = '') -> None:
        super().map_field(source, target, transform_func)
        if source in self.source.key_names:
            self.keys.append(source)

    # def map_pk(self, source: str, target: str = '', transform_func: 'FieldTransformation'=None, ref: str = '') -> None:

    def create_auto_mappings(self, source: Union['SourceTable', 'SourceQuery', 'File'], ignore_fields: List[str] = []):
        if isinstance(source, File):
            field_names = source.get_header()
            for fld_name in field_names:
                if fld_name in ignore_fields:
                    continue
                self.map_field(fld_name, fld_name)
        elif isinstance(source, SourceQuery):
            if not source.is_reflected:
                source.reflect()
            for col in source.columns:
                if col.name in ignore_fields:
                    continue
                self.map_field(col.name, col.name)
        elif isinstance(source, SourceTable):
            if not source.is_reflected:
                source.reflect()
            for col in source.columns:
                if col.name in ignore_fields:
                    continue
                self.map_field(col.name, col.name)
        self.keys = [name.lower() for name in source.key_names]


    def get_fields(self, alias: str = '') -> str:
        return super().get_target_fields(alias)

    def get_fields_compare(self, source_alias: str='', target_alias: str='') -> str:
        if not source_alias: source_alias = 'tmp'
        if not target_alias: target_alias = 'hstg'
        return super().get_fields_compare(source_alias, target_alias)

    def get_keys(self, alias: str='') -> str:
        if not alias:
            return ','.join(self.keys)
        else:
            keys = ''
            for key in self.keys:
                keys += '{}.{},'.format(alias, key)
                keys = keys[:-1]
            return keys

    def get_keys_compare(self,    source_alias: str = '', target_alias: str = '') -> str:
        if not source_alias: source_alias = 'tmp'
        if not target_alias: target_alias = 'hstg'
        keys_compare = ''
        for key in self.keys:
            keys_compare += '{0}.{2} = {1}.{2} AND '.format(source_alias, target_alias, key)
        keys_compare = keys_compare[:-4]
        return keys_compare

    def get_source_encoding(self):
        #postgres valid name
        file_encoding = 'UTF-8'
        if isinstance(self.source, File):
            file_encoding = self.source.encoding.lower()
            if file_encoding == 'ascii' or file_encoding == 'latin-1':
                # conversie naar geldige postgres var
                file_encoding = 'LATIN1'
        return file_encoding

    def get_delimiter(self):
        delimiter = ';'
        if isinstance(self.source, File):
            delimiter = self.source.delimiter.lower() or ';'
        return delimiter

    def get_quote(self):
        quote = '|'
        if isinstance(self.source, File):
            quote = self.source.quote.lower() or '|'
        return quote class SorValidation(Validation):
    def __init__(self, tbl: 'Table' = None, schema: 'Schema' = None) -> None:
        super().__init__(tbl, schema)
        self.check_for_duplicate_keys = []


    def set_check_for_duplicate_keys(self, keys):
        if isinstance(keys, str):
            self.check_for_duplicate_keys = [keys]
        else:
            self.check_for_duplicate_keys = keys

    def get_keys(self):
        keys = ','.join(self.check_for_duplicate_keys)
        return keys class DvValidation(Validation):
    def set_condition(self, field: Union[Column, str], operator: str = '=', sql_value="''"):
        if isinstance(field, str):
            self.sql_condition = field
        elif isinstance(field, Condition):
            self.sql_condition = field.name
            self.table = field.table
        else:
            params = {'field': field.name, 'operator': operator, 'sql_value': sql_value}
            self.table = field.table
            if field.name == 'bk':
                pass
            self.sql_condition = "{field} {operator} {sql_value}".format(**params) class Validation():
    class Operators:
        equals = '='
        not_equals = '!='
        gt = '>'
        lt = '<'
        like = 'like'
        not_like = 'not like'

    def __init__(self, tbl: 'Table' = None, schema: 'Schema' = None) -> None:
        if isinstance(tbl, str):
            tbl = Table(tbl, schema)
        self.tbl = tbl  # type: Table
        self.msg = ''  # type: str
        self.sql_condition = ''  # type: str


    def __str__(self):
        return self.sql_condition

    def report_run(self, pipeline):
        runid = pipeline.runid
        sql = "SELECT schema, table_name, message, key_fields, fields FROM dv._exceptions WHERE _runid = {0} ORDER BY schema;".format(runid)
        rows = pipeline.dwh.execute_read(sql)
        msg = ''
        if rows:
            pipeline.logger.log_simple('<red>ROW VALIDATION ERRORS DURING ETL</>')
            pipeline.logger.log_simple('<red>(validation errors at sor-layers are not imported in dv-layer!)</>')
        for row in rows:
            msg = 'schema: <blue>{}</>, table : <blue>{}</>, message: <blue>{}</>, key: <blue>{}</>, row: <blue>{}</>'.format(row[0], row[1], row[2], row[3], row[4])
            pipeline.logger.log_simple(msg)
            # msg += 'VALIDATION ERRORS DURING ETL: FOUTEN: ' + str(row) + '<br>'
        return '' class EtlSourceToSor(BaseEtl):
    def __init__(self, pipe):
        super().__init__(pipe)

    def _get_fixed_params(self) -> Dict[str, Any]:
        params = {}
        params['runid'] = self.runid
        params['source_system'] = self.pipe.source_system
        params['sor'] = self.pipe.sor.name
        return params

    # def source_to_sor(self, mappings):
    #     if isinstance(mappings.source, File):
    #         self.source_file_to_sor(mappings)
    #     elif isinstance(mappings.source, SourceTable):
    #         self.source_db_to_sor(mappings)
    def init_source_to_sor_old(self, mappings):
        params = mappings.__dict__
        params.update(self._get_fixed_params())
        params['fields'] = mappings.get_fields()
        params['key_fields'] = mappings.get_keys()
        params['tmp_fields'] = mappings.get_fields(alias='tmp')
        params['fields_compare'] = mappings.get_fields_compare(source_alias='tmp', target_alias='hstg')
        params['keys_compare'] = mappings.get_keys_compare(source_alias='tmp', target_alias='hstg')

        self.append_step('source to csv', self.source_to_csv_old, params)
        sql = "TRUNCATE TABLE {sor}.{temp_table}_hash;".format(**params)
        self.append_step('truncate temp', self.dwh.execute, sql)
        sql = "COPY {sor}.{temp_table}_hash ({key_fields}, _hash) FROM  '{file_name}' DELIMITER ';' CSV HEADER;".format(**params)
        self.append_step('copy into temp hash', self.dwh.execute, sql)
        sql = "SELECT COUNT(*) FROM {sor}.{sor_table};".format(**params)
        result = None
        self.append_step('get rowcount', self.dwh.execute_read, sql, result)
        #todo append voorwaarde & return
        self.append_condition(result > 0, yes = [], no = [] )
        self.append_step('', self.dwh.execute, sql)
        self.append_step('', self.dwh.execute, sql)
        self.append_step('', self.dwh.execute, sql)

    def source_to_csv_old(self, params):
        if isinstance(params['source'], SourceTable):
            file_name = params['source'].to_csv()
            params['file_name'] = file_name

    def source_to_sor(self, mappings):
        by_md5 = type(mappings.source) is SourceTable or type(mappings.source) is SourceQuery  # isinstance(mappings.source, SourceTable)
        if by_md5:
            self.source_to_sor_by_hash(mappings)
            return
        try:
            debug = 'debug' in self.pipe.pipeline.config and self.pipe.pipeline.config['debug']

            params = mappings.__dict__
            params.update(self._get_fixed_params())
            params['fields'] = mappings.get_fields()
            params['key_fields'] = mappings.get_keys()
            params['tmp_fields'] = mappings.get_fields(alias='tmp')
            params['fields_compare'] = mappings.get_fields_compare(source_alias='tmp', target_alias='hstg')
            params['keys_compare'] = mappings.get_keys_compare(source_alias='tmp', target_alias='hstg')
            params['encoding'] = mappings.get_source_encoding()
            params['delimiter'] = mappings.get_delimiter()
            params['quote'] = mappings.get_quote()

            # STAP 1
            if isinstance(mappings.source, SourceTable):
                file_name = mappings.source.to_csv(filter=mappings.filter, ignore_fields=mappings.ignore_fields, debug=debug)
                params['file_name'] = file_name
                # self.logger.log('  source to csv'.format(mappings))

            # STAP 2
            sql = "TRUNCATE TABLE {sor}.{temp_table};".format(**params)
            self.execute(sql, 'truncate {}'.format(params['temp_table']))

            # STAP 3 Bron data naar temp
            # we faken de quote voor textvelden opdat json velden (met dubbele quotes) goed worden ingelezen en later eenvoudig zijn te parsen naar jsonb
            sql = "COPY {sor}.{temp_table} ({fields}) FROM  '{file_name}' DELIMITER '{delimiter}' CSV HEADER ENCODING '{encoding}' QUOTE '{quote}';".format(**params)
            self.execute(sql, 'copy into {}'.format(params['temp_table']))

            # STAP 4a
            # sql = """INSERT INTO {sor}.{sor_table}(_runid, _insert_date, _hash, _revision, {fields})
            # SELECT {runid},now(), tmp._hash, 1, {fields}
            # FROM {sor}.{temp_table} tmp
            # WHERE NOT EXISTS (SELECT 1 FROM {sor}.{sor_table} hstg WHERE ({keys_compare}));""".format(**params)
            # self.execute(sql,  'insert new into sor')
            #
            # # STAP 4b
            # sql = """INSERT INTO {sor}.{sor_table}(_runid, _insert_date, _hash, _revision, {fields})
            #     SELECT {runid},now(), tmp._hash, hstg._revision + 1, {tmp_fields}
            #     FROM {sor}.{temp_table} tmp JOIN {sor}.{sor_table} hstg ON ({keys_compare})
            #     WHERE _active = True AND ({fields_compare});""".format(**params)
            # self.execute(sql,  'insert changed into sor')

            sql = """INSERT INTO {sor}.{sor_table}(_runid, _source_system, _insert_date, _hash, _active, {fields})
            SELECT {runid}, '{source_system}', now(), '', True, {fields}
            FROM {sor}.{temp_table} tmp
            WHERE ({key_fields}) NOT IN (SELECT {key_fields} FROM {sor}.{sor_table} hstg WHERE not _valid )
            EXCEPT
            SELECT {runid}, '{source_system}', now(), '', True, {fields}
            FROM {sor}.{sor_table} hstg WHERE _active;""".format(**params)
            self.execute(sql, 'insert new into {}'.format(params['sor_table']))

            # STAP 5 REVISION en ACTIVE
            params['keys_compare'] = mappings.get_keys_compare(source_alias='previous', target_alias='current')
            # oude is nog actief, maar runid is kleiner. Dit is het laatste record
            sql = """update {sor}.{sor_table} current set _revision = previous._revision + 1
                    from {sor}.{sor_table} previous where current._active AND previous._active and previous._valid AND ({keys_compare}) and previous._runid < current._runid;""".format(
                **params)
            self.execute(sql, 'update revision')

            # hierna pas oude inactief zetten
            sql = """update {sor}.{sor_table} previous set _active = False, _finish_date = current._insert_date
                    from {sor}.{sor_table} current where previous._active AND ({keys_compare}) and previous._runid < current._runid;""".format(
                **params)
            self.execute(sql, 'update set old ones inactive')

            # STAP 6 DELETED Markeren
            sql = """update {sor}.{sor_table} set _deleted_runid = {runid}, _active = FALSE, _finish_date = now()
                    WHERE _active AND ({key_fields}) NOT IN (SELECT {key_fields} FROM {sor}.{temp_table})""".format(
                **params)
            self.execute(sql,  'update sor set deleted ones')

        except Exception as ex:
            self.logger.log_error(mappings.name, ex=ex)





    def source_to_sor_by_hash(self, mappings):
        """Eerst worden alleen hash-keys van bron database opgehaald. Deze worden vergeleken met huidige data.
        zijn hashes gewijzigd dan betreft het een gewijzigd record.
        Alleen van deze worden alle gegevens opgehaald"""
        try:
            debug = 'debug' in self.pipe.pipeline.config and self.pipe.pipeline.config['debug']

            params = mappings.__dict__
            params.update(self._get_fixed_params())
            params['fields'] = mappings.get_fields()
            params['key_fields'] = mappings.get_keys()
            params['tmp_fields'] = mappings.get_fields(alias='tmp')
            params['fields_compare'] = mappings.get_fields_compare(source_alias='tmp', target_alias='hstg')
            params['keys_compare'] = mappings.get_keys_compare(source_alias='tmp', target_alias='hstg')

              # STAP 1 data van database in csv file
            if isinstance(mappings.source, SourceTable):
                file_name = mappings.source.to_csv(md5_only=True, filter=mappings.filter, ignore_fields=mappings.ignore_fields, debug=debug)
                params['file_name'] = file_name
            # self.logger.log_simple('    source hash to csv'.format(mappings))

            # STAP 2
            sql = "TRUNCATE TABLE {sor}.{temp_table}_hash;".format(**params)
            self.execute(sql, 'truncate temp')

            # STAP 3 Bron data naar temp_hash
            sql = "COPY {sor}.{temp_table}_hash ({key_fields}, _hash) FROM  '{file_name}' DELIMITER ';' CSV HEADER;".format(
                    **params)
            self.execute(sql, 'copy into temp hash')

            # STAP 4a kijk of sor_table al data bevat zo ja dan wijzigingen bepalen. Zo nee dan alle data ophalen
            sql = "SELECT COUNT(*) FROM {sor}.{sor_table};".format(**params)
            result = self.execute_read(sql, 'get rowcount')
            rowcount = result[0][0]

            # STAP 6
            sql = "TRUNCATE TABLE {sor}.{temp_table};".format(**params)
            self.execute(sql, 'truncate temp')

            filter = ''
            i = 0
            if rowcount > 0:
                # STAP 4 bepaal wijzigingen van temp tov laatste sor data
                # CM : Vergelijk alleen met actieve records.
                sql = "UPDATE {sor}.{temp_table}_hash tmp SET _changed = TRUE WHERE NOT EXISTS (SELECT 1 FROM {sor}.{sor_table} hstg WHERE ({keys_compare}) AND tmp._hash = hstg._hash AND _active = True);".format(
                        **params)
                self.execute(sql, 'set status(changed) of temp')

                # STAP 5 haal alle data op uit bron
                # maak where IN

                sql = """SELECT {key_fields} FROM {sor}.{temp_table}_hash WHERE _changed;""".format(**params)
                # changed_keys = []
                changed_keys_str = ''
                rows = self.dwh.engine.execute(sql)
                # if len(rows) > 1000:
                #     debug = True
                for row in rows:
                    i += 1
                    keys_concat = ''.join(list(row))
                    changed_keys_str += "'{}',".format(keys_concat)
                    if i % 1000 == 0:

                        changed_keys_str = changed_keys_str[:-1]
                        key_concat = params['key_fields'].replace(',', '||')
                        filter = 'WHERE {} IN ({})'.format(key_concat, changed_keys_str)
                        file_name = mappings.source.to_csv(md5_only=False, filter=filter,  ignore_fields=mappings.ignore_fields, debug=debug)
                        params['file_name'] = file_name

                        sql = "COPY {sor}.{temp_table} ({fields}) FROM  '{file_name}' DELIMITER ';' CSV HEADER;".format(
                            **params)

                        self.execute(sql, 'copy into temp row {} - {}'.format(i, i+ 1000))

                        changed_keys_str = ''

                changed_keys_str = changed_keys_str[:-1]
                if changed_keys_str.strip():
                # changed_keys_str = ','.join(changed_keys)
                    key_concat = params['key_fields'].replace(',', '||')
                    filter = 'WHERE {} IN ({})'.format(key_concat, changed_keys_str)
                # self.logger.log_simple('    generated where in sql'.format(mappings))
                else:
                    # CM: zonder gewijzigde records nooit iets ophalen
                    filter= 'WHERE 1=2'
            else:
                filter = mappings.filter

            #STAP 5b
            if isinstance(mappings.source, SourceTable):
                file_name = mappings.source.to_csv(md5_only=False, filter=filter, ignore_fields=mappings.ignore_fields, debug=debug)
                params['file_name'] = file_name
            # self.logger.log_simple('    source complete to csv'.format(mappings))



            # STAP 7 Bron data naar temp
            sql = "COPY {sor}.{temp_table} ({fields}) FROM  '{file_name}' DELIMITER ';' CSV HEADER;".format(**params)
            self.execute(sql, 'copy into temp row {} - end'.format(i))

            # STAP 7a Update _hash
            params['tmp_fields'] = mappings.get_fields(alias='tmp')
            params['keys_compare'] = mappings.get_keys_compare(source_alias='tmp_hash', target_alias='tmp')

            sql = "UPDATE {sor}.{temp_table} tmp set _hash = tmp_hash._hash FROM {sor}.{temp_table}_hash tmp_hash WHERE  {keys_compare};".format(**params)
            self.execute(sql, 'update _hash in temp')

            # STAP 8  insert into sor
            sql = """INSERT INTO {sor}.{sor_table}(_runid, _source_system, _insert_date, _hash, {fields})
                SELECT {runid},'{source_system}', now(), tmp._hash, {tmp_fields}
                FROM {sor}.{temp_table} tmp
                 --JOIN {sor}.{temp_table}_hash tmp_hash ON {keys_compare};""".format(**params)
            self.execute(sql, 'insert new into sor')

            params['keys_compare'] = mappings.get_keys_compare(source_alias='previous', target_alias='current')

            # STAP 9a SET revisienummers
            sql = """UPDATE {sor}.{sor_table} current SET _revision = previous._revision + 1
                        --FROM (SELECT {key_fields}, max(_revision) as _revision, max(_runid) as _runid FROM {sor}.{sor_table} WHERE _active = TRUE GROUP BY {key_fields}) as previous
                        FROM {sor}.{sor_table} previous
                        WHERE current._active = True AND previous._active = True AND previous._runid < current._runid AND {keys_compare};
                        --current._revision = 0 AND {keys_compare};""".format(**params)
            self.execute(sql, 'update sor set _revision')

            # STAP 9b SET actieve records
            sql = """UPDATE {sor}.{sor_table} previous SET _active = False, _finish_date = current._insert_date
                        FROM {sor}.{sor_table} current WHERE previous._active = True AND ({keys_compare}) AND current._revision = (previous._revision + 1);""".format(
                **params)
            self.execute(sql, 'update sor set old ones inactive')


        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def validate_duplicate_keys(self, mappings, sor_schema):
        try:

            validation = SorValidation(tbl=mappings.sor_table, schema=sor_schema)
            validation.msg = 'duplicate key error'
            validation.set_check_for_duplicate_keys(mappings.get_keys())
            self.validate_duplicates(validation)

        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])
            # raise Exception(ex.args[0])

    def validate_duplicate_fks(self, sor_table, sor_schema, fk_name):
        try:
            validation = SorValidation(tbl=sor_table, schema=sor_schema)
            validation.msg = 'duplicate fk error'
            validation.set_check_for_duplicate_keys(fk_name)
            self.validate_duplicates(validation)

        except Exception as ex:
            self.logger.log_error('duplicate check ' + fk_name, err_msg=ex.args[0])
            # raise Exception(ex.args[0])

    def validate_duplicates(self, validation: SorValidation):
        try:
            params = validation.__dict__
            params.update(self._get_fixed_params())
            params['keys'] = validation.get_keys()
            params['sor_table'] = validation.tbl.name

            sql = """UPDATE {sor}.{sor_table} set _validation_msg = '{msg}', _valid = FALSE
                where (_runid, {keys}) in (select _runid, {keys} from {sor}.{sor_table} GROUP BY _runid, {keys} HAVING count(*) > 1);""".format(
                **params)
            self.execute(sql, 'validate sor: mark duplicate keys')

            # 2017-04: uitgezet want anders gaan updates (wijzigingen niet meer goed)
            # sql = """update {sor}.{sor_table} set _valid = TRUE, _validation_msg = '{msg}-first item'
            #     where _validation_msg = '{msg}' AND floor(_runid) = floor({runid}) AND _id in (select DISTINCT ON ({keys}) _id from {sor}.{sor_table} WHERE floor(_runid) = floor({runid}) ORDER BY {keys}, _id);""".format(
            #     **params)
            # self.execute(sql, 'validate sor: mark first items of duplicate keys')

            # self.copy_to_exceptions_table(mappings.source)

        except Exception as ex:
            self.logger.log_error(validation.msg, err_msg=ex.args[0])
            # raise Exception(ex.args[0])

    def validate_sor(self, validation: SorValidation):
        try:
            if validation.check_for_duplicate_keys:
                self.validate_duplicates(validation)
            else:
                params = validation.__dict__
                params.update(self._get_fixed_params())
                params['sor_table'] = validation.tbl.name
                sql = """UPDATE {sor}.{sor_table} set _valid = False, _validation_msg = COALESCE(_validation_msg, '') || '{msg}; '
                    where _runid = {runid} AND {sql_condition};""".format(
                    **params)
                self.execute(sql, 'validate sor: ' + validation.msg)

            # self.copy_to_exceptions_table(validation.tbl)
        except Exception as ex:
            self.logger.log_error(validation.msg, err_msg=ex.args[0]) class EtlSorToDv(BaseEtl):
    def __init__(self, pipe):
        super().__init__(pipe)

    def _get_fixed_params(self) -> Dict[str, Any]:
        params = {}
        params['runid'] = self.runid
        params['source_system'] = self.pipe.source_system
        params['sor'] = self.pipe.sor.name
        return params

    def sor_to_entity(self, mappings: 'SorToEntityMapping'):
        self.logger.log('START <blue>{}</>'.format(mappings), indent_level=3)
        try:
            if not mappings.filter:
                mappings.filter = '1=1'
            params = mappings.__dict__
            dv_schema = mappings.target.cls_get_schema(self.dwh)
            params.update(self._get_fixed_params())
            params['dv_schema'] = dv_schema.name
            params['hub'] = mappings.target.cls_get_hub_name()
            if 'zorgactiviteit' in str(mappings):
                debug = True
            params['hub_type'] = mappings.target.cls_get_hub_type()
            params['filter_type'] = '1=1'
            if mappings.type:
                params['hub_type'] = mappings.type
                params['filter_type'] = "type = '{}'".format(mappings.type)
            params['sor_table'] = mappings.source.name
            params['relation_type'] = mappings.type
            # params['filter_hub'] = params['filter']

            sql = "SELECT COUNT(*) FROM {dv_schema}.{hub} WHERE {filter_type};".format(**params)
            result = self.execute_read(sql, 'get rowcount')
            rowcount_hub = result[0][0]

            if rowcount_hub == 0:
                params['filter_runid'] = '1=1'
            else:
                params['filter_runid'] = 'floor(hstg._runid) = floor({runid})'.format(**params)

            if mappings.bk_mapping and isinstance(mappings.source, SorTable):
                sql = """INSERT INTO {dv_schema}.{hub} (_runid, _insert_date, _source_system, type, bk)
SELECT DISTINCT {runid}, now(), '{source_system}', '{hub_type}', {bk_mapping}
FROM {sor}.{sor_table} hstg
WHERE hstg._valid AND ({bk_mapping}) IS NOT NULL AND NOT EXISTS (SELECT 1 FROM {dv_schema}.{hub} WHERE bk = ({bk_mapping})) AND {filter}
--AND {filter_runid};""".format(
                    **params)
                self.execute(sql, 'insert new '.format(params['hub']))

                # onderstaande regels voor performance
                sql = """SELECT hub._id FROM {dv_schema}.{hub} hub JOIN {sor}.{sor_table} hstg ON {bk_mapping} = hub.bk WHERE hstg._valid AND {filter}
--AND {filter_runid};""".format(
                    **params)
                self.execute(sql, 'load hub_ids in mem (performance)')

                sql = """UPDATE {sor}.{sor_table} hstg SET fk_{relation_type}{hub} = hub._id FROM {dv_schema}.{hub} hub WHERE {bk_mapping} = hub.bk AND hstg._valid AND {filter}
--AND {filter_runid};""".format(
                    **params)
                self.execute(sql, 'update fk_hub in sor table')
            elif mappings.bk_mapping and isinstance(mappings.source, SorQuery):
                params['sql'] = mappings.source.sql
                params['sor_table'] = mappings.source.main_table

                sql = """INSERT INTO {dv_schema}.{hub} (_runid, _insert_date, _source_system, type, bk)
SELECT DISTINCT {runid}, now(), '{source_system}', '{hub_type}', {bk_mapping}
FROM ({sql}) hstg
WHERE hstg._valid AND ({bk_mapping}) IS NOT NULL
AND NOT EXISTS (SELECT 1 FROM {dv_schema}.{hub} hub WHERE hub.bk = {bk_mapping})
AND {filter} AND {filter_runid};""".format( **params)
                self.execute(sql, 'insert new '.format(params['hub']))

                if mappings.source.update_query_fk_hub:
                    sql = mappings.source.update_query_fk_hub
                    self.execute(sql, 'update fk_hub in sor table')
                else:
                    sql = """SELECT * FROM ({sql}) as q""".format(**params)
                    self.execute(sql, 'load in mem performance')

                    sql = """WITH sorquery as (SELECT {bk_mapping} as bk FROM ({sql}) hstg
    WHERE hstg._valid AND {filter} AND {filter_runid})
    UPDATE {sor}.{sor_table} hstg SET fk_{relation_type}{hub} = hub._id FROM {dv_schema}.{hub} hub JOIN sorquery ON sorquery.bk = hub.bk""".format(**params)
                    self.execute(sql, 'update fk_hub in sor table')

            elif mappings.key_mappings:
                target_key_sat = ''
                compare_key_fields = ''
                for key_mapping in mappings.key_mappings:
                    target_key_sat = key_mapping.target.table.name
                    compare_key_fields += "hstg.{} = key_sat.{} AND ".format(key_mapping.source, key_mapping.target)
                compare_key_fields = compare_key_fields[:-5]
                params['target_key_sat'] = target_key_sat
                params['compare_key_fields'] = compare_key_fields
                sql = """UPDATE {sor}.{sor_table} hstg SET fk_{relation_type}{hub} = hub._id FROM {dv_schema}.{hub} hub, {dv_schema}.{target_key_sat} key_sat WHERE hub._id = key_sat._id AND key_sat._active AND {compare_key_fields} AND hstg._valid AND {filter_hub};""".format(
                    **params)
                self.execute(sql, 'update fk_hub in sor table')

            # fk_name = "fk_{relation_type}{hub}".format(**params)
            # EtlSourceToSor(self.pipe).validate_duplicate_fks(mappings.source, dv_schema, fk_name)

            for sat_mappings in mappings.sat_mappings.values():
                self.__sor_to_sat(params, sat_mappings)

            # deletes
            hub_entity = mappings.target  # type: HubEntity
            if hub_entity.cls_has_record_status_sat():
                params['record_satus_sat'] = hub_entity.cls_get_record_status_sat().__dbname__
                sql = """INSERT INTO {dv_schema}.{record_satus_sat} (_id, _runid, _source_system, _insert_date, deleted)
                                    SELECT _id, {runid}, '{source_system}', now(), now()
                                    FROM {dv_schema}.{hub}
        WHERE _id NOT IN (SELECT _id FROM {dv_schema}.{record_satus_sat}) AND
              _id IN (SELECT fk_{relation_type}{hub}
                        FROM sor_test_system.traject_hstage hstg
                        WHERE hstg._deleted_runid = {runid})""".format(**params)
                self.execute(sql, 'update deleted records')
            self.logger.log('FINISH {}'.format(mappings), indent_level=3)
        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def __sor_to_sat(self, params, sat_mappings):
        self.logger.log('START {}'.format(sat_mappings), indent_level=4)
        satparams = sat_mappings.__dict__
        satparams.update(self._get_fixed_params())
        sat_cls = sat_mappings.target
        satparams['dv_schema'] = params['dv_schema']
        satparams['sat'] = sat_cls.cls_get_name()
        if satparams['sat'] == 'afspraak_sat_identificatie':
            bedug = True

        if 'hub' in params:
            satparams['hub_or_link'] = params['hub']
            satparams['relation_type'] = params['relation_type']
        else:
            satparams['hub_or_link'] = params['link']
            satparams['relation_type'] = ''
        satparams['filter'] = params['filter']
        satparams['filter_runid'] = params['filter_runid']
        satparams['sor_fields'] = sat_mappings.get_source_fields(alias='hstg')
        satparams['sat_fields'] = sat_mappings.get_sat_fields()
        satparams['fields_compare'] = sat_mappings.get_fields_compare(source_alias='hstg', target_alias='sat')
        satparams['from'] = "{sor}.{sor_table} AS hstg".format(**params)
        if isinstance(sat_mappings.source, SorQuery):
            satparams['from'] = "({}) AS hstg".format(sat_mappings.source.sql)

        # sql = "SELECT COUNT(*) FROM {dv_schema}.{sat};".format(**satparams)
        # result = self.execute_read(sql, 'get rowcount')
        # rowcount_sat = result[0][0]
        #
        # if rowcount_sat == 0:
        #     satparams['filter_runid'] = '1=1'
        # else:
        #     satparams['filter_runid'] = 'floor(hstg._runid) = floor({runid})'.format(**params)

        if sat_cls.__base__ == HybridSat:
            sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, type, _source_system, _insert_date, _revision, {sat_fields})
                    SELECT DISTINCT ON(fk_{relation_type}{hub_or_link})  fk_{relation_type}{hub_or_link}, {runid}, '{type}', '{source_system}', now(), 0, {sor_fields}
                    FROM {from} WHERE hstg._valid AND hstg._active AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND {filter}
                    --AND NOT EXISTS (SELECT 1 FROM {dv_schema}.{sat} sat where sat._id = fk_{relation_type}{hub_or_link} and sat._runid = {runid} AND type = '{type}')
                    --AND {filter_runid}
                    EXCEPT
                    SELECT _id, {runid}, '{type}', '{source_system}', now(), 0, {sat_fields}
                    FROM {dv_schema}.{sat} sat
                    WHERE _active AND type = '{type}';""".format(
                **satparams)
            self.execute(sql, 'insert new in sat')

            # oude is nog actief, maar runid is kleiner. Dit is het laatste record
            sql = """update {dv_schema}.{sat} current set _revision = previous._revision + 1
                    from {dv_schema}.{sat} previous where current._active = True AND previous._active = True AND previous._id = current._id and previous._runid < current._runid AND current.type = '{type}' AND previous.type = '{type}';""".format(
                **satparams)
            self.execute(sql, 'update sat revision')
            # nu oude inctief maken
            sql = """update {dv_schema}.{sat} previous set _active = False, _finish_date = current._insert_date
                    from {dv_schema}.{sat} current where previous._active = True AND previous._id = current._id and previous._runid < current._runid AND current.type = '{type}' AND previous.type = '{type}';""".format(
                **satparams)
            self.execute(sql, 'update sat set old ones inactive')

            self.logger.log('FINISH {}'.format(sat_mappings), indent_level=4)
        else:
            # satparams['from'] = "{sor}.{sor_table} AS hstg".format(**params)
            # if isinstance(sat_mappings.source, SorQuery):
            #     satparams['from'] = "({}) AS hstg".format(sat_mappings.source.sql)


            sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                    SELECT DISTINCT ON(fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{source_system}', now(), 0, {sor_fields}
                    FROM {from} WHERE hstg._valid AND hstg._active AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND {filter}
                    --AND NOT EXISTS (SELECT 1 FROM {dv_schema}.{sat} sat where sat._id = fk_{relation_type}{hub_or_link} and sat._runid = {runid})
                    -- AND {filter_runid}
                    EXCEPT
                    SELECT _id, {runid}, '{source_system}', now(), 0, {sat_fields}
                    FROM {dv_schema}.{sat} sat
                    WHERE sat._active;""".format(
                **satparams)
            self.execute(sql, 'insert new in sat')

            # oude is nog actief, maar runid is kleiner. Dit is het laatste record
            sql = """update {dv_schema}.{sat} current set _revision = previous._revision + 1
                    from {dv_schema}.{sat} previous where current._active = True AND previous._active = True AND previous._id = current._id and previous._runid < current._runid;""".format(
                **satparams)
            self.execute(sql, 'update sat revision')
            # nu oude inctief maken
            sql = """update {dv_schema}.{sat} previous set _active = False, _finish_date = current._insert_date
                    from {dv_schema}.{sat} current where previous._active = True AND previous._id = current._id and previous._runid < current._runid;""".format(
                **satparams)
            self.execute(sql, 'update sat set old ones inactive')

            self.logger.log('FINISH {}'.format(sat_mappings), indent_level=4)

    def __sor_to_sat_old(self, params, sat_mappings):
        self.logger.log('    START {}'.format(sat_mappings))
        satparams = sat_mappings.__dict__
        satparams.update(self._get_fixed_params())
        sat_cls = sat_mappings.target
        if 'hl7' in sat_cls.cls_get_name():
            debug = True
        satparams['sat'] = sat_cls.cls_get_name()
        if 'hub' in params:
            satparams['hub_or_link'] = params['hub']
            satparams['relation_type'] = params['relation_type']
        else:
            satparams['hub_or_link'] = params['link']
            satparams['relation_type'] = ''
        satparams['filter'] = params['filter']
        satparams['sor_fields'] = sat_mappings.get_source_fields(alias='hstg')
        satparams['sat_fields'] = sat_mappings.get_sat_fields()
        satparams['fields_compare'] = sat_mappings.get_fields_compare(source_alias='hstg', target_alias='sat')
        sql = "SELECT COUNT(*) FROM {dv_schema}.{sat};".format(**satparams)
        result = self.execute_read(sql, 'get rowcount')
        rowcount = result[0][0]
        if sat_cls.__base__ == HybridSat:
            if rowcount == 0:
                # alle zijn nieuw; geen rekening houden met _runid
                sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, type, _source_system, _insert_date, _revision, {sat_fields})
                        SELECT DISTINCT ON (fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{type}', '{source_system}', now(), 1, {sor_fields}
                        FROM {sor}.{sor_table} hstg
                        WHERE hstg._valid AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND NOT EXISTS (select 1 from {dv_schema}.{sat} sat where sat._id  =  _fk_{relation_type}{hub_or_link} AND sat.type = '{type}') AND {filter};""".format(
                    **satparams)
                self.execute(sql, '  insert new in sat')
            else:
                sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, type, _source_system, _insert_date, _revision, {sat_fields})
                        SELECT DISTINCT ON (fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{type}', '{source_system}', now(), 1, {sor_fields}
                        FROM {sor}.{sor_table} hstg
                        WHERE floor(hstg._runid) = floor({runid}) AND hstg._valid AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND NOT EXISTS (select 1 from {dv_schema}.{sat} sat where sat._id  =  fk_{relation_type}{hub_or_link} AND sat.type = '{type}') AND {filter};""".format(
                    **satparams)
                self.execute(sql, '  insert new in sat')

                sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                        SELECT DISTINCT ON (fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{source_system}', now(), sat._revision + 1, {sor_fields}
                        FROM {sor}.{sor_table} hstg JOIN {dv_schema}.{sat} sat ON sat._id =  fk_{relation_type}{hub_or_link}
                        WHERE floor(hstg._runid) = floor({runid}) AND hstg._valid AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND sat._active = True AND sat.type = '{type}' AND ({fields_compare}) AND {filter};""".format(
                    **satparams)
                self.execute(sql, '  insert changed in sat')

                sql = """UPDATE {dv_schema}.{sat} previous SET _active = FALSE, _finish_date = current._insert_date
                        FROM {dv_schema}.{sat} current WHERE previous._active = TRUE AND previous._id = current._id AND current._revision = (previous._revision + 1) AND current.type = '{type}' AND previous.type = '{type}' ;""".format(
                    **satparams)
                self.execute(sql, '  update sat set old ones inactive')
                self.logger.log('    FINISH {}'.format(sat_mappings))
        else:

            if rowcount == 0:
                # alle zijn nieuw; geen rekening houden met _runid
                sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                        SELECT DISTINCT ON (fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{source_system}', now(), 1, {sor_fields}
                        FROM {sor}.{sor_table} hstg
                        WHERE hstg._valid AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND NOT EXISTS (select 1 from {dv_schema}.{sat} sat where sat._id  =  _fk_{relation_type}{hub_or_link}) AND {filter};""".format(
                    **satparams)
                self.execute(sql, '  insert new in sat')
            else:
                sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                        SELECT DISTINCT ON (fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{source_system}', now(), 1, {sor_fields}
                        FROM {sor}.{sor_table} hstg
                        WHERE floor(hstg._runid) = floor({runid}) AND hstg._valid AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND NOT EXISTS (select 1 from {dv_schema}.{sat} sat where sat._id  =  _fk_{relation_type}{hub_or_link}) AND {filter};""".format(
                    **satparams)
                self.execute(sql, '  insert new in sat')

                sql = """INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                        SELECT DISTINCT ON (fk_{relation_type}{hub_or_link}) fk_{relation_type}{hub_or_link}, {runid}, '{source_system}', now(), sat._revision + 1, {sor_fields}
                        FROM {sor}.{sor_table} hstg JOIN {dv_schema}.{sat} sat ON sat._id =  fk_{relation_type}{hub_or_link}
                        WHERE floor(hstg._runid) = floor({runid}) AND hstg._valid AND hstg.fk_{relation_type}{hub_or_link} IS NOT NULL AND sat._active = True AND ({fields_compare}) AND {filter};""".format(
                    **satparams)
                self.execute(sql, '  insert changed in sat')

                sql = """UPDATE {dv_schema}.{sat} previous SET _active = FALSE, _finish_date = current._insert_date
                        FROM {dv_schema}.{sat} current WHERE previous._active = TRUE AND previous._id = current._id AND current._revision = (previous._revision + 1);""".format(
                    **satparams)
                self.execute(sql, '  update sat set old ones inactive')
                self.logger.log('    FINISH {}'.format(sat_mappings))

    def sor_to_link(self, mappings):
        self.logger.log('  START {}'.format(mappings))
        try:
            if not mappings.filter:
                mappings.filter = '1=1'
            params = mappings.__dict__
            params.update(self._get_fixed_params())
            dv_schema = mappings.target.cls_get_schema(self.dwh)
            params['dv_schema'] = dv_schema.name
            params['link'] = mappings.target.Link.cls_get_name()
            params['link_type'] = mappings.type
            params['sor_table'] = mappings.source.name
            params['source_fks'] = self.__get_link_source_fks(mappings)
            params['source_fks_is_not_null'] = self.__get_link_source_fks_is_not_null(mappings)
            params['source_fks_is_null'] = params['source_fks_is_not_null'].replace('NOT ', '')
            params['target_fks'] = self.__get_link_target_fks(mappings)

            params['join'] = self.__get_link_join(mappings, schema_name=self.pipe.sor.name)
            params['fks_compare'] = self.__get_link_fks_compare(mappings, source_alias='hstg', target_alias='link')

            sql = "SELECT COUNT(*) FROM {dv_schema}.{link};".format(**params)
            result = self.execute_read(sql, 'get rowcount')
            rowcount_link = result[0][0]

            if rowcount_link == 0:
                params['filter'] = 'hstg._active'
            else:
                params['filter'] = 'floor(hstg._runid) = floor({runid})'.format(**params)

            sql = """
            INSERT INTO {dv_schema}.{link} (_runid, _source_system, _insert_date, type, {target_fks})
            SELECT DISTINCT {runid}, '{source_system}', now(), '{link_type}', {source_fks}
            FROM {sor}.{sor_table} hstg {join}
            WHERE {filter}
            AND NOT ({source_fks_is_null})
            AND NOT EXISTS (SELECT 1 FROM {dv_schema}.{link} link WHERE {fks_compare} AND link.type='{link_type}') AND {filter};""".format(**params)
            self.execute(sql,  'insert new links')

            if len(mappings.sat_mappings) > 0:
                # todo refactor
                from_sql = ''
                where_sql = ''
                for field_mapping in mappings.field_mappings:
                    join = field_mapping.join
                    if field_mapping.bk:
                        join_tbl = field_mapping.get_source_table()
                        from_sql += ', dv.{}'.format(join_tbl)
                        where_sql += 'dv.{}.bk = {} AND '.format(join_tbl, field_mapping.bk)
                # from_sql = from_sql[:-1]
                where_sql = where_sql[:-5]

                params['from'] = from_sql
                params['where'] = where_sql

                sql = """UPDATE {sor}.{sor_table} hstg SET fk_{type}{link} = link._id
FROM {dv_schema}.{link} link {from}
WHERE {fks_compare} AND {where}
AND hstg._valid AND {filter};""".format(
                    **params)

                self.execute(sql, 'update fk_link in sor table')

            for sat_mappings in mappings.sat_mappings.values():
                self.__sor_to_sat(params, sat_mappings)

            #deletes
            link_entity = mappings.target #type: LinkEntity
            if link_entity.cls_has_record_status_sat():
                params['record_satus_sat'] = link_entity.cls_get_record_status_sat().__dbname__
                sql = """INSERT INTO {dv_schema}.{record_satus_sat} (_id, _runid, _source_system, _insert_date, deleted)
                            SELECT _id, {runid}, '{source_system}', now(), now()
                            FROM {dv_schema}.{link}
WHERE _id NOT IN (SELECT _id FROM {dv_schema}.{record_satus_sat}) AND
      ({target_fks}) IN (SELECT {source_fks}
                            FROM sor_test_system.traject_hstage hstg
                              {join}
  WHERE hstg._deleted_runid = {runid})""".format(**params)
                self.execute(sql, 'update deleted records')




            self.logger.log('  FINISH {}'.format(mappings))
        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def __get_link_source_fks(self, mappings: 'SorToLinkMapping'):
        fks = ''
        for field_mapping in mappings.field_mappings:
            if field_mapping.source.table == mappings.sor_table_name:
                fks += '{}, '.format(field_mapping.source)
                # fks += 'hstg.{}, '.format(field_mapping.source)
            else:
                # fks += '{}.{}, '.format(field_mapping.source.table, field_mapping.source)
                fks += '{}.{}, '.format(field_mapping.source_alias, field_mapping.source)
        fks = fks[:-2]
        return fks

    def __get_link_source_fks_is_not_null(self, mappings: 'SorToLinkMapping'):
        fks = ''
        for field_mapping in mappings.field_mappings:
            if field_mapping.source.table == mappings.sor_table_name:
                fks += '{} IS NOT NULL AND '.format(field_mapping.source)
            else:
                fks += '{}.{} IS NOT NULL AND '.format(field_mapping.source_alias, field_mapping.source)
        fks = fks[:-5]
        return fks

    def __get_link_target_fks(self, mappings: 'SorToLinkMapping'):
        fks = ''
        for field_mapping in mappings.field_mappings:
            fks += '{}, '.format(field_mapping.target)
        fks = fks[:-2]
        return fks

    def __get_link_fks_compare(self, mappings: 'SorToLinkMapping', source_alias='', target_alias=''):
        if not source_alias: source_alias = 'hstg'
        if not target_alias: target_alias = 'link'
        fks_compare = ''
        for field_mapping in mappings.field_mappings:
            if isinstance(field_mapping.source, ConstantValue):
                fks_compare += 'COALESCE({0}, 0) = COALESCE({1}.{2}, 0) AND '.format(field_mapping.source, target_alias, field_mapping.target)
            elif field_mapping.is_view_mapping:
                fks_compare += 'COALESCE({0}.{1}, 0) = COALESCE({2}.{3}, 0) AND '.format(field_mapping.source.table, field_mapping.source, target_alias, field_mapping.target)
            else:
                source_alias = self.__get_link_alias_of_source_tbl(field_mapping, fks_compare)
                fks_compare += 'COALESCE({0}.{2}, 0) = COALESCE({1}.{3}, 0) AND '.format(source_alias, target_alias, field_mapping.source, field_mapping.target)
        fks_compare = fks_compare[:-4]
        return fks_compare

    def __get_link_join(self, mappings, schema_name='sor'):
        #todo robuuster maken met aliassen en ands
        join_sql = ''
        for field_mapping in mappings.field_mappings:
            join = field_mapping.join
            bk = field_mapping.bk
            if join:
                join_tbl = field_mapping.get_source_table()
                if field_mapping.is_view_mapping:
                    schema_name = 'dv'
                    join_sql += ' INNER JOIN {}.{} ON {} \r\n'.format(schema_name, join_tbl, join)
                else:
                    join_alias = self.__get_link_alias_of_source_tbl(field_mapping, join_sql)
                    join = join.replace(join_tbl + '.', join_alias + '.')
                    join_sql += ' INNER JOIN {}.{} {} ON {} \r\n'.format(schema_name, join_tbl, join_alias, join)
            elif bk:
                join_tbl = field_mapping.get_source_table()
                join_alias = self.__get_link_alias_of_source_tbl(field_mapping, join_sql)
                join_sql += ' LEFT JOIN dv.{0} AS {1} ON {1}.bk = {2} \r\n'.format(join_tbl, join_alias, bk)
        return join_sql

    def __get_link_alias_of_source_tbl(self, field_mapping, total_sql, default_alias='hstg'):
        #bij een join in de sql moet elke bron tabel een eigen alias krijgen
        #dat wordt hier bepaald
        alias = default_alias
        if field_mapping.join:
            join_tbl = field_mapping.get_source_table()
            alias = join_tbl[:3] + '_hstg'
            index = 1
            while alias in total_sql:
                alias = alias + str(index)
                index += 1
        if field_mapping.bk:
            # join_tbl = field_mapping.get_source_table()
            alias = field_mapping.source_alias
        return alias

    def sor_to_valuesets(self, mappings: SorToValueSetMapping):
        self.logger.log('  START {}'.format(mappings))
        try:
            if not mappings.filter:
                mappings.filter = '1=1'
            params = mappings.__dict__
            params.update(self._get_fixed_params())
            params['valset_schema'] = self.pipeline.dwh.valset.name
            params['valset_table'] = mappings.target.cls_get_name()
            params['valset_fields'] =  mappings.get_target_fields()
            params['sor_table'] = mappings.source.name
            params['sor_fields'] = mappings.get_source_fields('hstg')
            params['from'] = "{sor}.{sor_table} AS hstg".format(**params)

            if isinstance(mappings.source, SorQuery):
                params['from'] = "({}) as hstg".format(mappings.source.sql)

            insert_sql = """INSERT INTO {valset_schema}.{valset_table} (_runid, _source_system, _insert_date, _revision, {valset_fields})
                                SELECT  {runid}, '{source_system}', now(), 0, {sor_fields}
                                FROM {from} WHERE hstg._valid AND hstg._active
                                AND {filter}
                                --AND NOT EXISTS (SELECT 1 FROM {valset_schema}.{valset_table} valset where valset.code = hst.code and valset._runid = {runid})
                                --AND filter_runid
                                EXCEPT
                                SELECT {runid}, '{source_system}', now(), 0, {valset_fields}
                                FROM {valset_schema}.{valset_table} valset
                                WHERE valset._active;""".format(
                **params)
            self.execute(insert_sql, 'insert valuesets')

            # oude is nog actief, maar runid is kleiner. Dit is het laatste record
            update_sql = """update {valset_schema}.{valset_table} current set _revision = previous._revision + 1
                    from {valset_schema}.{valset_table} previous where current._active = True AND previous._active = True AND previous.valueset_naam = current.valueset_naam AND previous.code = current.code and previous._runid < current._runid;""".format(
                **params)
            self.execute(update_sql, 'update valset revision')
            # nu oude inctief maken
            update_sql = """update {valset_schema}.{valset_table} previous set _active = False, _finish_date = current._insert_date
                    from {valset_schema}.{valset_table} current where previous._active = True AND previous.valueset_naam = current.valueset_naam AND previous.code = current.code and previous._runid < current._runid;""".format(
                **params)
            self.execute(update_sql, 'update valueset_code set old ones inactive')
            self.logger.log('  FINISH {}'.format(mappings))
        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def sor_to_valuesets_old(self, mappings: SorToValueSetMapping):
        self.logger.log('  START {}'.format(mappings))
        try:
            params = mappings.__dict__
            params.update(self._get_fixed_params())
            params['valset_schema'] = self.pipeline.dwh.valset.name

            if isinstance(mappings.source, dict):
                values = ''
                for code, descr in mappings.source.items():
                    params['code'] = code
                    params['descr'] = descr
                    values += "('{code}', '{descr}'),\r\n".format(**params)
                values = values [:-3]
                params['values'] = values
                insert_sql = """

                    CREATE TEMP TABLE _valueset_code_temp (code text, weergave_naam text);

                    INSERT INTO _valueset_code_temp (code, weergave_naam)
                    VALUES {values};

                    INSERT INTO {valset_schema}.valueset (_runid, _source_system, _insert_date, naam, oid)
                    SELECT {runid}, '{source_system}', now(), '{valueset}', NULL
                    WHERE NOT EXISTS (SELECT 1 FROM {valset_schema}.valueset sets WHERE sets.naam = '{valueset}');

                    INSERT INTO {valset_schema}.valueset_code (_runid, _active, _source_system, _insert_date, _revision, valueset_naam, code, weergave_naam)
                    SELECT DISTINCT {runid}, True, '{source_system}', now(), 0, '{valueset}', code, weergave_naam
                    FROM _valueset_code_temp tmp
                    WHERE
                      NOT EXISTS (SELECT 1 FROM {valset_schema}.valueset_code code WHERE code.valueset_naam = '{valueset}' AND code.code = tmp.code AND code.weergave_naam = tmp.weergave_naam);


                    DROP TABLE _valueset_code_temp;
                      """.format(**params)
            elif mappings.source_type_field:
                insert_sql = """
                    INSERT INTO {valset_schema}._ref_valuesets (_runid, _source_system, _insert_date, naam, oid)
                    SELECT DISTINCT {runid}, '{source_system}', now(), {source_type_field}, {source_oid_field}
                    FROM {sor}.{sor_table} hstg
                    WHERE floor(hstg._runid) = floor({runid})
                      AND hstg._valid AND hstg._active
                      AND NOT EXISTS (SELECT 1 FROM {valset_schema}._ref_valuesets sets WHERE sets.naam = hstg.{source_type_field});

                    INSERT INTO {valset_schema}._ref_values (_runid, _active, _source_system, _insert_date, _revision, valueset_oid, valueset_naam, code, weergave_naam, niveau, niveau_type)
                    SELECT DISTINCT {runid}, True, '{source_system}', now(), 0, {source_oid_field}, {source_type_field}, {source_code_field}, {source_descr_field}, {source_level_field}, {source_leveltype_field}
                    FROM {sor}.{sor_table} hstg
                    WHERE floor(hstg._runid) = floor({runid})
                      AND hstg._valid AND hstg._active
                      AND NOT EXISTS (SELECT 1 FROM {valset_schema}._ref_values ref WHERE ref.valueset_naam = hstg.{source_type_field} AND ref.code = hstg.{source_code_field}
                                AND ref.weergave_naam = hstg.{source_descr_field} AND ref.niveau = {source_level_field});""".format(**params)
            elif mappings.source_code_field and not mappings.source_descr_field:
                insert_sql = """
                    INSERT INTO {valset_schema}._ref_valuesets (_runid, _source_system, _insert_date, naam, oid)
                    SELECT {runid}, '{source_system}', now(), '{valueset}', NULL
                    WHERE NOT EXISTS (SELECT 1 FROM {dv_schema}._ref_valuesets sets WHERE sets.naam = '{valueset}');

                    INSERT INTO {valset_schema}._ref_values (_runid, _active, _source_system, _insert_date, _revision, valueset_naam, code, weergave_naam, niveau, niveau_type)
                    SELECT DISTINCT {runid}, True, '{source_system}', now(), 0, '{valueset}', {source_code_field}, NULL, NULL, NULL,
                    FROM {sor}.{sor_table} hstg
                    WHERE floor(hstg._runid) = floor({runid})
                      AND hstg._valid AND hstg._active
                      AND NOT EXISTS (SELECT 1 FROM {valset_schema}._ref_values ref WHERE ref.valueset_naam = '{valueset}' AND ref.code = hstg.{source_code_field}
                                    AND ref.weergave_naam = hstg.{source_descr_field} AND ref.niveau = {source_level_field} AND ref.niveau_type = {source_leveltype_field)};""".format(**params)
            else:
                insert_sql = """
                    INSERT INTO {valset_schema}._ref_valuesets (_runid, _source_system, _insert_date, naam, oid)
                    SELECT {runid}, '{source_system}', now(), '{valueset}', NULL
                    WHERE NOT EXISTS (SELECT 1 FROM {dv_schema}._ref_valuesets sets WHERE sets.naam = '{valueset}');

                    INSERT INTO {valset_schema}._ref_values (_runid, _active, _source_system, _insert_date, _revision, valueset_naam, code, weergave_naam, niveau, niveau_type))
                    SELECT DISTINCT {runid}, True, '{source_system}', now(), 0, '{valueset}', {source_code_field}, {source_descr_field}, {source_level_field}, {source_leveltype_field}
                    FROM {sor}.{sor_table} hstg
                    WHERE floor(hstg._runid) = floor({runid})
                      AND hstg._valid AND hstg._active
                      AND NOT EXISTS (SELECT 1 FROM {valset_schema}._ref_values ref WHERE ref.valueset_naam = '{valueset}' AND ref.code = hstg.{source_code_field}
                                    AND ref.weergave_naam = {source_descr_field} AND ref.niveau = {source_level_field} AND ref.niveau_type = {source_leveltype_field);""".format(**params)

            self.execute(insert_sql, 'insert valuesets')

            # oude is nog actief, maar runid is kleiner. Dit is het laatste record
            update_sql = """update {valset_schema}.valueset_code current set _revision = previous._revision + 1
                    from {valset_schema}.valueset_code previous where current._active = True AND previous._active = True AND previous.valueset_naam = current.valueset_naam AND previous.code = current.code and previous._runid < current._runid;""".format(
                **params)
            self.execute(update_sql, 'update valset revision')
            # nu oude inctief maken
            update_sql = """update {valset_schema}.valueset_code previous set _active = False, _finish_date = current._insert_date
                    from {valset_schema}.valueset_code current where previous._active = True AND previous.valueset_naam = current.valueset_naam AND previous.code = current.code and previous._runid < current._runid;""".format(
                **params)
            self.execute(update_sql, 'update valueset_code set old ones inactive')
            self.logger.log('  FINISH {}'.format(mappings))
        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def view_to_entity(self, mappings):
        self.logger.log('  START {}'.format(mappings))
        try:
            if not mappings.filter:
                mappings.filter = '1=1'
            params = mappings.__dict__
            params.update(self._get_fixed_params())

            sql = """
          INSERT INTO {dv_schema}.{hub} (_runid, _insert_date, _source_system, type, bk)
          SELECT DISTINCT {runid}, now(), '{source_system}', '{type}', {bk_mapping} FROM {dv_schema}.{view} view
          WHERE view._valid AND {bk_mapping} NOT IN (SELECT bk FROM {dv_schema}.{hub}) AND {filter};""".format(
                **params)
            self.execute(sql,  'insert new hub')

            # sql = """SELECT hub._id FROM {dv_schema}.{hub} hub JOIN {dv_schema}.{sor_table} view ON {bk_mapping} = hub.bk WHERE floor(view._runid) = floor({runid}) AND view._valid AND {filter};""".format(
            #     **params)
            # self.execute(sql,  'load hub_ids in mem (performance)')

            # sql = """UPDATE {dv_schema}.{sor_table} view SET _fk_{type}{hub} = hub._id FROM {dv_schema}.{hub} hub WHERE {bk_mapping} = hub.bk AND floor(view._runid) = floor({runid}) AND view._valid AND {filter};""".format(
            #     **params)
            # self.execute(sql,  'update fk_hub in sor table')

            for sat_mappings in mappings.sat_mappings.values():
                self.logger.log('    START {}'.format(sat_mappings))

                satparams = sat_mappings.__dict__
                satparams.update(self._get_fixed_params())
                satparams['view'] = sat_mappings.source
                satparams['type'] = params['type']
                satparams['filter'] = params['filter']
                satparams['bk_mapping'] = params['bk_mapping']
                satparams['view_fields'] = sat_mappings.get_source_fields(alias='view')
                satparams['sat_fields'] = sat_mappings.get_sat_fields()
                satparams['fields_compare'] = sat_mappings.get_fields_compare(source_alias='view', target_alias='sat')

                sql = """
                INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                SELECT DISTINCT ON (hub._id) hub._id, {runid}, '{source_system}', now(), 1, {view_fields}
                FROM {dv_schema}.{view} view
                JOIN {dv_schema}.{hub} hub ON hub.bk = {bk_mapping}
                WHERE view._valid AND NOT EXISTS (select 1 from {dv_schema}.{sat} sat where sat._id  =  hub._id) AND {filter};""".format(
                    **satparams)
                self.execute(sql,  '  insert new in sat')

                sql = """
                INSERT INTO {dv_schema}.{sat} (_id, _runid, _source_system, _insert_date, _revision, {sat_fields})
                SELECT DISTINCT ON (hub._id) hub._id, {runid}, '{source_system}', now(), sat._revision + 1, {view_fields}
                FROM {dv_schema}.{view} view
                JOIN {dv_schema}.{hub} hub ON hub.bk = view.bk
                JOIN {dv_schema}.{sat} sat ON sat._id =  hub._id
                WHERE view._valid AND sat._active = True AND ({fields_compare}) AND {filter};""".format(
                    **satparams)
                self.execute(sql,  '  insert changed in sat')

                sql = """
                UPDATE {dv_schema}.{sat} previous SET _active = FALSE, _finish_date = current._insert_date
                FROM {dv_schema}.{sat} current WHERE previous._active = TRUE AND previous._id = current._id AND current._revision = (previous._revision + 1);""".format(
                    **satparams)
                self.execute(sql,  '  update sat set old ones inactive')
                self.logger.log('    FINISH {}'.format(sat_mappings))
            self.logger.log('  FINISH {}'.format(mappings))
        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def view_to_link(self, mappings):
        self.logger.log('  START {}'.format(mappings))
        try:
            if not mappings.filter:
                mappings.filter = '1=1'
            params = mappings.__dict__
            params.update(self._get_fixed_params())
            params['source_fks'] = mappings.get_source_fks()
            params['target_fks'] = mappings.get_target_fks()
            params['from'] = mappings.get_from()
            params['join'] = mappings.get_join()
            params['fks_compare'] = mappings.get_fks_compare(target_alias='link')

            sql = """
            INSERT INTO {dv_schema}.{link} (_runid, _source_system, _insert_date, {target_fks})
            SELECT {runid}, '{source_system}', now(), {source_fks}
            FROM {dv_schema}.{view}  {join}
            WHERE floor({dv_schema}.{view}._runid) = floor({runid}) AND
              NOT EXISTS (SELECT 1 FROM {dv_schema}.{link} link WHERE {fks_compare}) AND {filter};""".format(**params)

            # sql = """
            # INSERT INTO {dv_schema}.{link} (_runid, _source_system, _insert_date, {target_fks})
            # SELECT {runid}, '{source_system}', now(), {source_fks}
            # FROM {from}
            # WHERE {join} AND floor({dv_schema}.{view}._runid) = floor({runid}) AND
            #   NOT EXISTS (SELECT 1 FROM {dv_schema}.{link} link WHERE {fks_compare}) AND {filter};""".format(**params)

            self.execute(sql,  'insert new links')
            self.logger.log('  FINISH {}'.format(mappings))
        except Exception as ex:
            self.logger.log_error(mappings.name, err_msg=ex.args[0])

    def validate_dv(self, validation: DvValidation):
        try:
            params = validation.__dict__
            params.update(self._get_fixed_params())
            params['dv_schema'] = validation.table.__dbschema__
            params['table'] = validation.table.name

            sql = """UPDATE {dv_schema}.{table} set _valid = False, _validation_msg = COALESCE(_validation_msg, '') || '{msg}; '
                where {sql_condition};""".format(
                **params)
            self.execute(sql, 'validate dv: ' + validation.msg)

            self.copy_to_exceptions_table(validation.table, self.dwh.dv)
        except Exception as ex:
            self.logger.log_error(validation.msg, err_msg=ex.args[0])
            # raise Exception(ex.args[0]) class SourceDatabase(Database):
    def __init__(self, conn_string='', default_schema='public', temp_datatransfer_path=''):
        super().__init__(conn_string, default_schema)
        self.driver = self.get_driver_from_conn_string(conn_string)
        self.user_name = self.get_user_name_from_conn_string(conn_string)
        self.temp_datatransfer_path = temp_datatransfer_path


    def get_driver_from_conn_string(self, conn_string):
        driver = DBDrivers.POSTGRESS
        if conn_string.startswith('ora'):
            driver = DBDrivers.ORACLE
        elif conn_string.startswith('mssql'):
            driver = DBDrivers.SQLSERVER
        elif conn_string.startswith('postgres'):
            driver = DBDrivers.POSTGRESS
        elif conn_string.startswith('mysql'):
            driver = DBDrivers.MYSQL
        return driver

    def get_user_name_from_conn_string(self, conn_string):
        user_name = conn_string.split(':')[1]
        user_name = user_name.replace('/', '')
        return user_name

    def get_or_create_datatransfer_path(self):
        # path = pyelt_config['datatransfer_path']
        path = self.temp_datatransfer_path
        if not path:
            from main import get_root_path
            path = get_root_path() + '/data/transfer/'
        path += '/' + self.name
        if not os.path.exists(path):
            os.makedirs(path)
        return path

    def execute_read(self, sql, log_message=''):
        if self.driver == DBDrivers.POSTGRESS:
            return super().execute_read(sql, log_message)
        connection = self.engine.raw_connection()
        cursor = connection.cursor()
        cursor.execute(sql)
        result = cursor.fetchall()
        cursor.close()
        return result
import datetime
import inspect
import math
        import subprocess, os
                import os
                import os
from collections import OrderedDict
from typing import List
from pyelt.datalayers.database import * #Schema, DbFunction
from pyelt.datalayers.dv import * #HubEntity, Link, EnsembleView, HybridLink, DvValueset
from pyelt.datalayers.dwh import Dwh, DwhLayerTypes
from pyelt.datalayers.sys import *
from pyelt.datalayers.valset import DvValueset
from pyelt.helpers.pyelt_logging import Logger, LoggerTypes
from pyelt.helpers.validations import DomainValidator, MappingsValidator
from pyelt.mappings.sor_to_dv_mappings import SorToValueSetMapping, EntityViewToEntityMapping, EntityViewToLinkMapping, SorToEntityMapping, SorToLinkMapping
from pyelt.mappings.source_to_sor_mappings import SourceToSorMapping
from pyelt.mappings.validations import SorValidation, DvValidation, Validation
from pyelt.process.ddl import * # DdlSor, DdlDv, Ddl, DdlDatamart
from pyelt.process.etl import EtlSourceToSor, EtlSorToDv
from pyelt.sources.databases import SourceDatabase
        from main import get_root_path
            from sys import platform
            from main import get_root_path
            from sys import platform
        """
        ddl = Ddl(self, self.sor)
        ddl.create_or_alter_functions(self.db_functions)

    def run_extra_sql(self):
        """
        Runt de aanwezige extra sql code.
        """
        ddl = Ddl(self, self.sor)
        for sql in self.extra_ddl_statements:
            if sql[0]:
                ddl.execute(sql[1], "EXTRA SQL STATEMENT")

    def run(self, parts = ['sor', 'valuesets', 'hubs', 'links', 'views', 'viewlinks']):
        """
        :param parts: lijst van keywords. De eventuele aanwezigheid van een keyword in deze lijst bepaald of het log bestand dat hoort bij dit keyword gemaakt wordt en of de bijhorende DDL uitgevoerd wordt.

        """

        for func in self.run_before_sor:
            func(self)

        if 'sor' in parts:
            #SOR
            etl = EtlSourceToSor(self)
            self.pipeline.logger.log('START FROM SOURCE TO SOR', indent_level=1)

            for mapping in self.mappings:
                if isinstance(mapping, SourceToSorMapping):
                    self.pipeline.logger.log('START <blue>{}</>'.format(mapping), indent_level=3)
                    etl.source_to_sor(mapping)
                    etl.validate_duplicate_keys(mapping, self.sor)
                    self.pipeline.logger.log('FINISH <blue>{}</>'.format(mapping), indent_level=3)
            for validation in self.validations:
                if isinstance(validation, SorValidation):
                    etl.validate_sor(validation)

            for func in self.run_after_sor:
                func(self)

            self.pipeline.logger.log('FINISH FROM SOURCE TO SOR', newline=True, indent_level=1)

        #DV
        ddl = DdlDv(self)
        etl = EtlSorToDv(self)
        if 'valuesets' in parts:
            self.pipeline.logger.log('START FROM SOR TO VALUESETS', indent_level=1)
            #DV refs
            for mapping in self.mappings:
                if isinstance(mapping, SorToValueSetMapping):
                    etl.sor_to_valuesets(mapping)
            self.pipeline.logger.log('FINISH FROM SOR TO REFS', newline=True, indent_level=1)

        # DV Entities (Hubs en Sats)
        if 'hubs' in parts:
            self.pipeline.logger.log('START FROM SOR TO HUBS', indent_level=1)
            for mapping in self.mappings:
                if type(mapping) == SorToEntityMapping:
                    if not isinstance(mapping.source, SorQuery):
                        DdlSor(self).try_add_fk_sor_hub(mapping)
                    etl.sor_to_entity(mapping)
            for validation in self.validations:
                if isinstance(validation, DvValidation):
                    etl.validate_dv(validation)
            self.pipeline.logger.log('FINISH FROM SOR TO HUBS', newline=True, indent_level=1)

        if 'views' in parts:
            self.pipeline.logger.log('START FROM HUBS TO HUBS', indent_level=1)
            for mapping in self.mappings:
                if isinstance(mapping, EntityViewToEntityMapping):
                    ddl.create_or_alter_entity(mapping)
                    ddl.create_or_alter_view(mapping)
                    etl.view_to_entity(mapping)
            self.pipeline.logger.log('FINISH FROM HUBS TO HUBS', newline=True, indent_level=1)

        #DV Links
        if 'links' in parts:
            self.pipeline.logger.log('START FROM SOR TO LINKS', indent_level=1)
            for mapping in self.mappings:
                if type(mapping) == SorToLinkMapping:
                    DdlSor(self).try_add_fk_sor_link(mapping)
                    etl.sor_to_link(mapping)
            self.pipeline.logger.log('FINISH FROM SOR TO LINKS', newline=True, indent_level=1)

        if 'viewlinks' in parts:
            self.pipeline.logger.log('START FROM HUBS TO LINKS', indent_level=1)
            for mapping in self.mappings:
                if isinstance(mapping, EntityViewToLinkMapping):
                    ddl.create_or_alter_link(mapping)
                    etl.view_to_link(mapping)
            self.pipeline.logger.log('FINISH FROM HUBS TO LINKS', newline=True, indent_level=1)

        for mapping in self.mappings:
            if isinstance(mapping, SourceToSorMapping):
                etl.copy_to_exceptions_table(mapping.sor_table, self.sor)


        #delete /tmp/datatranfser
        if self.source_db:
            path = self.source_db.get_or_create_datatransfer_path()


    def validate(self):
        """
        Voert achtereenvolgens uit: :class:'validate_domains' en validate_mappings. Geneereert de validatie-boodschap.

        :return: validation_msg. leeg indien validatie goed is
        """
        validation_msg = self.validate_domains()
        validation_msg += self.validate_mappings_after_ddl()
        return validation_msg

    def validate_mappings_before_ddl(self):
        """
        Valideert of de gebruikte mappings correcte domein classes gebruikt (entities, hubs, sats en links).

        zie: :class:`Pipeline.validate_mappings`

        :return: validation_msg
        """

        validator = MappingsValidator()