src/dirbs/listgen/generator.py
def _store_notifications_list_delta(self, operator_id, executor):
        """Job to update a notifications_list partition from the computed delta table."""
        per_type_counts = {}
        with create_db_connection(self._config.db_config) as conn, conn.cursor() as cursor, CodeProfiler() as cp:
            tbl = sql.Identifier(self._notifications_lists_part_tblname(operator_id))
            delta_tbl = sql.Identifier(self._notifications_lists_delta_part_tblname(operator_id))
            # We set the end_run_id on any current row where we have a change in the delta table
            cursor.execute(sql.SQL("""UPDATE {tbl} nl
                                         SET end_run_id = %s
                                       WHERE end_run_id IS NULL
                                         AND EXISTS (SELECT 1
                                                       FROM {delta_tbl}
                                                      WHERE imei_norm = nl.imei_norm
                                                        AND virt_imei_shard = nl.virt_imei_shard
                                                        AND imsi = nl.imsi
                                                        AND msisdn = nl.msisdn)
                                   """).format(tbl=tbl, delta_tbl=delta_tbl),
                           [self._run_id])
            per_type_counts['invalidated'] = cursor.rowcount
            # Now we should be able to just insert the delta list into the notifications list
            cursor.execute(sql.SQL("""INSERT INTO {tbl}(operator_id,
                                                        imei_norm,
                                                        virt_imei_shard,
                                                        imsi,
                                                        msisdn,
                                                        block_date,
                                                        reasons,
                                                        amnesty_granted,
                                                        start_run_id,
                                                        end_run_id,
                                                        delta_reason)
                                           SELECT operator_id,
                                                  imei_norm,
                                                  virt_imei_shard,
                                                  imsi,
                                                  msisdn,
                                                  block_date,
                                                  reasons,
                                                  amnesty_granted,
                                                  start_run_id,
                                                  end_run_id,
                                                  delta_reason
                                             FROM {delta_tbl}
                                   """).format(tbl=tbl, delta_tbl=delta_tbl),
                           [self._run_id])
            per_type_counts['new'] = cursor.rowcount
            self._create_missing_notifications_partition_indices(conn, operator_id)

        return per_type_counts, cp.duration
Job to update a notifications_list partition from the computed delta table.
Job to update a notifications_list partition from the computed delta table.
Job to update a notifications_list partition from the computed delta table .
def _store_notifications_list_delta(self, operator_id, executor):
        per_type_counts = {}
        with create_db_connection(self._config.db_config) as conn, conn.cursor() as cursor, CodeProfiler() as cp:
            tbl = sql.Identifier(self._notifications_lists_part_tblname(operator_id))
            delta_tbl = sql.Identifier(self._notifications_lists_delta_part_tblname(operator_id))
            cursor.execute(sql.SQL("""UPDATE {tbl} nl
                                         SET end_run_id = %s
                                       WHERE end_run_id IS NULL
                                         AND EXISTS (SELECT 1
                                                       FROM {delta_tbl}
                                                      WHERE imei_norm = nl.imei_norm
                                                        AND virt_imei_shard = nl.virt_imei_shard
                                                        AND imsi = nl.imsi
                                                        AND msisdn = nl.msisdn)
                                   """).format(tbl=tbl, delta_tbl=delta_tbl),
                           [self._run_id])
            per_type_counts['invalidated'] = cursor.rowcount
            cursor.execute(sql.SQL("""INSERT INTO {tbl}(operator_id,
                                                        imei_norm,
                                                        virt_imei_shard,
                                                        imsi,
                                                        msisdn,
                                                        block_date,
                                                        reasons,
                                                        amnesty_granted,
                                                        start_run_id,
                                                        end_run_id,
                                                        delta_reason)
                                           SELECT operator_id,
                                                  imei_norm,
                                                  virt_imei_shard,
                                                  imsi,
                                                  msisdn,
                                                  block_date,
                                                  reasons,
                                                  amnesty_granted,
                                                  start_run_id,
                                                  end_run_id,
                                                  delta_reason
                                             FROM {delta_tbl}
                                   """).format(tbl=tbl, delta_tbl=delta_tbl),
                           [self._run_id])
            per_type_counts['new'] = cursor.rowcount
            self._create_missing_notifications_partition_indices(conn, operator_id)
        return per_type_counts, cp.duration
def _store_notifications_list_delta ( self , operator_id , executor ) : per_type_counts = { } with create_db_connection ( self . _config . db_config ) as conn , conn . cursor ( ) as cursor , CodeProfiler ( ) as cp : tbl = sql . Identifier ( self . _notifications_lists_part_tblname ( operator_id ) ) delta_tbl = sql . Identifier ( self . _notifications_lists_delta_part_tblname ( operator_id ) ) cursor . execute ( sql . SQL ( """UPDATE {tbl} nl
                                         SET end_run_id = %s
                                       WHERE end_run_id IS NULL
                                         AND EXISTS (SELECT 1
                                                       FROM {delta_tbl}
                                                      WHERE imei_norm = nl.imei_norm
                                                        AND virt_imei_shard = nl.virt_imei_shard
                                                        AND imsi = nl.imsi
                                                        AND msisdn = nl.msisdn)
                                   """ ) . format ( tbl = tbl , delta_tbl = delta_tbl ) , [ self . _run_id ] ) per_type_counts [ 'invalidated' ] = cursor . rowcount cursor . execute ( sql . SQL ( """INSERT INTO {tbl}(operator_id,
                                                        imei_norm,
                                                        virt_imei_shard,
                                                        imsi,
                                                        msisdn,
                                                        block_date,
                                                        reasons,
                                                        amnesty_granted,
                                                        start_run_id,
                                                        end_run_id,
                                                        delta_reason)
                                           SELECT operator_id,
                                                  imei_norm,
                                                  virt_imei_shard,
                                                  imsi,
                                                  msisdn,
                                                  block_date,
                                                  reasons,
                                                  amnesty_granted,
                                                  start_run_id,
                                                  end_run_id,
                                                  delta_reason
                                             FROM {delta_tbl}
                                   """ ) . format ( tbl = tbl , delta_tbl = delta_tbl ) , [ self . _run_id ] ) per_type_counts [ 'new' ] = cursor . rowcount self . _create_missing_notifications_partition_indices ( conn , operator_id ) return per_type_counts , cp . duration
Job to update a notifications_list partition from the computed delta table.
Job to update a notifications_list partition from the computed delta table .
"""Job to update a notifications_list partition from the computed delta table.""" # We set the end_run_id on any current row where we have a change in the delta table # Now we should be able to just insert the delta list into the notifications list
self
operator_id
executor