src/dirbs/importer/base_delta_importer.py
def _remove_from_historic_single_shard(self, src_shard_name, dest_shard_name, job_start_time,
                                           change_type='remove'):
        """Remove entries fron historic table (if import supports IMEI sharding, is called once per shard)."""
        with utils.create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            cursor.execute(sql.SQL("""UPDATE {dest_shard} AS o
                                         SET end_date = %s
                                       WHERE end_date IS NULL
                                         AND EXISTS (SELECT 1
                                                       FROM {src_shard} n
                                                      WHERE change_type = %s
                                                        AND {join_tables_on_pk_sql})""")
                           .format(dest_shard=sql.Identifier(dest_shard_name),
                                   src_shard=sql.Identifier(src_shard_name),
                                   join_tables_on_pk_sql=self._compare_pks_sql()),
                           [job_start_time, change_type])

            return change_type, cursor.rowcount
Remove entries fron historic table (if import supports IMEI sharding, is called once per shard).
Remove entries fron historic table (if import supports IMEI sharding, is called once per shard).
Remove entries fron historic table ( if import supports IMEI sharding is called once per shard ) .
def _remove_from_historic_single_shard(self, src_shard_name, dest_shard_name, job_start_time,
                                           change_type='remove'):
        with utils.create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            cursor.execute(sql.SQL("""UPDATE {dest_shard} AS o
                                         SET end_date = %s
                                       WHERE end_date IS NULL
                                         AND EXISTS (SELECT 1
                                                       FROM {src_shard} n
                                                      WHERE change_type = %s
                                                        AND {join_tables_on_pk_sql})""")
                           .format(dest_shard=sql.Identifier(dest_shard_name),
                                   src_shard=sql.Identifier(src_shard_name),
                                   join_tables_on_pk_sql=self._compare_pks_sql()),
                           [job_start_time, change_type])
            return change_type, cursor.rowcount
def _remove_from_historic_single_shard ( self , src_shard_name , dest_shard_name , job_start_time , change_type = 'remove' ) : with utils . create_db_connection ( self . _db_config ) as conn , conn . cursor ( ) as cursor : cursor . execute ( sql . SQL ( """UPDATE {dest_shard} AS o
                                         SET end_date = %s
                                       WHERE end_date IS NULL
                                         AND EXISTS (SELECT 1
                                                       FROM {src_shard} n
                                                      WHERE change_type = %s
                                                        AND {join_tables_on_pk_sql})""" ) . format ( dest_shard = sql . Identifier ( dest_shard_name ) , src_shard = sql . Identifier ( src_shard_name ) , join_tables_on_pk_sql = self . _compare_pks_sql ( ) ) , [ job_start_time , change_type ] ) return change_type , cursor . rowcount
Remove entries fron historic table (if import supports IMEI sharding, is called once per shard).
Remove entries fron historic table ( if import supports IMEI sharding is called once per shard ) .
"""Remove entries fron historic table (if import supports IMEI sharding, is called once per shard)."""
self
src_shard_name
dest_shard_name
job_start_time
change_type