G-gen の菊池です。当記事では GitHub などのリモートリポジトリに接続していない Dataform において、grep のようなファイル内検索をする手順について解説します。

はじめに

Dataform とは

Dataform とは、BigQuery のためのデータパイプライン管理ツールです。 Dataform の基本的な概念や使い方については、以下の記事で解説しています。

blog.g-gen.co.jp

Dataformでは、SQL ワークフローを構成する SQLX ファイル、JavaScript ファイル、各種設定ファイルなどを Dataform リポジトリで一元管理します。

このリポジトリは Git によってバージョン管理されており、チームでの共同開発を可能にします。Dataform が内蔵するリポジトリでは、単一のデフォルトブランチと、複数作成できるワークスペースと呼ばれる作業空間で、基本的なバージョン管理を行うことができます。さらに、GitHub や GitLab などの外部リポジトリと連携すれば、複数のブランチを活用した開発が可能になります。

参考：Dataform の概要
参考：コードのバージョン管理を行う

Dataform のファイル内検索

Dataform に格納している SQLX ファイル内の文字列を検索したい場合、Dataform のコンソールにはそのような機能がありません。

GitHub に接続している場合は、 git grep コマンドを使用するなどして、ファイル内検索ができます。しかし、セキュリティ上の制約によりサードパーティの Git リポジトリに接続できない場合には、その手段は使用できません。

GitHub に接続していない Dataform でも、Dataform API を使用することでファイル内のテキストを取得できます。当記事では、Python 用の Dataform 向けクライアントライブラリを使用することで、Dataform API へリクエストし、テキスト検索する方法を紹介します。

また当記事では、ローカルに Python 環境を構築する手順を省くために、Cloud Shell を使用しています。

ソースコード

main.py

今回使用したコードの全文を以下に記載します。

import re
from google.cloud import dataform_v1
  
# 定数: ご自身のプロジェクト情報に書き換えてください
PROJECT_ID = "PROJECT_ID"
REGION = "asia-northeast1"
REPOSITORY_ID = "REPOSITORY_NAME"
WORKSPACE_ID = "WORKSPACE_NAME"
SEARCH_WORD = "tags:"
  
  
def read_file(client, workspace_path, file_path):
    """指定されたDataformワークスペース内のファイルを読み取ります。"""
    request = dataform_v1.ReadFileRequest(
        workspace=workspace_path,
        path=file_path,
    )
    response = client.read_file(request=request)
    return response.file_contents.decode('utf-8')
  
  
def search_files_in_workspace(workspace_path, search_word):
    """
    Dataformワークスペース内の.sqlxファイルを検索し、
    指定された単語を含む行を特定して表示します。
    """
    client = dataform_v1.DataformClient()
    request = dataform_v1.SearchFilesRequest(workspace=workspace_path)
    page_result = client.search_files(request=request)
  
    pattern = f".*{re.escape(search_word)}.*"
  
    print(f"検索ワード: {search_word}\n---")
  
    for response in page_result:
        # 「ファイルであること」と「パスの末尾が.sqlxであること」の2つの条件をチェック
        if response.file and response.file.path.endswith(".sqlx"):
            file_path = response.file.path
            file_text = read_file(client, workspace_path, file_path)
  
            for line_num, line in enumerate(file_text.splitlines(), 1):
                if re.match(pattern, line):
                    # パターンにマッチした行を表示
                    print(f"ファイルパス: {file_path}, 行番号: {line_num}, 内容: {line.strip()}")
  
  
def main():
    """スクリプトのメイン処理を実行します。"""
    workspace = (
        f"projects/{PROJECT_ID}/locations/{REGION}/repositories/"
        f"{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
    )
    search_files_in_workspace(workspace, SEARCH_WORD)
  
  
if __name__ == "__main__":
    main()

コードの解説

main.py では、以下のような処理でSQLXファイル内の文字列を検索しています。

# 定数: ご自身のプロジェクト情報に書き換えてください
PROJECT_ID = "PROJECT_ID"
REGION = "asia-northeast1"
REPOSITORY_ID = "REPOSITORY_NAME"
WORKSPACE_ID = "WORKSPACE_NAME"
SEARCH_WORD = "tags:"

検索対象と検索ワードの指定

プロジェクトIDやDataform のリージョン、リポジトリ、ワークスペース、検索したい文字を指定します。

