fix(workflows): Fix Quickstart for Cloud Client (#13063)

* fix(workflows): Fix Quickstart for Cloud Client

- Add __main__ to call execute_workflow() when running `python main.py`
- Remove redundant default arguments LOCATION and WORKFLOW_ID (as we have them in main.py and main_test.py as constants)
- Apply style recomendations to pass linter check.

* fix(workflows): fix blank line with whitespace

* fix(workflows): add default values for environment variables

To be compatible with documentation in https://cloud.google.com/workflows/docs/execute-workflow-client-libraries#run_the_sample

Author: eapl-gemugami

workflows/cloud-client/main.py

@@ -13,23 +13,28 @@
 # limitations under the License.
 
 # [START workflows_api_quickstart]
+import os
 import time
 
 from google.cloud import workflows_v1
 from google.cloud.workflows import executions_v1
 from google.cloud.workflows.executions_v1 import Execution
 from google.cloud.workflows.executions_v1.types import executions
 
+PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")
+LOCATION = os.getenv("LOCATION", "us-central1")
+WORKFLOW_ID = os.getenv("WORKFLOW", "myFirstWorkflow")
 
-def execute_workflow(
-    project: str, location: str = "us-central1", workflow: str = "myFirstWorkflow"
-) -> Execution:
+
+def execute_workflow(project: str, location: str, workflow: str) -> Execution:
     """Execute a workflow and print the execution results.
 
-    A workflow consists of a series of steps described using the Workflows syntax, and can be written in either YAML or JSON.
+    A workflow consists of a series of steps described
+    using the Workflows syntax, and can be written in either YAML or JSON.
 
     Args:
-        project: The Google Cloud project id which contains the workflow to execute.
+        project: The Google Cloud project id
+            which contains the workflow to execute.
         location: The location for the workflow
         workflow: The ID of the workflow to execute.
 
@@ -41,6 +46,7 @@ def execute_workflow(
     execution_client = executions_v1.ExecutionsClient()
     workflows_client = workflows_v1.WorkflowsClient()
     # [END workflows_api_quickstart_client_libraries]
+
     # [START workflows_api_quickstart_execution]
     # Construct the fully qualified location path.
     parent = workflows_client.workflow_path(project, location, workflow)
@@ -54,7 +60,9 @@ def execute_workflow(
     backoff_delay = 1  # Start wait with delay of 1 second
     print("Poll for result...")
     while not execution_finished:
-        execution = execution_client.get_execution(request={"name": response.name})
+        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.
@@ -70,4 +78,7 @@ def execute_workflow(
     # [END workflows_api_quickstart_execution]
 
 
+if __name__ == "__main__":
+    assert PROJECT, "'GOOGLE_CLOUD_PROJECT' environment variable not set."
+    execute_workflow(PROJECT, LOCATION, WORKFLOW_ID)
 # [END workflows_api_quickstart]

workflows/cloud-client/main_test.py

@@ -19,13 +19,13 @@
 
 import main
 
-PROJECT = os.environ["GOOGLE_CLOUD_PROJECT"]
+PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")
 LOCATION = "us-central1"
 WORKFLOW_ID = "myFirstWorkflow"
 
 
 def test_workflow_execution() -> None:
-    assert PROJECT != ""
+    assert PROJECT, "'GOOGLE_CLOUD_PROJECT' environment variable not set."
 
     if not workflow_exists():
         workflow_file = open("myFirstWorkflow.workflows.yaml").read()
@@ -37,20 +37,25 @@ def test_workflow_execution() -> None:
                 # 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},
+                "workflow": {
+                    "name": WORKFLOW_ID,
+                    "source_contents": workflow_file
+                },
             }
         )
 
-    result = main.execute_workflow(PROJECT)
+    result = main.execute_workflow(PROJECT, LOCATION, WORKFLOW_ID)
     assert result.state == executions.Execution.State.SUCCEEDED
-    assert len(result.result) > 0
+    assert result.result  # Result not empty
 
 
 def workflow_exists() -> bool:
-    """Returns True if the workflow exists in this project"""
+    """Returns True if the workflow exists in this project."""
     try:
         workflows_client = workflows_v1.WorkflowsClient()
-        workflow_name = workflows_client.workflow_path(PROJECT, LOCATION, WORKFLOW_ID)
+        workflow_name = workflows_client.workflow_path(
+            PROJECT, LOCATION, WORKFLOW_ID
+        )
         workflows_client.get_workflow(request={"name": workflow_name})
         return True
     except Exception as e: