tests/_delta_helpers.py
def historic_table_insert_params_from_dict(db_conn, importer_name, imei_norm_to_insert_list):
    """Helper function to populate historic table.

    Inserts values into the historic table correspondent to the current importer, given the importer name and the
    values to insert.
    i.e. importer_name=pairing_list; imei_norm_to_insert_list=['12345678901234','22345678901234']
    """
    # generate historic table name
    tbl_name = 'historic_{0}'.format(importer_name)
    # generate metadata dict for the current importer
    importer_metadata = importer_to_fields_dict[importer_name]
    normalized_imei_field_name = importer_metadata['normalized_imei_field_name']
    extra_field_names = importer_metadata['extra_field_names']
    # generate historic table field names list; end_date is already null by default.
    field_names_list = [normalized_imei_field_name]
    supports_imei_sharding = importer_metadata['supports_imei_sharding']
    if supports_imei_sharding:
        field_names_list.append('virt_imei_shard')
        with db_conn.cursor() as cursor:
            cursor.execute("""SELECT imei_norm, calc_virt_imei_shard(imei_norm) AS virt_imei_shard
                                FROM (SELECT UNNEST(%s)) imei_list(imei_norm)""",
                           [imei_norm_to_insert_list])
            virt_imei_shard_values = {res.imei_norm: res.virt_imei_shard for res in cursor}
    field_names_list.extend(extra_field_names)
    field_names_list.append('start_date')

    # add a default value for extra field if there is any
    extra_field_values = []
    extra_field_to_values = {'make': '',
                             'model': '',
                             'status': '',
                             'imsi': '11111111111111',
                             'msisdn': '',
                             'reporting_date': '20160420',
                             'model_number': '',
                             'brand_name': '',
                             'device_type': '',
                             'radio_interface': '',
                             'device_id': '123'}
    for fn in extra_field_names:
        assert fn in extra_field_to_values.keys()
        extra_field_values.append(extra_field_to_values[fn])

    # generate historic table field values list
    values_list = []
    for i in imei_norm_to_insert_list:
        values_row = [i]
        if supports_imei_sharding:
            values_row.append(virt_imei_shard_values[i])
        values_row.extend(extra_field_values)
        values_row.append('20160420')
        values_list.append(tuple(values_row))

    insert_into_table(db_conn, field_names_list, values_list, tbl_name)
Helper function to populate historic table.

    Inserts values into the historic table correspondent to the current importer, given the importer name and the
    values to insert.
    i.e. importer_name=pairing_list; imei_norm_to_insert_list=['12345678901234','22345678901234']
    
Helper function to populate historic table.
Inserts values into the historic table correspondent to the current importer, given the importer name and the
values to insert.
Helper function to populate historic table . Inserts values into the historic table correspondent to the current importer given the importer name and the values to insert .
def historic_table_insert_params_from_dict(db_conn, importer_name, imei_norm_to_insert_list):
    tbl_name = 'historic_{0}'.format(importer_name)
    importer_metadata = importer_to_fields_dict[importer_name]
    normalized_imei_field_name = importer_metadata['normalized_imei_field_name']
    extra_field_names = importer_metadata['extra_field_names']
    field_names_list = [normalized_imei_field_name]
    supports_imei_sharding = importer_metadata['supports_imei_sharding']
    if supports_imei_sharding:
        field_names_list.append('virt_imei_shard')
        with db_conn.cursor() as cursor:
            cursor.execute("""SELECT imei_norm, calc_virt_imei_shard(imei_norm) AS virt_imei_shard
                                FROM (SELECT UNNEST(%s)) imei_list(imei_norm)""",
                           [imei_norm_to_insert_list])
            virt_imei_shard_values = {res.imei_norm: res.virt_imei_shard for res in cursor}
    field_names_list.extend(extra_field_names)
    field_names_list.append('start_date')
    extra_field_values = []
    extra_field_to_values = {'make': '',
                             'model': '',
                             'status': '',
                             'imsi': '11111111111111',
                             'msisdn': '',
                             'reporting_date': '20160420',
                             'model_number': '',
                             'brand_name': '',
                             'device_type': '',
                             'radio_interface': '',
                             'device_id': '123'}
    for fn in extra_field_names:
        assert fn in extra_field_to_values.keys()
        extra_field_values.append(extra_field_to_values[fn])
    values_list = []
    for i in imei_norm_to_insert_list:
        values_row = [i]
        if supports_imei_sharding:
            values_row.append(virt_imei_shard_values[i])
        values_row.extend(extra_field_values)
        values_row.append('20160420')
        values_list.append(tuple(values_row))
    insert_into_table(db_conn, field_names_list, values_list, tbl_name)
def historic_table_insert_params_from_dict ( db_conn , importer_name , imei_norm_to_insert_list ) : tbl_name = 'historic_{0}' . format ( importer_name ) importer_metadata = importer_to_fields_dict [ importer_name ] normalized_imei_field_name = importer_metadata [ 'normalized_imei_field_name' ] extra_field_names = importer_metadata [ 'extra_field_names' ] field_names_list = [ normalized_imei_field_name ] supports_imei_sharding = importer_metadata [ 'supports_imei_sharding' ] if supports_imei_sharding : field_names_list . append ( 'virt_imei_shard' ) with db_conn . cursor ( ) as cursor : cursor . execute ( """SELECT imei_norm, calc_virt_imei_shard(imei_norm) AS virt_imei_shard
                                FROM (SELECT UNNEST(%s)) imei_list(imei_norm)""" , [ imei_norm_to_insert_list ] ) virt_imei_shard_values = { res . imei_norm : res . virt_imei_shard for res in cursor } field_names_list . extend ( extra_field_names ) field_names_list . append ( 'start_date' ) extra_field_values = [ ] extra_field_to_values = { 'make' : '' , 'model' : '' , 'status' : '' , 'imsi' : '11111111111111' , 'msisdn' : '' , 'reporting_date' : '20160420' , 'model_number' : '' , 'brand_name' : '' , 'device_type' : '' , 'radio_interface' : '' , 'device_id' : '123' } for fn in extra_field_names : assert fn in extra_field_to_values . keys ( ) extra_field_values . append ( extra_field_to_values [ fn ] ) values_list = [ ] for i in imei_norm_to_insert_list : values_row = [ i ] if supports_imei_sharding : values_row . append ( virt_imei_shard_values [ i ] ) values_row . extend ( extra_field_values ) values_row . append ( '20160420' ) values_list . append ( tuple ( values_row ) ) insert_into_table ( db_conn , field_names_list , values_list , tbl_name )
Helper function to populate historic table.
Helper function to populate historic table .
"""Helper function to populate historic table.

    Inserts values into the historic table correspondent to the current importer, given the importer name and the
    values to insert.
    i.e. importer_name=pairing_list; imei_norm_to_insert_list=['12345678901234','22345678901234']
    """ # generate historic table name # generate metadata dict for the current importer # generate historic table field names list; end_date is already null by default. # add a default value for extra field if there is any # generate historic table field values list
db_conn
importer_name
imei_norm_to_insert_list