def main():
    """スクリプトのメイン処理を実行します。"""
    workspace = (
        f"projects/{PROJECT_ID}/locations/{REGION}/repositories/"
        f"{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
    )
    search_files_in_workspace(workspace, SEARCH_WORD)
  
  
if __name__ == "__main__":
    main()

ワークスペースパスの作成

検索対象のDataform のワークスペースパスを作成し、検索ワードと共に引数としてsearch_files_in_workspace関数に渡します。

def read_file(client, workspace_path, file_path):
    """指定されたDataformワークスペース内のファイルを読み取ります。"""
    request = dataform_v1.ReadFileRequest(
        workspace=workspace_path,
        path=file_path,
    )
    response = client.read_file(request=request)
    return response.file_contents.decode('utf-8')

ファイル内容の取得

Dataform のクライアントライブラリのread_fileメソッドは、引数としてパスを渡したファイルの中身をバイト形式で返します。それを文字コード（UTF-8）を指定して、人間が読める形式の文字列に変換します。

参考：Method: read_files | Python Client for Dataform API

def search_files_in_workspace(workspace_path, search_word):
    """
    Dataformワークスペース内の.sqlxファイルを検索し、
    指定された単語を含む行を特定して表示します。
    """
    client = dataform_v1.DataformClient()
    request = dataform_v1.SearchFilesRequest(workspace=workspace_path)
    page_result = client.search_files(request=request)
  
    pattern = f".*{re.escape(search_word)}.*"
  
    print(f"検索ワード: {search_word}\n---")
  
    for response in page_result:

Dataform のファイル一覧の取得

Dataform のクライアントライブラリのsearch_filesメソッドは、引数として渡したワークスペースパスにあるディレクトリやファイルの一覧を取得します。

参考：Method: search_files | Python Client for Dataform API

# 「ファイルであること」と「パスの末尾が.sqlxであること」の2つの条件をチェック
if response.file and response.file.path.endswith(".sqlx"):
     file_path = response.file.path
     file_text = read_file(client, workspace_path, file_path)

SQLX ファイルの抽出と読み取り

search_filesメソッドでは、ディレクトリとファイルでそれぞれ以下のような結果を返します。

これらの取得結果から、ファイルかつ末尾がsqlxのものを判定してread_file関数で処理します。

for line_num, line in enumerate(file_text.splitlines(), 1):
    if re.match(pattern, line):
        # パターンにマッチした行を表示
        print(f"ファイルパス: {file_path}, 行番号: {line_num}, 内容: {line.strip()}")

検索ワードとの一致判定

ファイルの中身を1行ずつ判定し、検索ワードを含む場合は、ファイルのパスと該当の行を表示します。

実行手順

Cloud Shell の有効化

Google Cloud コンソール画面の右上にある「Cloud Shell をアクティブにする」ボタンをクリックします。

Cloud Shell API の呼び出しにユーザーの認証情報を使用する権限を付与することを承認するか確認のメッセージ画面が表示されるので「承認」をクリックします。

Cloud Shell のエディタ画面が開きます。

もし、Cloud Shell のターミナル画面の方が開いた場合は、「エディタを開く」ボタンをクリックすることで、エディタ画面へ遷移します。

main.py の作成

[File] > [New Text File]を選択して、新規ファイルを開きます。

新規ファイルの編集タブにソースコードの main.py のコードを貼り付けます。

[File] > [Save]を選択して、ファイル名に main.py と入力して、OKをクリックして保存します。

クライアントライブラリのインストール

エディタ画面の「ターミナルを開く」をクリックして、ターミナル画面へ移動します。以下のコマンドを入力して、Dataform のクライアントライブラリをインストールします。

pip install --upgrade google-cloud-dataform

main.py の実行

以下のコマンドを実行することで、main.py の search_word で指定した単語を含むファイルパスと該当行が表示されます。

python main.py

はじめに

Dataform とは

Dataform のファイル内検索

ソースコード

main.py

コードの解説

実行手順

Cloud Shell の有効化

main.py の作成

クライアントライブラリのインストール

main.py の実行

Dataformのアサーション機能を解説

ライオンのデータマネジメント

ライオンのデータ基盤構築とSAPデータ活用体制

Dataformのテーブル定義と組み込み関数の使い方

Dataformライフサイクルの管理方法（環境分離と自動化）