GitHub ActionからDataplex Universal Catalogのエントリーにアスペクトタイプ・用語集を自動一括登録を実行させてみた

# こちらはツール内のコードの一部をご紹介するものです。
# 実際の動作には追加の実装（認証、エラーハンドリング、設定管理など）が必要です。

from typing import Dict, Any, Optional
import requests
import csv
import logging

logger = logging.getLogger(__name__)

# =============================================================================
# 1. エントリーレベルのアスペクト付与
# =============================================================================

def attach_aspect_to_entry(self,
                           entry_name: str,
                           aspect_type_id: str,
                           aspect_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    エントリーにアスペクトを付与

    Args:
        entry_name: エントリーの完全パス
        aspect_type_id: アスペクトタイプID
        aspect_data: アスペクトデータ（フィールド名と値のマッピング）
    """
    # アスペクトタイプの完全パスを構築
    aspect_type_path = f"projects/{self.project_id}/locations/{self.location}/aspectTypes/{aspect_type_id}"

    # プロジェクト番号を取得
    project_number = self._get_project_number()

    # アスペクトキーの形式: {project_number}.{location}.{aspect_type_id}
    aspect_key = f"{project_number}.{self.location}.{aspect_type_id}"

    # エントリーを更新（PATCHリクエスト）
    update_entry = {
        "aspects": {
            aspect_key: {
                "aspectType": aspect_type_path,
                "data": aspect_data
            }
        }
    }

    # updateMask=aspects と aspect_keys パラメータを使用
    update_url = f"{self.base_url}/{entry_name}?updateMask=aspects&aspectKeys={aspect_key}"

    response = requests.patch(update_url, headers=headers, json=update_entry)

    if response.status_code in [200, 201]:
        logger.info(f"Successfully attached aspect '{aspect_type_id}' to entry")
        return response.json()

# ポイント:
# - アスペクトキーは `{project_number}.{location}.{aspect_type_id}` 形式
# - REST APIの `PATCH /v1/{entry_name}` エンドポイントを使用
# - `updateMask=aspects` と `aspectKeys` パラメータで特定のアスペクトのみ更新

# =============================================================================
# 2. フィールドレベルのアスペクト付与
# =============================================================================

def attach_aspect_to_field(self,
                           entry_name: str,
                           field_name: str,
                           aspect_type_id: str,
                           aspect_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    エントリーのフィールドにアスペクトを付与

    Args:
        entry_name: エントリーの完全パス
        field_name: フィールド名（カラム名）
        aspect_type_id: アスペクトタイプID
        aspect_data: アスペクトデータ
    """
    # プロジェクト番号を取得
    project_number = self._get_project_number()

    # フィールドレベルアスペクトのキーは "@Schema.{field_name}" サフィックス付き
    # 注意: "Schema" は大文字のS
    aspect_key = f"{project_number}.{self.location}.{aspect_type_id}@Schema.{field_name}"

    # 新しいアスペクトのみを含む更新
    update_entry = {
        "aspects": {
            aspect_key: {
                "aspectType": aspect_type_path,
                "path": f"Schema.{field_name}",  # Schema.field_name 形式
                "data": aspect_data
            }
        }
    }

    # エントリーを更新
    update_url = f"{self.base_url}/{entry_name}?updateMask=aspects&aspectKeys={aspect_key}"

    response = requests.patch(update_url, headers=headers, json=update_entry)

    if response.status_code in [200, 201]:
        logger.info(f"Successfully attached aspect to field '{field_name}'")
        return response.json()

# ポイント:
# - フィールドレベルのアスペクトキーには `@Schema.{field_name}` サフィックスを追加
# - `path` フィールドに `Schema.{field_name}` を指定（大文字のS）
# - エントリーレベルと同じPATCH APIを使用

# =============================================================================
# 3. Glossary Term付与（エントリーレベル＆フィールドレベル対応）
# =============================================================================

def attach_glossary_term(self,
                        entry_name: str,
                        glossary_id: str,
                        term_id: str,
                        term_location: str = "global",
                        field_name: Optional[str] = None) -> Dict[str, Any]:
    """
    エントリーまたはフィールドにglossary termを付与

    Args:
        entry_name: エントリーの完全パス
        glossary_id: Glossary ID
        term_id: Term ID
        term_location: Termのロケーション（デフォルト: global）
        field_name: フィールド名（カラムレベルの場合）、Noneの場合はエントリーレベル
    """
    # プロジェクト番号を取得
    project_number = self._get_project_number()

    # entryLink IDを生成（glossary_id-term_id-field_nameの形式）
    entry_link_id = f"{glossary_id}-{term_id}"
    if field_name:
        entry_link_id = f"{entry_link_id}-{field_name}"
    entry_link_id = entry_link_id.lower()  # 小文字に変換

    # entryLinksエンドポイント
    entry_links_url = f"{self.base_url}/projects/{project_number}/locations/{entry_location}/entryGroups/{entry_group}/entryLinks?entryLinkId={entry_link_id}"

    # SOURCE: データエントリー
    source_entry = {
        "name": entry_name,
        "type": "SOURCE"
    }

    # フィールドレベルの場合、pathフィールドを追加
    if field_name:
        source_entry["path"] = f"Schema.{field_name}"

    # TARGET: glossary term
    term_entry_name = f"projects/{project_number}/locations/{term_location}/entryGroups/@dataplex/entries/projects/{project_number}/locations/{term_location}/glossaries/{glossary_id}/terms/{term_id}"
    target_entry = {
        "name": term_entry_name,
        "type": "TARGET"
    }

    # リクエストボディ
    request_body = {
        "entry_link_type": "projects/dataplex-types/locations/global/entryLinkTypes/definition",
        "entry_references": [source_entry, target_entry]
    }

    response = requests.post(entry_links_url, headers=headers, json=request_body)

    if response.status_code in [200, 201]:
        logger.info(f"Successfully attached glossary term '{term_id}' to entry")
        return response.json()

# ポイント:
# - `POST .../entryLinks` エンドポイントを使用してentryLinkを作成
# - `field_name` が指定されている場合はフィールドレベル、Noneの場合はエントリーレベル
# - entryLinkIdは小文字、数字、ハイフンのみ使用可能
# - フィールドレベルの場合、SOURCEの`path`に`Schema.{field_name}`を指定

# =============================================================================
# 4. Entry Path自動検索機能（概要）
# =============================================================================

def _attach_from_unified_file(self, unified_file: str, entries_file: str = None) -> AttachmentResult:
    """統合フォーマットファイルからメタデータを付与"""
    result = AttachmentResult()

    # エントリーパスのキャッシュ（同じentry_nameの重複検索を避ける）
    entry_path_cache = {}

    # 統合ファイルを読み込んでパース
    entries_data = self._parse_unified_format(unified_file)

    # エントリーごとに処理
    for entry_name, aspects in entries_data.items():
        # キャッシュから取得、なければDataplexで検索
        if entry_name not in entry_path_cache:
            logger.info(f"Searching entry path for: {entry_name}")
            entry_path = self.dataplex_client.search_entry_by_name(entry_name)
            if entry_path:
                entry_path_cache[entry_name] = entry_path

        # メタデータ付与処理
        # ... 詳細は省略 ...

    return result

# ポイント:
# - `entry_name`（例: `transaction_data`）から完全なEntry Pathを自動検索
# - キャッシュ機構により同じエントリーの重複検索を回避
# - 内部で`gcloud dataplex entries search`コマンドを使用
# - 検索結果例：
#   projects/PROJECT_NUMBER/locations/LOCATION/entryGroups/@bigquery/entries/...

# =============================================================================
# 5. 統合CSVフォーマットのパース（概念）
# =============================================================================

def _parse_unified_format(self, file_path: str) -> Dict[str, Dict[str, Dict[str, Any]]]:
    """
    統合フォーマットをパースする

    Returns:
        {
            entry_name: {
                aspect_name: {
                    'level': 'entry' or 'field',
                    'fields': {field_name: field_value, ...},
                    'entry_field_name': 'column_name',
                    'glossary_id': 'glossary-id',
                    'term_id': 'term-id'
                }
            }
        }
    """
    entries_data = {}

    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)

        for row in reader:
            # CSVから必要な情報を抽出
            entry_name = row.get('entry_name', '').strip()
            entry_field_name = row.get('entry_field_name', '').strip()
            aspect_name = row.get('aspect_name', '').strip()

            # エントリーレベル or フィールドレベルを判定
            level = 'field' if entry_field_name else 'entry'

            # フィールド値を動的に抽出（field_name_01, field_value_01, ...）
            # ... 実装の詳細は省略 ...

    return entries_data

