Data Catalog Tag Templateの値の初期入力と更新をPythonで行う

概要
タグテンプレートの作成
テーブルへタグテンプレートをアタッチ
テーブルにアタッチしたタグの値の変更

概要

下記ドキュメントを参考にして、タグの値に常に最新状態を載せるためのPythonプログラムを書きました。
フィールド値はドキュメントにある例を使ったため、複数のタグを作ったり、複数のテーブルに適用する場合は定義ファイルを先に作っておくと良いかと思います。
cloud.google.com

タグテンプレートの作成

まず.envを用意します。

PROJECT=sample-project
DATASET=sample_dataset
TABLE=sample_table

import os
from google.cloud import datacatalog_v1
from dotenv import load_dotenv
from google.api_core.exceptions import GoogleAPIError

load_dotenv()

project_id = os.getenv("PROJECT")
location = "us-central1"
tag_template_id = "example_tag_template_00"

datacatalog_client = datacatalog_v1.DataCatalogClient()

# タグテンプレートの作成
tag_template = datacatalog_v1.types.TagTemplate()
tag_template.display_name = "Demo Tag Template 00"

# source フィールドにSTRING型の"Source of data asset"というdisplay nameを設定
tag_template.fields["source"] = datacatalog_v1.types.TagTemplateField()
tag_template.fields["source"].display_name = "Source of data asset"
tag_template.fields["source"].type_.primitive_type = datacatalog_v1.types.FieldType.PrimitiveType.STRING

# num_rows フィールドにDOUBLE型の"Number of rows in data asset"というdisplay nameを設定
tag_template.fields["num_rows"] = datacatalog_v1.types.TagTemplateField()
tag_template.fields["num_rows"].display_name = "Number of rows in data asset"
tag_template.fields["num_rows"].type_.primitive_type = datacatalog_v1.types.FieldType.PrimitiveType.DOUBLE

# has_pii フィールドにBOOL型の"Has PII"というdisplay nameを設定
tag_template.fields["has_pii"] = datacatalog_v1.types.TagTemplateField()
tag_template.fields["has_pii"].display_name = "Has PII"
tag_template.fields["has_pii"].type_.primitive_type = datacatalog_v1.types.FieldType.PrimitiveType.BOOL

# pii_type フィールドにENUM型で("EMAIL", "SOCIAL SECURITY NUMBER", "NONE")という選択肢を設定
tag_template.fields["pii_type"] = datacatalog_v1.types.TagTemplateField()
tag_template.fields["pii_type"].display_name = "PII type"
tag_template.fields["pii_type"].type_.enum_type.allowed_values = [
    datacatalog_v1.types.FieldType.EnumType.EnumValue(display_name=dn)
    for dn in ["EMAIL", "SOCIAL SECURITY NUMBER", "NONE"]
]

# 指定したプロジェクトのlocationにタグテンプレートを作成
expected_template_name = datacatalog_v1.DataCatalogClient.tag_template_path(
    project_id, location, tag_template_id
)

try:
    tag_template = datacatalog_client.create_tag_template(
        parent=f"projects/{project_id}/locations/{location}",
        tag_template_id=tag_template_id,
        tag_template=tag_template,
    )
    print(f"Created template: {tag_template.name}")
except GoogleAPIError as e:
    print(f"Cannot create template: {expected_template_name}")
    print(f"Error: {e}")

テーブルへタグテンプレートをアタッチ

import os
from google.cloud import datacatalog_v1
from dotenv import load_dotenv

load_dotenv()

project_id = os.getenv("PROJECT")
dataset_id = os.getenv("DATASET")
table_id = os.getenv("TABLE")
location = "us-central1"
tag_template_id = "example_tag_template_00"

datacatalog_client = datacatalog_v1.DataCatalogClient()

# タグテンプレートの名前
tag_template_name = datacatalog_v1.DataCatalogClient.tag_template_path(
    project_id, location, tag_template_id
)

# Data Catalogでテーブルのエントリを取得
resource_name = (
    f"//bigquery.googleapis.com/projects/{project_id}"
    f"/datasets/{dataset_id}/tables/{table_id}"
)
table_entry = datacatalog_client.lookup_entry(
    request={"linked_resource": resource_name}
)

# テーブルにタグを付与
tag = datacatalog_v1.types.Tag()
tag.template = tag_template_name

# タグフィールドに値を設定
tag.fields["source"] = datacatalog_v1.types.TagField()
tag.fields["source"].string_value = "Copied from tlc_yellow_trips_2018"

tag.fields["num_rows"] = datacatalog_v1.types.TagField()
tag.fields["num_rows"].double_value = 113496874

tag.fields["has_pii"] = datacatalog_v1.types.TagField()
tag.fields["has_pii"].bool_value = False

tag.fields["pii_type"] = datacatalog_v1.types.TagField()
tag.fields["pii_type"].enum_value.display_name = "NONE"

# テーブルにタグを作成
tag = datacatalog_client.create_tag(parent=table_entry.name, tag=tag)
print(f"Created tag: {tag.name}")

テーブルにアタッチしたタグの値の変更

import os
from dotenv import load_dotenv
from google.cloud import datacatalog_v1
load_dotenv()

project_id = os.getenv("PROJECT")
dataset_id = os.getenv("DATASET")
table_id = os.getenv("TABLE")

# Data Catalog クライアントの初期化
datacatalog_client = datacatalog_v1.DataCatalogClient()

# タグが付与されているリソースの名前を設定（BigQuery テーブルなど）
resource_name = (
    f"//bigquery.googleapis.com/projects/{project_id}"
    f"/datasets/{dataset_id}/tables/{table_id}"
)

# リソースに関連付けられているタグのリストを取得
tags = datacatalog_client.list_tags(parent=table_entry.name)

# 例えば最初のタグを取得
for tag in tags:
    # tag: <class 'google.cloud.datacatalog_v1.types.tags.Tag'>
    # 特定のタグ名に基づいてフィルタリングすることも可能
    if tag.template_display_name == "Demo Tag Template 00":  # タグテンプレートID
        # タグフィールドの値を変更する
        tag.fields["source"].string_value = "Updated Source Value"
        tag.fields["num_rows"].double_value = 2000000
        tag.fields["has_pii"].bool_value = True
        tag.fields["pii_type"].enum_value.display_name = "EMAIL"
        
        # 変更を反映する
        updated_tag = datacatalog_client.update_tag(tag=tag)
        print(f"Updated tag: {updated_tag.name}")