def nest_fields()

in scripts/schema/loader.py [0:0]

• 26 lines of code
• 2 McCabe index (conditional complexity)

def nest_fields(field_array: List[Field]) -> Dict[str, Dict[str, FieldEntry]]:
    schema_root: Dict[str, Dict[str, FieldEntry]] = {'fields': {}}
    for field in field_array:
        nested_levels: List[str] = field['name'].split('.')
        parent_fields: List[str] = nested_levels[:-1]
        leaf_field: str = nested_levels[-1]
        # "nested_schema" is a cursor we move within the schema_root structure we're building.
        # Here we reset the cursor for this new field.
        nested_schema = schema_root['fields']

        current_path = []
        for idx, level in enumerate(parent_fields):
            nested_schema.setdefault(level, {})
            # Where nested fields will live
            nested_schema[level].setdefault('fields', {})

            # Make type:object explicit for intermediate parent fields
            nested_schema[level].setdefault('field_details', {})
            field_details = nested_schema[level]['field_details']
            field_details['node_name'] = level
            # Respect explicitly defined object fields
            if 'type' in field_details and field_details['type'] in ['object', 'nested']:
                field_details.setdefault('intermediate', False)
            else:
                field_details.setdefault('type', 'object')
                field_details.setdefault('name', '.'.join(parent_fields[:idx + 1]))
                field_details.setdefault('intermediate', True)

            # moving the nested_schema cursor deeper
            current_path.extend([level])
            nested_schema = nested_schema[level]['fields']
        nested_schema.setdefault(leaf_field, {})
        # Overwrite 'name' with the leaf field's name. The flat_name is already computed.
        field['node_name'] = leaf_field
        nested_schema[leaf_field]['field_details'] = field
    return schema_root