dataform-tools 1.2.0

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

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}")