def _load_table_annotations()

in common/annotations_loader.py [0:0]

• 38 lines of code
• 10 McCabe index (conditional complexity)

def _load_table_annotations(table_annotations: typing.Dict[str, typing.Any],
                            client: bigquery.Client):
    full_table_id = table_annotations["id"]
    table_description = table_annotations["description"]
    description_changed = False
    schema_changed = False
    try:
        table = client.get_table(full_table_id)
    except NotFound:
        logging.info("Table or view `%s` was not found. Skipping it.",
                     full_table_id)
        return
    table_description = table.description or table_description
    description_changed = table_description != table.description
    table.description = table_description
    annotation_fields = {
        field_item["name"]: field_item["description"]
        for field_item in table_annotations["fields"]
    }

    schema = table.schema.copy()
    for index, field in enumerate(schema):
        description = field.description or annotation_fields.get(field.name, "")
        if description != field.description:
            schema_changed = True
            field_dict = field.to_api_repr()
            field_dict["description"] = description
            schema[index] = bigquery.SchemaField.from_api_repr(field_dict)
    changes = []
    if schema_changed:
        table.schema = schema
        changes.append("schema")
    if description_changed:
        changes.append("description")
    if len(changes) > 0:
        client.update_table(table, changes)
        logging.info("Table/view `%s` has been updated.", full_table_id)
    else:
        logging.info("No changes in `%s`.", full_table_id)