How to map a Python Dict to a Big Query Schema

Question

I have a dict with some nested values as this:

my_dict = {
    "id": 1,
    "name": "test",
    "system": "x",
    "date": "2015-07-27",
    "profile": {
        "location": "My City",
        "preferences": [
            {
                "code": "5",
                "description": "MyPreference",
            }
        ]
    },
    "logins": [
        "2015-07-27 07:01:03",
        "2015-07-27 08:27:41"
    ]
}

and, I have a Big Query Table Schema as follows:

schema = {
    "fields": [
        {'name':'id', 'type':'INTEGER', 'mode':'REQUIRED'},
        {'name':'name', 'type':'STRING', 'mode':'REQUIRED'},
        {'name':'date', 'type':'TIMESTAMP', 'mode':'REQUIRED'},
        {'name':'profile', 'type':'RECORD', 'fields':[
            {'name':'location', 'type':'STRING', 'mode':'NULLABLE'},
            {'name':'preferences', 'type':'RECORD', 'mode':'REPEATED', 'fields':[
                {'name':'code', 'type':'STRING', 'mode':'NULLABLE'},
                {'name':'description', 'type':'STRING', 'mode':'NULLABLE'}
            ]},
        ]},
        {'name':'logins', 'type':'TIMESTAMP', 'mode':'REPEATED'}
    ]
}

I'd like to traverse all the original my_dict and build a new dict based on the structure of the schema. In other words, iterate over the schema and pick up just the right values from the original my_dict.

To build a new dict like this (note that the field "system", not present in the schema, is not copied):

new_dict = {
    "id": 1,
    "name": "test",
    "date": "2015-07-27",
    "profile": {
        "location": "My City",
        "preferences": [
            {
                "code": "5",
                "description": "MyPreference",
            }
        ]
    },
    "logins": [
        "2015-07-27 07:01:03",
        "2015-07-27 08:27:41"
    ]
}

It could be easier without the nested fields iterating a simple dict.items() and copy values, but how can I build the new dict accessing the original dict recursively?

Answer 1

Consider using schema_from_json:

my_schema = bq_client.schema_from_json('path/to/schema/file.json')

In case you need the schema code, then you can use copy the representation

my_schema
>>> [SchemaField('city', 'STRING', 'NULLABLE', None, (), None),
SchemaField('address', 'STRING', 'NULLABLE', None, (), None)]

and edit it:

from google.cloud import bigquery as bq
my_edited_schema = [bq.SchemaField('city', 'STRING', 'NULLABLE', None, (), None),
bq.SchemaField('address', 'STRING', 'NULLABLE', None, (), None)]

Answer 2

I updated the function as the use of Schemafield has changed a bit.

# [START] map_dict_to_bq_schema
# Function to take a dictionary and the bigquery schema
# and return a tuple to feed into bigquery
def map_dict_to_bq_schema(source_dict, schema, dest_dict=None):
    if dest_dict is None:
        dest_dict = dict()
    # Use the existing schema to iterate over all the fields.
    # Note: some fields may be nested (those are then flagged as a RECORD)
    if not isinstance(schema, list):
        # This is an individual field.
        schema = [schema]
    # List of fields...
    for field in schema:
        if field.name in source_dict:
            # Nested object
            if field.field_type == "RECORD" and len(field.fields) > 0:
                # This is a nested field.
                if field.mode == "REPEATED":
                    dest_dict[field.name] = []
                    for item in source_dict[field.name]:
                        new_item = {}
                        # Recursive!
                        map_dict_to_bq_schema( item, field, new_item )
                        dest_dict[field.name].append(new_item)
                else:
                    dest_dict[field.name] = {}
                    # Recursive!
                    map_dict_to_bq_schema( source_dict[field.name], field, dest_dict[field.name] )
            # Array
            elif field.mode == "REPEATED":
                if field.name in source_dict:
                    dest_dict[field.name] = []
                    for item in source_dict[field.name]:
                        dest_dict[field.name].append(item)
                else:
                    dest_dict[field.name] = [""]
            # Regular field
            else:
                dest_dict[field.name] = source_dict[field.name]
    # Done...
    return dest_dict
# [END] map_dict_to_bq_schema

Answer 3

I've build a recursive function to do this. I'm not sure if it's the better way, but worked:

def map_dict_to_bq_schema(source_dict, schema, dest_dict):
    #iterate every field from current schema
    for field in schema['fields']:
        #only work in existant values
        if field['name'] in source_dict:
            #nested field
            if field['type'].lower()=='record' and 'fields' in field:
                #list
                if 'mode' in field and field['mode'].lower()=='repeated':
                    dest_dict[field['name']] = []
                    for item in source_dict[field['name']]:
                        new_item = {}
                        map_dict_to_bq_schema( item, field, new_item )
                        dest_dict[field['name']].append(new_item)
                #record
                else:
                    dest_dict[field['name']] = {} 
                    map_dict_to_bq_schema( source_dict[field['name']], field, dest_dict[field['name']] )
            #list
            elif 'mode' in field and field['mode'].lower()=='repeated':
                dest_dict[field['name']] = []
                for item in source_dict[field['name']]:
                    dest_dict[field['name']].append(item)
            #plain field
            else:
                dest_dict[field['name']]=source_dict[field['name']]

                format_value_bq(source_dict[field['name']], field['type'])

new_dict = {}
map_dict_to_bq_schema (my_dict, schema, new_dict)