<img src="https://avatars2.githubusercontent.com/u/2810941?v=3&s=96" alt="Google Cloud Platform logo" title="Google Cloud Platform" align="right" height="96" width="96"/>

This sample shows how to execute a Cloud Workflow and wait for the workflow execution results using the Python client libraries.

1. Deploy the workflow, `myFirstWorkflow`:

1. Copy the YAML from this file: https://github.com/GoogleCloudPlatform/workflows-samples/blob/main/src/myFirstWorkflow.workflows.yaml

1. Paste the YAML into a file called `myFirstWorkflow.workflows.yaml`.

1. Run the command: `gcloud workflows deploy myFirstWorkflow --source myFirstWorkflow.workflows.yaml`

Install [`pip`][pip] and [`virtualenv`][virtualenv] if you do not already have them.

You may want to refer to the [`Python Development Environment Setup Guide`][setup] for Google Cloud Platform for instructions.

1. Create a virtualenv. Samples are compatible with Python 2.7 and 3.4+.

```sh

virtualenv env

source env/bin/activate

```

1. Install the dependencies needed to run the samples.

```sh

1. Start the application, setting your project name in an environment variable, `GOOGLE_CLOUD_PROJECT`:

```sh

1. Observe the results:

In stdout, you should see a JSON response from your workflow like the following:

```json

[prereq]: ../README.md#prerequisities

[setup]: https://cloud.google.com/python/setup

[pip]: https://pip.pypa.io/

[virtualenv]: https://virtualenv.pypa.io/

import os

def execute_workflow(

project, location="us-central1", workflow="myFirstWorkflow"

):

"""Execute a workflow and print the execution results."""

# [START workflows_api_quickstart]

import time

from google.cloud import workflows_v1beta

from google.cloud.workflows import executions_v1beta

from google.cloud.workflows.executions_v1beta.types import executions

# TODO(developer): Uncomment these lines and replace with your values.

# project = 'my-project-id'

# location = 'us- central1'

# workflow = 'myFirstWorkflow'

if not project:

raise Exception('GOOGLE_CLOUD_PROJECT env var is required.')

# Set up API clients.

execution_client = executions_v1beta.ExecutionsClient()

workflows_client = workflows_v1beta.WorkflowsClient()

# Construct the fully qualified location path.

parent = workflows_client.workflow_path(project, location, workflow)

# Execute the workflow.

response = execution_client.create_execution(request={"parent": parent})

print(f"Created execution: {response.name}")

# Wait for execution to finish, then print results.

execution_finished = False

backoff_delay = 1 # Start wait with delay of 1 second

print('Poll every second for result...')

while (not execution_finished):

execution = execution_client.get_execution(request={"name": response.name})

execution_finished = execution.state != executions.Execution.State.ACTIVE

# If we haven't seen the result yet, wait a second.

if not execution_finished:

print('- Waiting for results...')

time.sleep(backoff_delay)

backoff_delay *= 2 # Double the delay to provide exponential backoff.

else:

print(f'Execution finished with state: {execution.state.name}')

print(execution.result)

return execution.result

# [END workflows_api_quickstart]

if __name__ == "__main__":

project = os.environ.get('GOOGLE_CLOUD_PROJECT')

execute_workflow(project=project)

import os

from google.cloud import workflows_v1beta

import main

PROJECT = os.environ["GOOGLE_CLOUD_PROJECT"]

LOCATION = "us-central1"

WORKFLOW_ID = "myFirstWorkflow"

def test_workflow_execution():

assert PROJECT != ""

if not workflow_exists():

workflow_file = open("myFirstWorkflow.workflows.yaml", "r").read()

workflows_client = workflows_v1beta.WorkflowsClient()

workflows_client.create_workflow(request={

# Manually construct the location

# https://github.com/googleapis/python-workflows/issues/21

"parent": f'projects/{PROJECT}/locations/{LOCATION}',

"workflow_id": WORKFLOW_ID,

"workflow": {

"name": WORKFLOW_ID,

"source_contents": workflow_file

}

})

result = main.execute_workflow(PROJECT)

assert len(result) > 0

def workflow_exists():

"""Returns True if the workflow exists in this project

try:

workflows_client = workflows_v1beta.WorkflowsClient()

workflow_name = workflows_client.workflow_path(PROJECT, LOCATION, WORKFLOW_ID)

workflows_client.get_workflow(request={"name": workflow_name})

return True

except Exception as e:

print(f"Workflow doesn't exist: {e}")

return False

- getCurrentTime:

call: http.get

args:

url: https://us-central1-workflowsample.cloudfunctions.net/datetime

result: currentTime

- readWikipedia:

call: http.get

args:

url: https://en.wikipedia.org/w/api.php

query:

action: opensearch

search: ${currentTime.body.dayOfTheWeek}

result: wikiResult

- returnResult:

return: ${wikiResult.body[1]}

pytest==6.2.1