google_cloud_workflows_create_and_execute_with_polling.py

1import sys
2
3from google.cloud import workflows_v1
4from google.cloud.workflows import executions_v1
5
6
7def sample_create_workflow(project_id, location, workflow_id):
8    # Set up the Workflows client
9    client = workflows_v1.WorkflowsClient()
10
11    # Define the location path
12    parent = f"projects/{project_id}/locations/{location}"
13
14    # Define the workflow
15    workflow = workflows_v1.Workflow(
16        description="A simple workflow",
17        source_contents="""
18        - getCurrentTime:
19            call: http.get
20            args:
21                url: https://us-central1-workflowsample.cloudfunctions.net/datetime
22            result: currentTime
23        - returnResult:
24            return: ${currentTime.body}
25        """,
26    )
27
28    # Construct the request
29    request = workflows_v1.CreateWorkflowRequest(
30        parent=parent,
31        workflow_id=workflow_id,
32        workflow=workflow,
33    )
34
35    # Create the workflow
36    operation = client.create_workflow(request=request)
37    print("Waiting for operation to complete...")
38    response = operation.result()
39
40    print(f"Created workflow: {response.name}")
41    return response
42
43
44def sample_execute_workflow(project_id, location, workflow_id):
45    # Set up the Executions client
46    execute_client = executions_v1.ExecutionsClient()
47
48    # Define the workflow path
49    parent = f"projects/{project_id}/locations/{location}/workflows/{workflow_id}"
50
51    # Execute the workflow
52    response = execute_client.create_execution(parent=parent, execution={})
53    print(f"Created execution: {response.name}")
54
55    # Wait for execution to finish and print results
56    import time
57    while response.state == executions_v1.Execution.State.ACTIVE:
58        print("Waiting for execution to complete...")
59        time.sleep(2)
60        response = execute_client.get_execution(name=response.name)
61
62    print(f"Execution finished with state: {response.state.name}")
63    print(f"Result: {response.result}")
64
65
66if __name__ == "__main__":
67    if len(sys.argv) < 4:
68        print("Usage: python main.py <PROJECT_ID> <LOCATION> <WORKFLOW_ID>")
69        sys.exit(1)
70
71    project = sys.argv[1]
72    loc = sys.argv[2]
73    wf_id = sys.argv[3]
74
75    sample_create_workflow(project, loc, wf_id)
76    sample_execute_workflow(project, loc, wf_id)