GoogleCloudPlatform · glasnt · Apr 4, 2025 · Mar 28, 2025 · Mar 28, 2025 · Mar 28, 2025
@@ -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,
+        }
+    )
@@ -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")
@@ -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
@@ -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,

@@ -1 +1,2 @@
-pytest==8.2.0
+pytest==8.3.5
+backoff==2.2.1
@@ -1 +1 @@
-google-cloud-workflows==1.15.1
+google-cloud-workflows==1.18.0
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		google-cloud-workflows==1.15.1
		google-cloud-workflows==1.18.0