src/dirbs/listgen/generator.py
def _write_delta_csv_exceptions_list(self, operator_id):
        """Write delta CSV exceptions list for a particular operator."""
        cursor_name = 'listgen_write_delta_csv_exceptions_list_{0}'.format(operator_id)
        is_valid_query, imei_norm_with_check_digit_query = self._is_valid_and_check_digit_queries
        with create_db_connection(self._config.db_config) as conn, \
                conn.cursor(name=cursor_name) as cursor, contextlib.ExitStack() as stack, CodeProfiler() as cp:
            allowed_delta_reasons = self._allowed_delta_reasons(conn, 'exceptions_lists')
            fnames = {r: os.path.join(self._output_dir, '{0}_exceptions_{1}_delta_{2:d}_{3:d}_{4}.csv'
                                      .format(self._date_str,
                                              operator_id,
                                              self._base_run_id,
                                              self._run_id,
                                              r))
                      for r in allowed_delta_reasons}
            files = {r: stack.enter_context(open(fn, 'w', encoding='utf8'))
                     for r, fn in fnames.items()}
            csv_writers = {r: csv.DictWriter(f,
                                             fieldnames=['imei', 'imsi', 'msisdn'],
                                             extrasaction='ignore') for r, f in files.items()}
            for csv_writer in csv_writers.values():
                csv_writer.writeheader()

            cursor.execute(sql.SQL("""SELECT {imei_col} AS imei, imsi, msisdn, delta_reason
                                        FROM (SELECT *
                                                FROM gen_delta_exceptions_list(%s, %s),
                                                     LATERAL ({is_valid_query}) iv,
                                                     LATERAL ({imei_norm_with_check_digit_query}) cd) changes
                                       WHERE {valid_filter}
                                   """).format(is_valid_query=is_valid_query,
                                               imei_norm_with_check_digit_query=imei_norm_with_check_digit_query,
                                               valid_filter=self._valid_filter_query,
                                               imei_col=self._output_imei_column),
                           [operator_id, self._base_run_id])

            metrics = {r: defaultdict(int) for r in allowed_delta_reasons}
            for row_data in cursor:
                row_data_dict = row_data._asdict()
                delta_reason = row_data_dict.pop('delta_reason')
                csv_writer = csv_writers[delta_reason]
                csv_writer.writerow(row_data_dict)
                metrics[delta_reason]['num_records'] += 1

        return [self._gen_metadata_for_list(
            fn, **metrics[r]) for r, fn in fnames.items()], 'exceptions_lists_delta', cp.duration
Write delta CSV exceptions list for a particular operator.
Write delta CSV exceptions list for a particular operator.
Write delta CSV exceptions list for a particular operator .
def _write_delta_csv_exceptions_list(self, operator_id):
        cursor_name = 'listgen_write_delta_csv_exceptions_list_{0}'.format(operator_id)
        is_valid_query, imei_norm_with_check_digit_query = self._is_valid_and_check_digit_queries
        with create_db_connection(self._config.db_config) as conn, \
                conn.cursor(name=cursor_name) as cursor, contextlib.ExitStack() as stack, CodeProfiler() as cp:
            allowed_delta_reasons = self._allowed_delta_reasons(conn, 'exceptions_lists')
            fnames = {r: os.path.join(self._output_dir, '{0}_exceptions_{1}_delta_{2:d}_{3:d}_{4}.csv'
                                      .format(self._date_str,
                                              operator_id,
                                              self._base_run_id,
                                              self._run_id,
                                              r))
                      for r in allowed_delta_reasons}
            files = {r: stack.enter_context(open(fn, 'w', encoding='utf8'))
                     for r, fn in fnames.items()}
            csv_writers = {r: csv.DictWriter(f,
                                             fieldnames=['imei', 'imsi', 'msisdn'],
                                             extrasaction='ignore') for r, f in files.items()}
            for csv_writer in csv_writers.values():
                csv_writer.writeheader()
            cursor.execute(sql.SQL("""SELECT {imei_col} AS imei, imsi, msisdn, delta_reason
                                        FROM (SELECT *
                                                FROM gen_delta_exceptions_list(%s, %s),
                                                     LATERAL ({is_valid_query}) iv,
                                                     LATERAL ({imei_norm_with_check_digit_query}) cd) changes
                                       WHERE {valid_filter}
                                   """).format(is_valid_query=is_valid_query,
                                               imei_norm_with_check_digit_query=imei_norm_with_check_digit_query,
                                               valid_filter=self._valid_filter_query,
                                               imei_col=self._output_imei_column),
                           [operator_id, self._base_run_id])
            metrics = {r: defaultdict(int) for r in allowed_delta_reasons}
            for row_data in cursor:
                row_data_dict = row_data._asdict()
                delta_reason = row_data_dict.pop('delta_reason')
                csv_writer = csv_writers[delta_reason]
                csv_writer.writerow(row_data_dict)
                metrics[delta_reason]['num_records'] += 1
        return [self._gen_metadata_for_list(
            fn, **metrics[r]) for r, fn in fnames.items()], 'exceptions_lists_delta', cp.duration
