Google Cloud Workflows Client Library

Warnings

• breaking Workflows no longer supports HTTP requests to IP-based endpoints of Google Kubernetes Engine (GKE) cluster control planes. You must now use DNS-based endpoints for such communications.
  Fix: Update your workflow definitions to access GKE control planes via their DNS-based endpoints. Refer to the official Workflows release notes for details.
• gotcha Google Cloud Workflows do not natively support user-defined environment variables within the workflow definition itself. This can complicate deploying the same workflow to multiple environments with different configurations.
  Fix: Google recommends using placeholders in your YAML/JSON workflow definition and replacing them at deployment time using your CI/CD pipeline.
• gotcha Workflows have strict resource limitations, including a maximum HTTP response size of 2MB (if saved to a variable) and a 512KB memory limit for all variables within a single execution. Events larger than the maximum argument size may fail to trigger executions.
  Fix: Design workflows to handle smaller payloads or process data in chunks. For large events, ensure event sizes are within quotas. Filter data at the API level where possible.
• gotcha When defining workflows in YAML, placing a `for` loop directly after a `try` block will cause a deployment error. This is a syntax parsing issue.
  Fix: Add a named step (e.g., `steps: - loopStep:`) between the `try` block and the `for` loop to resolve the parsing conflict.
• gotcha Each step in a Google Cloud Workflow incurs a latency overhead of approximately 100-200ms. For workflows with many sequential steps, this can significantly increase overall execution time.
  Fix: Consider batching operations where possible, especially for I/O-bound tasks, to reduce the number of discrete steps.
• gotcha Workflows have an execution limit of 10,000 steps per workflow. Long-running or data-intensive workflows that iterate many times may unexpectedly terminate if this limit is reached.
  Fix: For processes that involve large datasets or extensive iteration, implement chunking mechanisms to process data in smaller, manageable segments across multiple workflow executions or external services.

Install

• pip install google-cloud-workflows Install stable version

Imports

• WorkflowsClient

  from google.cloud.workflows_v1 import WorkflowsClient

  Used for managing workflows (creating, deploying, deleting, getting definitions).
• ExecutionsClient

  from google.cloud.workflows.executions_v1 import ExecutionsClient

  Used for managing workflow executions (starting, getting status, listing).

Quickstart

This quickstart demonstrates how to create a new execution for an existing Google Cloud Workflow using the client library. It assumes you have authenticated via Application Default Credentials (e.g., `gcloud auth application-default login`) and that a workflow named `my-first-workflow` exists in `us-central1`. Replace placeholder values with your actual project ID, location, and workflow name. The example passes a JSON argument to the workflow.

import os
import json
from google.cloud.workflows_v1 import WorkflowsClient
from google.cloud.workflows.executions_v1 import ExecutionsClient
from google.cloud.workflows.executions_v1.types.executions import Execution

# Set your Google Cloud Project ID, region, and existing workflow name
PROJECT_ID = os.environ.get('GOOGLE_CLOUD_PROJECT', 'your-project-id')
LOCATION = os.environ.get('GOOGLE_CLOUD_LOCATION', 'us-central1')
WORKFLOW_NAME = os.environ.get('GOOGLE_CLOUD_WORKFLOW_NAME', 'my-first-workflow')

# Initialize clients
workflows_client = WorkflowsClient()
executions_client = ExecutionsClient()

# Construct the parent path for the workflow
parent_path = workflows_client.workflow_path(PROJECT_ID, LOCATION, WORKFLOW_NAME)

# Define optional runtime arguments for the workflow (as a JSON string)
runtime_arguments = {
    "name": "World"
}

# Create an execution instance
execution = Execution(argument=json.dumps(runtime_arguments))

print(f"Creating execution for workflow: {WORKFLOW_NAME} in {LOCATION}...")
# Send the request to create the execution
response = executions_client.create_execution(parent=parent_path, execution=execution)

print(f"Created execution: {response.name} with state: {response.state.name}")
print("To view the execution in the console, visit:")
print(f"https://console.cloud.google.com/workflows/executions/{LOCATION}/{WORKFLOW_NAME}/{response.name.split('/')[-1]}?project={PROJECT_ID}")