feat(workflows): modernize samples and add idempotency samples

The Workflows Python samples only ran executions: they covered 2 of the
10 client methods, defined one region tag in two files, used TODO
placeholders, and pinned an old library version. The new idempotency
tutorial also needs a Python sample that's safe to retry.

Client-library samples (workflows/cloud-client), using
google-cloud-workflows 1.24.0:
- Updated: workflows_execute_without_arguments,
  workflows_execute_with_arguments, workflows_api_quickstart (main.py,
  matching the Node.js tab; the duplicate tag and unused sub-tags are
  removed). File paths are unchanged, because docs embed them by path.
- New:
  - workflows_client_library (list workflows, matching Node.js and C++)
  - workflows_create_workflow
  - workflows_get_workflow
  - workflows_update_workflow
  - workflows_delete_workflow
  - workflows_get_execution
  - workflows_list_executions (uses the FULL view, which includes errors)
  - workflows_cancel_execution

Idempotency samples, each a Cloud Run function backed by Firestore:
- idempotent_payment: workflows_idempotent_process_payment and
  workflows_idempotent_order_workflow, for the "Make API calls
  idempotent" tutorial.
  - The function claims an order with a lease and an attempt ID, then
    charges outside the transaction with the order ID as an idempotency
    key.
  - Orders have explicit states.
  - On a duplicate request, the workflow resumes from the stored state
    instead of exiting.
- deduplicate_scheduled_runs: workflows_deduplicate_scheduled_runs
  starts a workflow once per Cloud Scheduler run. It keys on the
  scheduled minute, because a forced run can send the schedule time
  without Cloud Scheduler's per-job offset.
- order_saga: workflows_order_saga_workflow and
  workflows_order_saga_refund refund a charge once when fulfillment
  fails.

Tests live in tests/ subdirectories:
- Unit tests run without credentials.
- Integration tests are skipped unless RUN_INTEGRATION_TESTS is set.
  They cover:
  - duplicate and concurrent executions
  - partial failures
  - ambiguous failures and timeouts at two tiers
  - crashed attempts
  - Cloud Scheduler duplicate deliveries
  - saga compensation
  Each one checks that a stand-in payment service charged exactly once.

Unit and integration tests pass on Python 3.13.

Author: iennae

workflows/cloud-client/cancel_execution.py

@@ -0,0 +1,97 @@
+# Copyright 2026 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.
+
+# [START workflows_cancel_execution]
+import sys
+
+from google.api_core import exceptions
+from google.cloud.workflows import executions_v1
+
+
+def cancel_execution(
+    project_id: str, location_id: str, workflow_id: str, execution_id: str
+) -> None:
+    """Cancels an execution that's still running.
+
+    Args:
+        project_id: The Google Cloud project ID (for example, 'example-project-id').
+        location_id: The region of the workflow (for example, 'us-central1').
+        workflow_id: The ID of the workflow (for example, 'myFirstWorkflow').
+        execution_id: The ID of the execution.
+    """
+    client = executions_v1.ExecutionsClient()
+    name = (
+        f"projects/{project_id}/locations/{location_id}/workflows/{workflow_id}"
+        f"/executions/{execution_id}"
+    )
+
+    try:
+        execution = client.cancel_execution(name=name)
+    except exceptions.NotFound as err:
+        print(
+            f"error: execution '{execution_id}' not found "
+            f"for workflow '{workflow_id}' in location '{location_id}': {err.message}",
+            file=sys.stderr,
+        )
+    except exceptions.GoogleAPICallError as err:
+        print(
+            f"error: failed to cancel execution '{execution_id}' "
+            f"for workflow '{workflow_id}' in location '{location_id}': {err.message}",
+            file=sys.stderr,
+        )
+    else:
+        print(f"Resource ID: {execution.name}")
+        print(f"End Time: {execution.end_time}")
+        print(f"State: {execution.state.name}")
+
+
+# [END workflows_cancel_execution]
+
+
+if __name__ == "__main__":
+    import argparse
+    import os
+
+    # Each flag defaults to its environment variable, then to an example value.
+    parser = argparse.ArgumentParser(
+        description=cancel_execution.__doc__.splitlines()[0]
+    )
+    parser.add_argument(
+        "--project-id",
+        default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"),
+        help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)",
+    )
+    parser.add_argument(
+        "--location-id",
+        default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"),
+        help="Region of the workflow (default: $GOOGLE_CLOUD_LOCATION)",
+    )
+    parser.add_argument(
+        "--workflow-id",
+        default=os.environ.get("WORKFLOW_ID", "myFirstWorkflow"),
+        help="ID of the workflow (default: $WORKFLOW_ID)",
+    )
+    parser.add_argument(
+        "--execution-id",
+        default=os.environ.get("EXECUTION_ID", "example-execution-id"),
+        help="ID of the execution (default: $EXECUTION_ID)",
+    )
+    args = parser.parse_args()
+
+    cancel_execution(
+        project_id=args.project_id,
+        location_id=args.location_id,
+        workflow_id=args.workflow_id,
+        execution_id=args.execution_id,
+    )