def _write_delta_csv_exceptions_list ( self , operator_id ) : cursor_name = 'listgen_write_delta_csv_exceptions_list_{0}' . format ( operator_id ) is_valid_query , imei_norm_with_check_digit_query = self . _is_valid_and_check_digit_queries with create_db_connection ( self . _config . db_config ) as conn , conn . cursor ( name = cursor_name ) as cursor , contextlib . ExitStack ( ) as stack , CodeProfiler ( ) as cp : allowed_delta_reasons = self . _allowed_delta_reasons ( conn , 'exceptions_lists' ) fnames = { r : os . path . join ( self . _output_dir , '{0}_exceptions_{1}_delta_{2:d}_{3:d}_{4}.csv' . format ( self . _date_str , operator_id , self . _base_run_id , self . _run_id , r ) ) for r in allowed_delta_reasons } files = { r : stack . enter_context ( open ( fn , 'w' , encoding = 'utf8' ) ) for r , fn in fnames . items ( ) } csv_writers = { r : csv . DictWriter ( f , fieldnames = [ 'imei' , 'imsi' , 'msisdn' ] , extrasaction = 'ignore' ) for r , f in files . items ( ) } for csv_writer in csv_writers . values ( ) : csv_writer . writeheader ( ) cursor . execute ( sql . SQL ( """SELECT {imei_col} AS imei, imsi, msisdn, delta_reason
                                        FROM (SELECT *
                                                FROM gen_delta_exceptions_list(%s, %s),
                                                     LATERAL ({is_valid_query}) iv,
                                                     LATERAL ({imei_norm_with_check_digit_query}) cd) changes
                                       WHERE {valid_filter}
                                   """ ) . format ( is_valid_query = is_valid_query , imei_norm_with_check_digit_query = imei_norm_with_check_digit_query , valid_filter = self . _valid_filter_query , imei_col = self . _output_imei_column ) , [ operator_id , self . _base_run_id ] ) metrics = { r : defaultdict ( int ) for r in allowed_delta_reasons } for row_data in cursor : row_data_dict = row_data . _asdict ( ) delta_reason = row_data_dict . pop ( 'delta_reason' ) csv_writer = csv_writers [ delta_reason ] csv_writer . writerow ( row_data_dict ) metrics [ delta_reason ] [ 'num_records' ] += 1 return [ self . _gen_metadata_for_list ( fn , ** metrics [ r ] ) for r , fn in fnames . items ( ) ] , 'exceptions_lists_delta' , cp . duration
Write delta CSV exceptions list for a particular operator.
Write delta CSV exceptions list for a particular operator .
"""Write delta CSV exceptions list for a particular operator."""
self
operator_id