src/dirbs/importer/base_delta_importer.py
def _validate_check_helper_single_shard(self, change_type, staging_tbl_part_name, import_tbl_part_name):
        """Common code for validation_checks on a single shard."""
        with utils.create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            not_sql = sql.SQL('NOT') if change_type != 'add' else sql.SQL('')
            cursor.execute(sql.SQL("""SELECT {pks}
                                        FROM {staging_tbl} o
                                       WHERE change_type = %s
                                         AND {not_sql} EXISTS (SELECT 1
                                                                 FROM {historic_tbl} n
                                                                WHERE {join_tables_on_pk_sql}
                                                                  AND end_date IS NULL)""")
                           .format(pks=self._pk_field_columns,
                                   staging_tbl=sql.Identifier(staging_tbl_part_name),
                                   historic_tbl=sql.Identifier(self._historic_tbl_name),
                                   not_sql=not_sql,
                                   join_tables_on_pk_sql=self._compare_pks_sql()), [change_type])
            return [x._asdict() for x in cursor.fetchall()]
Common code for validation_checks on a single shard.
Common code for validation_checks on a single shard.
Common code for validation_checks on a single shard .
def _validate_check_helper_single_shard(self, change_type, staging_tbl_part_name, import_tbl_part_name):
        with utils.create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            not_sql = sql.SQL('NOT') if change_type != 'add' else sql.SQL('')
            cursor.execute(sql.SQL("""SELECT {pks}
                                        FROM {staging_tbl} o
                                       WHERE change_type = %s
                                         AND {not_sql} EXISTS (SELECT 1
                                                                 FROM {historic_tbl} n
                                                                WHERE {join_tables_on_pk_sql}
                                                                  AND end_date IS NULL)""")
                           .format(pks=self._pk_field_columns,
                                   staging_tbl=sql.Identifier(staging_tbl_part_name),
                                   historic_tbl=sql.Identifier(self._historic_tbl_name),
                                   not_sql=not_sql,
                                   join_tables_on_pk_sql=self._compare_pks_sql()), [change_type])
            return [x._asdict() for x in cursor.fetchall()]
def _validate_check_helper_single_shard ( self , change_type , staging_tbl_part_name , import_tbl_part_name ) : with utils . create_db_connection ( self . _db_config ) as conn , conn . cursor ( ) as cursor : not_sql = sql . SQL ( 'NOT' ) if change_type != 'add' else sql . SQL ( '' ) cursor . execute ( sql . SQL ( """SELECT {pks}
                                        FROM {staging_tbl} o
                                       WHERE change_type = %s
                                         AND {not_sql} EXISTS (SELECT 1
                                                                 FROM {historic_tbl} n
                                                                WHERE {join_tables_on_pk_sql}
                                                                  AND end_date IS NULL)""" ) . format ( pks = self . _pk_field_columns , staging_tbl = sql . Identifier ( staging_tbl_part_name ) , historic_tbl = sql . Identifier ( self . _historic_tbl_name ) , not_sql = not_sql , join_tables_on_pk_sql = self . _compare_pks_sql ( ) ) , [ change_type ] ) return [ x . _asdict ( ) for x in cursor . fetchall ( ) ]
Common code for validation_checks on a single shard.
Common code for validation_checks on a single shard .
"""Common code for validation_checks on a single shard."""
self
change_type
staging_tbl_part_name
import_tbl_part_name