from google.cloud import bigquery

def get_policy_tags_from_bq_table(project_id, dataset_id, table_id):

    print("################ bigquery.Client.get_table().schema start ################")

    print(f"Target table: {project_id}.{dataset_id}.{table_id}")

    bq_client = bigquery.Client()

    table = bq_client.get_table(f"{project_id}.{dataset_id}.{table_id}")

    schema = table.schema

    policy_tags = []

    for field in schema:

        print(f"Column: {field.name}")

        if field.policy_tags:

            tags = [tag for tag in field.policy_tags.names]

            policy_tags.extend(tags)

            print(f"Policy Tags: {tags}")

        else:

            print("> No Policy Tags assigned.")

    return policy_tags

PROJECT_ID = "prj"

DATASET_ID = "ds"

TABLE_ID = "test001"

policy_tags = get_policy_tags_from_bq_table(PROJECT_ID, DATASET_ID, TABLE_ID)

print("Collected Policy Tags:", policy_tags)

■ポリシータグ設定

from google.cloud import datacatalog_v1

from google.cloud import bigquery

PROJECT_ID = "prj"

DATASET_ID = "ds"

TABLE_ID = "tbl01"

COLUMN_NAME = "aaa"

POLICY_TAG_PROJECT = "prj"

POLICY_TAG_NAME = "projects/prj/locations/us/taxonomies/83893110/policyTags/11089383"

def list_taxonomy_and_policy_tag():

    print("############# Start #############")

    list_policy_tags = []

    client = datacatalog_v1.PolicyTagManagerClient()

    request = datacatalog_v1.ListTaxonomiesRequest(

        parent=f"projects/{POLICY_TAG_PROJECT}/locations/us"

    )

    try:

        page_result = client.list_taxonomies(request=request)

    except google.api_core.exceptions.PermissionDenied as e:

        print(f"Skipping project {POLICY_TAG_PROJECT} due to PermissionDenied error: {e}")

        return []

    except Exception as e:

        print(f"An error occurred for project {POLICY_TAG_PROJECT}: {e}")

        return []

    for taxonomy in page_result:

        print(f"############ Taxonomy display_name: {taxonomy.display_name} #############")

        print(f"############ Taxonomy name: {taxonomy.name} #############")

        request_tag = datacatalog_v1.ListPolicyTagsRequest(parent=taxonomy.name)

        try:

            page_result_tag = client.list_policy_tags(request=request_tag)

        except Exception as e:

            print(f"Error on {request_tag}: {e}")

            break

        for policy_tag in page_result_tag:

            print("Policy tag:")

            print(policy_tag)

            list_policy_tags.append({

                "project_id": POLICY_TAG_PROJECT,

                "taxonomy_display_name": taxonomy.display_name,

                "taxonomy_name": taxonomy.name,

                "policy_tag_name": policy_tag.name,

                "policy_tag_display_name": policy_tag.display_name,

            })

    return list_policy_tags

def update_table_schema_with_policy_tag(list_policy_tags):

    for policy_tag in list_policy_tags:

        if policy_tag['policy_tag_name'] == POLICY_TAG_NAME:

            print(

                f"Target policy tag:\n"

                f"  Project ID: {policy_tag['project_id']}\n"

                f"  Taxonomy Display Name: {policy_tag['taxonomy_display_name']}\n"

                f"  Taxonomy Name: {policy_tag['taxonomy_name']}\n"

                f"  Policy Tag Name: {policy_tag['policy_tag_name']}\n"

                f"  Policy Tag Display Name: {policy_tag['policy_tag_display_name']}"

            )

            client = bigquery.Client()

            table_ref = f"{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}"

            table = client.get_table(table_ref)

            new_schema = []

            for field in table.schema:

                if field.name == COLUMN_NAME:

                    new_schema.append(

                        bigquery.SchemaField(

                            name=field.name,

                            field_type=field.field_type,  # Keep original field type

                            mode=field.mode,             # Keep original mode

                            description=field.description,

                            policy_tags=bigquery.PolicyTagList([POLICY_TAG_NAME]),

                        )

                    )

                else:

                    new_schema.append(field)

            table.schema = new_schema

            updated_table = client.update_table(table, ["schema"])

            print(

                f"Updated table {updated_table.project}.{updated_table.dataset_id}.{updated_table.table_id} schema\n"

                f"with policy_tag {POLICY_TAG_NAME} on the column {COLUMN_NAME} successfully."

            )

if __name__ == "__main__":

    list_policy_tags = list_taxonomy_and_policy_tag()

    update_table_schema_with_policy_tag(list_policy_tags)