fix(workflows): refactor sample 'workflows_api_quickstart' to comply with the Samples Style Guide and fix tests (#13261)

* fix(workflows): refactor sample to comply with the Samples Style Guide

- Take out unneccesary `os.getenv`s and the return from the sample
- As the `__main__` entry point is requested to stay in the script, but not in the sample, it was adjusted accordingly

* fix(workflows): adjust function definition for line length

* fix(workflows): fix grammar in comments

* fix(workflows): allow testing in Python 3.9+, and update to latest google-cloud-workflows library

* test(workflows): move test from main to conftest, and add backoff to workflow creation

* fix(workflows): fix header in conftest.py

* fix(workflows): add api_core.retry as suggested in https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/AUTHORING_GUIDE.md#retry-rpcs

* fix(workflows): add exponential backoff to main_test

* fix(workflows): add missing period to comment

* fix(workflows): apply Camie Kim's feedback

update the copyright year
line 31: change "The Google Cloud project id" to "The ID of the Google Cloud project"
line 33: add a period
line 50: add one more space before the line comment per https://google.github.io/styleguide/pyguide#385-block-and-inline-comments
line 51: add a line comment (# For example: myFirstWorkflow)

* fix(workflows): add two missing spaces before in-line comment

Author: eapl-gemugami

workflows/cloud-client/conftest.py

@@ -0,0 +1,102 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import time
+import uuid
+
+from google.cloud import workflows_v1
+
+import pytest
+
+
+PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
+LOCATION = "us-central1"
+WORKFLOW_ID = "myFirstWorkflow_" + str(uuid.uuid4())
+
+
+def workflow_exists(client: workflows_v1.WorkflowsClient) -> bool:
+    """Returns True if the workflow exists in this project."""
+    try:
+        workflow_name = client.workflow_path(
+            PROJECT_ID, LOCATION, WORKFLOW_ID
+        )
+        client.get_workflow(request={"name": workflow_name})
+        return True
+    except Exception as e:
+        print(f"Workflow doesn't exist: {e}")
+        return False
+
+
+@pytest.fixture(scope="module")
+def client() -> str:
+    assert PROJECT_ID, "'GOOGLE_CLOUD_PROJECT' environment variable not set."
+    workflows_client = workflows_v1.WorkflowsClient()
+    return workflows_client
+
+
+@pytest.fixture(scope="module")
+def project_id() -> str:
+    return PROJECT_ID
+
+
+@pytest.fixture(scope="module")
+def location() -> str:
+    return LOCATION
+
+
+@pytest.fixture(scope="module")
+def workflow_id(client: workflows_v1.WorkflowsClient) -> str:
+    creating_workflow = False
+    backoff_delay = 1  # Start wait with delay of 1 second.
+
+    # Create the workflow if it doesn't exist.
+    while not workflow_exists(client):
+        if not creating_workflow:
+            # Create the workflow.
+            workflow_file = open("myFirstWorkflow.workflows.yaml").read()
+
+            parent = client.common_location_path(PROJECT_ID, LOCATION)
+
+            client.create_workflow(
+                request={
+                    "parent": parent,
+                    "workflow_id": WORKFLOW_ID,
+                    "workflow": {
+                        "name": WORKFLOW_ID,
+                        "source_contents": workflow_file
+                    },
+                }
+            )
+
+            creating_workflow = True
+
+        # Wait until the workflow is created.
+        print("- Waiting for the Workflow to be created...")
+        time.sleep(backoff_delay)
+        # Double the delay to provide exponential backoff.
+        backoff_delay *= 2
+
+    yield WORKFLOW_ID
+
+    # Delete the workflow.
+    workflow_full_name = client.workflow_path(
+        PROJECT_ID, LOCATION, WORKFLOW_ID
+    )
+
+    client.delete_workflow(
+        request={
+            "name": workflow_full_name,
+        }
+    )

workflows/cloud-client/main.py

@@ -1,4 +1,4 @@
-# Copyright 2021 Google LLC
+# Copyright 2025 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,60 +12,70 @@
 # See the License for the specific language governing permissions and
 # 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, workflow: str) -> Execution:
+def execute_workflow(
+    project_id: str,
+    location: str,
+    workflow_id: 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.
 
     Args:
-        project: The Google Cloud project id
+        project: The ID of the Google Cloud project
             which contains the workflow to execute.
-        location: The location for the workflow
+        location: The location for the workflow.
         workflow: The ID of the workflow to execute.
 
     Returns:
         The execution response.
     """
+
+# [START workflows_api_quickstart]
+    import time
+
+    from google.cloud import workflows_v1
+    from google.cloud.workflows import executions_v1
+
+    from google.cloud.workflows.executions_v1.types import executions
+
+    # TODO(developer): Update and uncomment the following lines.
+    # project_id = "MY_PROJECT_ID"
+    # location = "MY_LOCATION"  # For example: us-central1
+    # workflow_id = "MY_WORKFLOW_ID"  # For example: myFirstWorkflow
+
     # [START workflows_api_quickstart_client_libraries]
-    # Set up API clients.
+    # Initialize API clients.
     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)
+    parent = workflows_client.workflow_path(project_id, location, workflow_id)
 
     # 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
+    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_finished = execution.state != executions.Execution.State.ACTIVE
 
-        # If we haven't seen the result yet, wait a second.
+        # If we haven't seen the result yet, keep waiting.
         if not execution_finished:
             print("- Waiting for results...")
             time.sleep(backoff_delay)
@@ -74,11 +84,13 @@ def execute_workflow(project: str, location: str, workflow: str) -> Execution:
         else:
             print(f"Execution finished with state: {execution.state.name}")
             print(f"Execution results: {execution.result}")
-            return execution
     # [END workflows_api_quickstart_execution]
+# [END workflows_api_quickstart]
+            return execution
 
 
 if __name__ == "__main__":
+    PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")
     assert PROJECT, "'GOOGLE_CLOUD_PROJECT' environment variable not set."
-    execute_workflow(PROJECT, LOCATION, WORKFLOW_ID)
-# [END workflows_api_quickstart]
+
+    execute_workflow(PROJECT, "us-central1", "myFirstWorkflow")

workflows/cloud-client/main_test.py

@@ -12,52 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
+import backoff
 
-from google.cloud import workflows_v1
 from google.cloud.workflows.executions_v1.types import executions
 
 import main
 
-PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")
-LOCATION = "us-central1"
-WORKFLOW_ID = "myFirstWorkflow"
 
-
-def test_workflow_execution() -> None:
-    assert PROJECT, "'GOOGLE_CLOUD_PROJECT' environment variable not set."
-
-    if not workflow_exists():
-        workflow_file = open("myFirstWorkflow.workflows.yaml").read()
-
-        workflows_client = workflows_v1.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, LOCATION, WORKFLOW_ID)
+@backoff.on_exception(backoff.expo, AssertionError, max_time=60)
+def test_workflow_execution(project_id: str, location: str, workflow_id: str) -> None:
+    result = main.execute_workflow(project_id, location, workflow_id)
     assert result.state == executions.Execution.State.SUCCEEDED
-    assert result.result  # Result not empty
-
-
-def workflow_exists() -> bool:
-    """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
-        )
-        workflows_client.get_workflow(request={"name": workflow_name})
-        return True
-    except Exception as e:
-        print(f"Workflow doesn't exist: {e}")
-        return False
+    assert result.result

workflows/cloud-client/noxfile_config.py

@@ -22,7 +22,7 @@
 
 TEST_CONFIG_OVERRIDE = {
     # You can opt out from the test for specific Python versions.
-    "ignored_versions": ["2.7", "3.7", "3.8", "3.10", "3.11"],
+    "ignored_versions": ["2.7", "3.7", "3.8"],
     # Old samples are opted out of enforcing Python type hints
     # All new samples should feature them
     "enforce_type_hints": True,

workflows/cloud-client/requirements-test.txt

@@ -1 +1,2 @@
-pytest==8.2.0
+pytest==8.3.5
+backoff==2.2.1

workflows/cloud-client/requirements.txt

@@ -1 +1 @@
-google-cloud-workflows==1.15.1
+google-cloud-workflows==1.18.0