Dataform Tools for Google Cloud Platform

Wrapper for the Google Cloud Dataform Python Package (google-cloud-dataform) that simplifies common operations such as listing repositories, managing workspaces, creating compilation results, and triggering workflow invocations.

Installation

Required python >=3.10

pip install dataform-tools

Usage

Run Dataform workflow in GCP

from dataform_tools import DataformTools, CodeCompilationConfigType, InvocationConfigType
 
tags_to_run = ["your-tag"]
code_compilation_config: CodeCompilationConfigType = {} # overrides such as table prefix
invocation_config: InvocationConfigType = {
    "included_tags": tags_to_run,
    "fully_refresh_incremental_tables_enabled": False,
    "transitive_dependents_included": False,
    "transitive_dependencies_included":False,
}
 
repository_name = "repository-name"
workspace_name = None  # use workspace name if you want to compile a workspace.
git_commitish = "develop" # branch name, tag, commit sha
 
client = DataformTools("your-gcp-project", "europe-west2")
output = client.run_dataform_remotely(repository_name, code_compilation_config, invocation_config, workspace_name, git_commitish)
print(output)

List Repositories

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
repositories = client.list_repositories()
print(repositories)

List Workspaces

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
workspaces = client.list_workspaces("repository_name")
print(workspaces)

Create Workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
workspace = client.create_workspace("repository_name", "workspace_name")
print(workspace)

Get repository

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
repository = client.get_repository("repository_name")
print(repository)

Create Compilation Result

Creates a compilation object from Dataform Pipeline using either the code from a specific git_commitish or workspace.

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
compilation_result = client.create_compilation_result("repository_name", "git_branch_name", None, code_compilation_config={"table_prefix": "aa"})
print(compilation_result)

Compilation Result Actions

Quries a list of actions that will be created by a compilation object from Dataform Pipeline using either the code from a specific git_commitish or workspace.

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
compilation_result = client.create_compilation_result("repository_name", "git_branch_name", None, code_compilation_config={"table_prefix": "aa"})
if(compilation_result and compilation_result.name):
    actions = client.query_compilation_result_actions(compilation_result.name)
    print(actions)

Create Workflow Invocation

Creates a execution of Dataform Pipeline using either the code from a specific git_commitish or workspace.

from dataform_tools import DataformTools
from dataform_tools import InvocationConfigType
client = DataformTools("your-gcp-project-id", "europe-west2")

repository_name = "repository_name"
compilation_result = client.create_compilation_request(repository_name, "git_branch_name", None, code_compilation_config={"table_prefix": "aa"})
if(compilation_result and compilation_result.name):
    invocation_config: InvocationConfigType = {
        "included_tags" : ["your-tag"],
        "transitive_dependencies_included" : False,
        "transitive_dependents_included" : False,
        "fully_refresh_incremental_tables_enabled" : False,
    }
    workflow_invocation = client.create_workflow_invocation(repository_name, compilation_result.name, invocation_config)
    workflow_invocation_id = workflow_invocation.name.split("/").pop()
    if(workflow_invocation_id):
        workflow_invocation_url = client.get_workflow_invocation_url(repository_name, workflow_invocation_id)
        print(workflow_invocation_url)

Write content to a file in workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
client.write_file("repository_name", "workspace_name", "relative/path/to/file/in/workspace.sql", "select 1 as a")

Remove a file from workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
client.remove_file("repository_name", "workspace_name", "relative/path/to/file/in/workspace.sql")

Installs NPM packages in a Dataform workspace.

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
client.install_npm_packages("repository-name", "my-workspace");

Pull git commits from remote repository to workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
client.pull_git_commits("repository-name", "workspace-name", {"remote-git-branch", "git-user-name", "git-user-email"});

Get git status of the remote workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")

client.get_workspace_git_state("repository-name", "workspace-name")

Reset changes in workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
paths = []  # array of file paths to reset. If empty, all changes will be reset.
clean = True # If True, untracked files will be removed. Defaults to True.
client.reset_workspace_changes("repository-name", "workspace-name", paths, clean)

Fetch Git ahead/behind against a remote branch for a workspace

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
client.fetch_git_ahead_behind("repository-name", "workspace-name", "remote-git-branch")

Push workspace commits to git remote repository

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")
client.push_workspace_commits("repository-name", "my-worksapce", "remote_git_branch") 

List Workflow Invocations

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")

# Fetch the latest 5 workflow invocations
invocations = client.list_workflow_invocations(
    repository_name="repository-name",
    page_size=5,
    order_by="name desc"
)

# Extract and print invocation details
for invocation in invocations:
    invocation_id = invocation.name.split("/")[-1]
    url = client.get_workflow_invocation_url("repository-name", invocation_id)
    print(f"ID: {invocation_id}")
    print(f"URL: {url}")
    break # Remove or change logic to iterate through all results

Get Latest Workflow Invocation

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")

latest = client.get_latest_workflow_invocation("repository-name")
if latest is not None:
    invocation_id = latest.name.split("/")[-1]
    url = client.get_workflow_invocation_url("repository-name", invocation_id)
    print(f"ID: {invocation_id}")
    print(f"State: {latest.state}")
    print(f"URL: {url}")

Query Workflow Invocation Actions

Lists the actions (tables, assertions, operations, etc.) executed by a specific workflow invocation, along with their state.

from dataform_tools import DataformTools
client = DataformTools("your-gcp-project-id", "europe-west2")

latest = client.get_latest_workflow_invocation("repository-name")
if latest is not None:
    invocation_id = latest.name.split("/")[-1]
    # Optional: pass page_size / page_token for a single page
    actions = client.query_workflow_invocation_actions(
        "repository-name",
        invocation_id,
        page_size=50,
    )
    for action in actions:
        print(f"{action.target.name}: {action.state}")