# ポイント:
# - 1つのCSVファイルでアスペクトとGlossary Termを同時管理
# - `entry_field_name`が空の場合はエントリーレベル、値がある場合はフィールドレベル
# - `field_name_XX` / `field_value_XX` の動的なペアを解析
# - 実装の詳細はツール固有のロジックとして非公開

# =============================================================================
# REST APIエンドポイント一覧
# =============================================================================

# -------------------------
# アスペクト付与
# -------------------------

# エントリーレベル:
# PATCH https://dataplex.googleapis.com/v1/{entry_name}?updateMask=aspects&aspectKeys={aspect_key}
#
# Body:
# {
#   "aspects": {
#     "{project_number}.{location}.{aspect_type_id}": {
#       "aspectType": "projects/{project_id}/locations/{location}/aspectTypes/{aspect_type_id}",
#       "data": { /* アスペクトデータ */ }
#     }
#   }
# }

# フィールドレベル:
# PATCH https://dataplex.googleapis.com/v1/{entry_name}?updateMask=aspects&aspectKeys={aspect_key}
#
# Body:
# {
#   "aspects": {
#     "{project_number}.{location}.{aspect_type_id}@Schema.{field_name}": {
#       "aspectType": "projects/{project_id}/locations/{location}/aspectTypes/{aspect_type_id}",
#       "path": "Schema.{field_name}",
#       "data": { /* アスペクトデータ */ }
#     }
#   }
# }

