src/dirbs/listgen/generator.py
def _populate_delta_exceptions_list(self, operator_id, executor):
        """Function to populate the old exceptions list for a given operator id."""
        with create_db_connection(self._config.db_config) as conn, conn.cursor() as cursor, CodeProfiler() as cp:
            delta_tbl_name = self._exceptions_lists_delta_part_tblname(operator_id)
            delta_tbl = sql.Identifier(delta_tbl_name)
            old_tbl = sql.Identifier(self._exceptions_lists_old_part_tblname(operator_id))
            new_tbl = sql.Identifier(self._exceptions_lists_new_part_tblname(operator_id))
            # First, generate a delta for any 'added' records (new adds)
            cursor.execute(sql.SQL("""INSERT INTO {delta_tbl}(operator_id,
                                                              imei_norm,
                                                              virt_imei_shard,
                                                              imsi,
                                                              start_run_id,
                                                              end_run_id,
                                                              delta_reason,
                                                              msisdn)
                                           SELECT %s, imei_norm, virt_imei_shard, imsi, %s, NULL, 'added', msisdn
                                             FROM {new_tbl} nt
                                            WHERE NOT EXISTS(SELECT 1
                                                               FROM {old_tbl}
                                                              WHERE imei_norm = nt.imei_norm
                                                                AND virt_imei_shard = nt.virt_imei_shard
                                                                AND imsi = nt.imsi
                                                                AND msisdn = nt.msisdn)
                                   """).format(delta_tbl=delta_tbl, old_tbl=old_tbl, new_tbl=new_tbl),  # noqa: Q449
                           [operator_id, self._run_id])
            num_records = cursor.rowcount
            # Next, generate a delta for any 'removed' records (new removals) (table direction flipped)
            cursor.execute(sql.SQL("""INSERT INTO {delta_tbl}(operator_id,
                                                              imei_norm,
                                                              virt_imei_shard,
                                                              imsi,
                                                              start_run_id,
                                                              end_run_id,
                                                              delta_reason,
                                                              msisdn)
                                           SELECT %s, imei_norm, virt_imei_shard, imsi, %s, NULL, 'removed', msisdn
                                             FROM {old_tbl} ot
                                            WHERE NOT EXISTS(SELECT 1
                                                               FROM {new_tbl}
                                                              WHERE imei_norm = ot.imei_norm
                                                                AND virt_imei_shard = ot.virt_imei_shard
                                                                AND imsi = ot.imsi
                                                                AND msisdn = ot.msisdn)
                                   """).format(delta_tbl=delta_tbl, old_tbl=old_tbl, new_tbl=new_tbl),  # noqa: Q449
                           [operator_id, self._run_id])
            num_records += cursor.rowcount

        return num_records, cp.duration
Function to populate the old exceptions list for a given operator id.
Function to populate the old exceptions list for a given operator id.
Function to populate the old exceptions list for a given operator id .
def _populate_delta_exceptions_list(self, operator_id, executor):
        with create_db_connection(self._config.db_config) as conn, conn.cursor() as cursor, CodeProfiler() as cp:
            delta_tbl_name = self._exceptions_lists_delta_part_tblname(operator_id)
            delta_tbl = sql.Identifier(delta_tbl_name)
            old_tbl = sql.Identifier(self._exceptions_lists_old_part_tblname(operator_id))
            new_tbl = sql.Identifier(self._exceptions_lists_new_part_tblname(operator_id))
            cursor.execute(sql.SQL("""INSERT INTO {delta_tbl}(operator_id,
                                                              imei_norm,
                                                              virt_imei_shard,
                                                              imsi,
                                                              start_run_id,
                                                              end_run_id,
                                                              delta_reason,
                                                              msisdn)
                                           SELECT %s, imei_norm, virt_imei_shard, imsi, %s, NULL, 'added', msisdn
                                             FROM {new_tbl} nt
                                            WHERE NOT EXISTS(SELECT 1
                                                               FROM {old_tbl}
                                                              WHERE imei_norm = nt.imei_norm
                                                                AND virt_imei_shard = nt.virt_imei_shard
                                                                AND imsi = nt.imsi
                                                                AND msisdn = nt.msisdn)
                                   """).format(delta_tbl=delta_tbl, old_tbl=old_tbl, new_tbl=new_tbl),  
                           [operator_id, self._run_id])
            num_records = cursor.rowcount
            cursor.execute(sql.SQL("""INSERT INTO {delta_tbl}(operator_id,
                                                              imei_norm,
                                                              virt_imei_shard,
                                                              imsi,
                                                              start_run_id,
                                                              end_run_id,
                                                              delta_reason,
                                                              msisdn)
                                           SELECT %s, imei_norm, virt_imei_shard, imsi, %s, NULL, 'removed', msisdn
                                             FROM {old_tbl} ot
                                            WHERE NOT EXISTS(SELECT 1
                                                               FROM {new_tbl}
                                                              WHERE imei_norm = ot.imei_norm
                                                                AND virt_imei_shard = ot.virt_imei_shard
                                                                AND imsi = ot.imsi
                                                                AND msisdn = ot.msisdn)
                                   """).format(delta_tbl=delta_tbl, old_tbl=old_tbl, new_tbl=new_tbl),  
                           [operator_id, self._run_id])
            num_records += cursor.rowcount
        return num_records, cp.duration
