def tag_bq_columns()

in genai-for-marketing/installation_scripts/aux_data/bq_tag_generation.py [0:0]

• 39 lines of code
• 3 McCabe index (conditional complexity)

def tag_bq_columns(
        PROJECT_ID: str,
        TAG_TEMPLATE_PATH: str,
        TAG_TEMPLATE_ID: str,
        DATASET_ID: str, 
        table_id: str, 
        column_id: str, 
        values: List):
    # Lookup Data Catalog's Entry referring to the table.
    resource_name = (
        f"//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/{table_id}"
    )

    table_entry = datacatalog_client.lookup_entry(
        request={"linked_resource": resource_name}
    )
    # Attach a Tag to the table.
    tag = datacatalog_v1.types.Tag()

    tag.template = TAG_TEMPLATE_PATH
    tag.name = "talktodata tag"
    tag.fields["description"] = datacatalog_v1.types.TagField()
    tag.fields["description"].string_value = values[0]

    tag.fields["is_primary_key"] = datacatalog_v1.types.TagField()
    tag.fields["is_primary_key"].bool_value = values[1]

    tag.fields["is_foreign_key"] = datacatalog_v1.types.TagField()
    tag.fields["is_foreign_key"].bool_value = values[2]

    tag.fields["data_type"] = datacatalog_v1.types.TagField()
    tag.fields["data_type"].string_value = values[3]

    tag.column = column_id
    try:
        tag = datacatalog_client.create_tag(parent=table_entry.name, tag=tag)
        print('tag created/updated for {}'.format(column_id))
    except Exception as e:
        print(e)
        print(
            'Failed to create template {} for {}.{}.{}.{}'.format(
                TAG_TEMPLATE_ID, 
                DATASET_ID, 
                DATASET_ID, 
                table_id,
                column_id)
        )