# -------------------------
# Glossary Term付与
# -------------------------

# エントリーレベル:
# POST https://dataplex.googleapis.com/v1/projects/{project_number}/locations/{location}/entryGroups/{entry_group}/entryLinks?entryLinkId={glossary_id}-{term_id}
#
# Body:
# {
#   "entry_link_type": "projects/dataplex-types/locations/global/entryLinkTypes/definition",
#   "entry_references": [
#     {
#       "name": "{entry_name}",
#       "type": "SOURCE"
#     },
#     {
#       "name": "projects/{project_number}/locations/{term_location}/entryGroups/@dataplex/entries/projects/{project_number}/locations/{term_location}/glossaries/{glossary_id}/terms/{term_id}",
#       "type": "TARGET"
#     }
#   ]
# }

# フィールドレベル:
# POST https://dataplex.googleapis.com/v1/projects/{project_number}/locations/{location}/entryGroups/{entry_group}/entryLinks?entryLinkId={glossary_id}-{term_id}-{field_name}
#
# Body:
# {
#   "entry_link_type": "projects/dataplex-types/locations/global/entryLinkTypes/definition",
#   "entry_references": [
#     {
#       "name": "{entry_name}",
#       "type": "SOURCE",
#       "path": "Schema.{field_name}"  // フィールドレベルの場合のみ
#     },
#     {
#       "name": "projects/{project_number}/locations/{term_location}/entryGroups/@dataplex/entries/projects/{project_number}/locations/{term_location}/glossaries/{glossary_id}/terms/{term_id}",
#       "type": "TARGET"
#     }
#   ]
# }

# =============================================================================
# 実装の重要ポイント
# =============================================================================

# 1. プロジェクト番号の取得
# - アスペクトキーには`project_id`ではなく`project_number`を使用
# - Cloud Resource Manager APIで取得: `GET https://cloudresourcemanager.googleapis.com/v1/projects/{project_id}`

# 2. アスペクトキーの形式
# - エントリーレベル: `{project_number}.{location}.{aspect_type_id}`
# - フィールドレベル: `{project_number}.{location}.{aspect_type_id}@Schema.{field_name}`

# 3. pathフィールドの形式
# - フィールドレベルのアスペクトとterm付与では`Schema.{field_name}`形式
# - "Schema"は大文字のS、ドット区切り

# 4. entryLinkIdの制約
# - 小文字、数字、ハイフンのみ使用可能
# - フィールドレベル: `{glossary_id}-{term_id}-{field_name}`

# 5. エラーハンドリング
# - Term付与失敗時もアスペクト付与は継続
# - 失敗はwarningとして扱い、全体の成功ステータスに影響させない

#!/bin/bash
# Deploy Dataplex Metadata Tool to Google Cloud Run

set -e

# カラー出力
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color

# 設定変数（環境変数で上書き可能）
PROJECT_ID="${GOOGLE_CLOUD_PROJECT}"
REGION="${CLOUD_RUN_REGION}"
SERVICE_NAME="${CLOUD_RUN_SERVICE_NAME}"
REPOSITORY_NAME="${ARTIFACT_REGISTRY_REPO}"
IMAGE_NAME="${SERVICE_NAME}"
IMAGE_TAG="${IMAGE_TAG:-latest}"