def _populate_delta_exceptions_list ( self , operator_id , executor ) : with create_db_connection ( self . _config . db_config ) as conn , conn . cursor ( ) as cursor , CodeProfiler ( ) as cp : delta_tbl_name = self . _exceptions_lists_delta_part_tblname ( operator_id ) delta_tbl = sql . Identifier ( delta_tbl_name ) old_tbl = sql . Identifier ( self . _exceptions_lists_old_part_tblname ( operator_id ) ) new_tbl = sql . Identifier ( self . _exceptions_lists_new_part_tblname ( operator_id ) ) cursor . execute ( sql . SQL ( """INSERT INTO {delta_tbl}(operator_id,
                                                              imei_norm,
                                                              virt_imei_shard,
                                                              imsi,
                                                              start_run_id,
                                                              end_run_id,
                                                              delta_reason,
                                                              msisdn)
                                           SELECT %s, imei_norm, virt_imei_shard, imsi, %s, NULL, 'added', msisdn
                                             FROM {new_tbl} nt
                                            WHERE NOT EXISTS(SELECT 1
                                                               FROM {old_tbl}
                                                              WHERE imei_norm = nt.imei_norm
                                                                AND virt_imei_shard = nt.virt_imei_shard
                                                                AND imsi = nt.imsi
                                                                AND msisdn = nt.msisdn)
                                   """ ) . format ( delta_tbl = delta_tbl , old_tbl = old_tbl , new_tbl = new_tbl ) , [ operator_id , self . _run_id ] ) num_records = cursor . rowcount cursor . execute ( sql . SQL ( """INSERT INTO {delta_tbl}(operator_id,
                                                              imei_norm,
                                                              virt_imei_shard,
                                                              imsi,
                                                              start_run_id,
                                                              end_run_id,
                                                              delta_reason,
                                                              msisdn)
                                           SELECT %s, imei_norm, virt_imei_shard, imsi, %s, NULL, 'removed', msisdn
                                             FROM {old_tbl} ot
                                            WHERE NOT EXISTS(SELECT 1
                                                               FROM {new_tbl}
                                                              WHERE imei_norm = ot.imei_norm
                                                                AND virt_imei_shard = ot.virt_imei_shard
                                                                AND imsi = ot.imsi
                                                                AND msisdn = ot.msisdn)
                                   """ ) . format ( delta_tbl = delta_tbl , old_tbl = old_tbl , new_tbl = new_tbl ) , [ operator_id , self . _run_id ] ) num_records += cursor . rowcount return num_records , cp . duration
Function to populate the old exceptions list for a given operator id.
Function to populate the old exceptions list for a given operator id .
"""Function to populate the old exceptions list for a given operator id.""" # First, generate a delta for any 'added' records (new adds) # noqa: Q449 # Next, generate a delta for any 'removed' records (new removals) (table direction flipped) # noqa: Q449
self
operator_id
executor