workflows/cloud-client/conftest.py

@@ -0,0 +1,120 @@
+# Copyright 2026 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.
+
+# [START workflows_create_workflow]
+import sys
+
+from google.api_core import exceptions
+from google.cloud import workflows_v1
+
+
+def create_workflow(
+    project_id: str, location_id: str, workflow_id: str, service_account_email: str
+) -> None:
+    """Creates a workflow with logging, labels, and environment variables.
+
+    Args:
+        project_id: The Google Cloud project ID (for example, 'example-project-id').
+        location_id: The region of the workflow (for example, 'us-central1').
+        workflow_id: The ID for the new workflow (for example, 'example-workflow').
+        service_account_email: The service account that the workflow runs as.
+    """
+    client = workflows_v1.WorkflowsClient()
+    parent = f"projects/{project_id}/locations/{location_id}"
+
+    # The workflow definition, in YAML. It returns a greeting from an
+    # environment variable that's set on the workflow below.
+    source_contents = """
+main:
+  steps:
+    - greet:
+        return: ${sys.get_env("EXAMPLE_GREETING")}
+"""
+    workflow = workflows_v1.Workflow(
+        source_contents=source_contents,
+        service_account=service_account_email,
+        # Logs each call that fails, without logging successful calls.
+        call_log_level=workflows_v1.Workflow.CallLogLevel.LOG_ERRORS_ONLY,
+        labels={"environment": "example"},
+        user_env_vars={"EXAMPLE_GREETING": "example-greeting"},
+    )
+
+    try:
+        operation = client.create_workflow(
+            parent=parent, workflow=workflow, workflow_id=workflow_id
+        )
+        print("Waiting for the workflow to deploy...")
+        response = operation.result(timeout=300)
+    except exceptions.AlreadyExists as err:
+        print(
+            f"error: workflow '{workflow_id}' already exists "
+            f"in location '{location_id}' for project '{project_id}': {err.message}",
+            file=sys.stderr,
+        )
+    except exceptions.GoogleAPICallError as err:
+        print(
+            f"error: failed to create workflow '{workflow_id}' "
+            f"in location '{location_id}' for project '{project_id}': {err.message}",
+            file=sys.stderr,
+        )
+    else:
+        print(f"Resource ID: {response.name}")
+        print(f"Creation Date: {response.create_time}")
+        print(f"Call Log Level: {response.call_log_level.name}")
+        print(f"Revision ID: {response.revision_id}")
+        print(f"State: {response.state.name}")
+
+
+# [END workflows_create_workflow]
+
+
+if __name__ == "__main__":
+    import argparse
+    import os
+
+    # Each flag defaults to its environment variable, then to an example value.
+    parser = argparse.ArgumentParser(
+        description=create_workflow.__doc__.splitlines()[0]
+    )
+    parser.add_argument(
+        "--project-id",
+        default=os.environ.get("GOOGLE_CLOUD_PROJECT", "example-project-id"),
+        help="Google Cloud project ID (default: $GOOGLE_CLOUD_PROJECT)",
+    )
+    parser.add_argument(
+        "--location-id",
+        default=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"),
+        help="Region of the workflow (default: $GOOGLE_CLOUD_LOCATION)",
+    )
+    parser.add_argument(
+        "--workflow-id",
+        default=os.environ.get("WORKFLOW_ID", "example-workflow"),
+        help="ID for the new workflow (default: $WORKFLOW_ID)",
+    )
+    parser.add_argument(
+        "--service-account-email",
+        default=os.environ.get(
+            "SERVICE_ACCOUNT_EMAIL",
+            "example-runner@example-project-id.iam.gserviceaccount.com",
+        ),
+        help="Service account the workflow runs as (default: $SERVICE_ACCOUNT_EMAIL)",
+    )
+    args = parser.parse_args()
+
+    create_workflow(
+        project_id=args.project_id,
+        location_id=args.location_id,
+        workflow_id=args.workflow_id,
+        service_account_email=args.service_account_email,
+    )