# Artifact Registryのフルパス
ARTIFACT_REGISTRY_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"

echo -e "${GREEN}=== Dataplex Metadata Tool - Cloud Run Deployment ===${NC}"
echo "Project ID: ${PROJECT_ID}"
echo "Region: ${REGION}"
echo "Service Name: ${SERVICE_NAME}"
echo "Image: ${ARTIFACT_REGISTRY_PATH}"
echo ""

# 1. 現在のプロジェクトIDを確認
echo -e "${YELLOW}[1/7] Checking current GCP project...${NC}"
CURRENT_PROJECT=$(gcloud config get-value project 2>/dev/null || echo "")
if [ "${CURRENT_PROJECT}" != "${PROJECT_ID}" ]; then
    echo "Setting project to ${PROJECT_ID}..."
    gcloud config set project ${PROJECT_ID}
fi

# 2. 必要なAPIを有効化
echo -e "${YELLOW}[2/7] Enabling required APIs...${NC}"
gcloud services enable \
    cloudbuild.googleapis.com \
    run.googleapis.com \
    artifactregistry.googleapis.com \
    --project=${PROJECT_ID}

# 3. Artifact Registryリポジトリの作成（存在しない場合）
echo -e "${YELLOW}[3/7] Creating Artifact Registry repository...${NC}"
if ! gcloud artifacts repositories describe ${REPOSITORY_NAME} \
    --location=${REGION} \
    --project=${PROJECT_ID} &>/dev/null; then
    echo "Creating repository ${REPOSITORY_NAME}..."
    gcloud artifacts repositories create ${REPOSITORY_NAME} \
        --repository-format=docker \
        --location=${REGION} \
        --description="Dataplex Metadata Tool container images" \
        --project=${PROJECT_ID}
else
    echo "Repository ${REPOSITORY_NAME} already exists."
fi

# 4. Docker認証の設定
echo -e "${YELLOW}[4/7] Configuring Docker authentication...${NC}"
gcloud auth configure-docker ${REGION}-docker.pkg.dev --quiet

# 5. Dockerイメージのビルド
echo -e "${YELLOW}[5/7] Building Docker image...${NC}"
docker build \
    --platform linux/amd64 \
    -f Dockerfile.cloudrun \
    -t ${ARTIFACT_REGISTRY_PATH} \
    --build-arg BUILD_DATE=$(date -u +'%Y-%m-%dT%H:%M:%SZ') \
    --build-arg VCS_REF=$(git rev-parse --short HEAD 2>/dev/null || echo "unknown") \
    .

# 6. Artifact RegistryへPush
echo -e "${YELLOW}[6/7] Pushing image to Artifact Registry...${NC}"
docker push ${ARTIFACT_REGISTRY_PATH}

# 7. Cloud Runへデプロイ
echo -e "${YELLOW}[7/7] Deploying to Cloud Run...${NC}"
gcloud run deploy ${SERVICE_NAME} \
    --image=${ARTIFACT_REGISTRY_PATH} \
    --platform=managed \
    --region=${REGION} \
    --project=${PROJECT_ID} \
    --allow-unauthenticated \
    --memory=2Gi \
    --cpu=2 \
    --timeout=600 \
    --max-instances=10 \
    --set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}"

# デプロイ完了
echo -e "${GREEN}=== Deployment Completed Successfully! ===${NC}"

# サービスURLの取得
SERVICE_URL=$(gcloud run services describe ${SERVICE_NAME} \
    --platform=managed \
    --region=${REGION} \
    --project=${PROJECT_ID} \
    --format='value(status.url)')

echo ""
echo -e "${GREEN}Service URL: ${SERVICE_URL}${NC}"
echo ""
echo "Test the service:"
echo "  curl ${SERVICE_URL}/health"
echo ""
echo "Example API calls:"
echo "  # Aspect Type Generation"
echo "  curl -X POST ${SERVICE_URL}/aspect/generate \\"
echo "    -H 'Content-Type: application/json' \\"
echo "    -d '{\"project_id\": \"${PROJECT_ID}\", \"input_dir\": \"./input_aspect\"}'"
echo ""
echo "  # Glossary Generation"
echo "  curl -X POST ${SERVICE_URL}/glossary/generate \\"
echo "    -H 'Content-Type: application/json' \\"
echo "    -d '{\"project_id\": \"${PROJECT_ID}\", \"input_dir\": \"./input_glossary\"}'"