refactor: update typehints in workflows/cloud-client

workflows/cloud-client/main.py

@@ -12,27 +12,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
+# [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 import Execution
+from google.cloud.workflows.executions_v1.types import executions
 
 
 def execute_workflow(
-    project, location="us-central1", workflow="myFirstWorkflow"
-):
-    """Execute a workflow and print the execution results."""
-    # [START workflows_api_quickstart]
-    import time
+    project: str, location: str = "us-central1", workflow: str = "myFirstWorkflow"
+) -> Execution:
+    """Execute a workflow and print the execution results.
 
-    from google.cloud import workflows_v1
-    from google.cloud.workflows import executions_v1
-    from google.cloud.workflows.executions_v1.types import executions
+    A workflow consists of a series of steps described using the Workflows syntax, and can be written in either YAML or JSON.
 
-    # TODO(developer): Uncomment these lines and replace with your values.
-    # project = 'my-project-id'
-    # location = 'us-central1'
-    # workflow = 'myFirstWorkflow'
+    Args:
+        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.
 
-    if not project:
-        raise Exception('GOOGLE_CLOUD_PROJECT env var is required.')
+    Returns:
+        The execution response.
+    """
 
     # Set up API clients.
     execution_client = executions_v1.ExecutionsClient()
@@ -48,25 +51,21 @@ def execute_workflow(
     # 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})
+    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...')
+            print("- Waiting for results...")
             time.sleep(backoff_delay)
             # Double the delay to provide exponential backoff.
             backoff_delay *= 2
         else:
-            print(f'Execution finished with state: {execution.state.name}')
-            print(f'Execution results: {execution.result}')
+            print(f"Execution finished with state: {execution.state.name}")
+            print(f"Execution results: {execution.result}")
             return execution
-    # [END workflows_api_quickstart]
 
 
-if __name__ == "__main__":
-    project = os.environ.get('GOOGLE_CLOUD_PROJECT')
-    execute_workflow(project=project)
+# [END workflows_api_quickstart]

workflows/cloud-client/main_test.py

@@ -24,36 +24,33 @@
 WORKFLOW_ID = "myFirstWorkflow"
 
 
-def test_workflow_execution():
+def test_workflow_execution() -> None:
     assert PROJECT != ""
 
     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
+        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 result.state == executions.Execution.State.SUCCEEDED
     assert len(result.result) > 0
 
 
-def workflow_exists():
-    """Returns True if the workflow exists in this project
-    """
+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)
+        workflow_name = workflows_client.workflow_path(PROJECT, LOCATION, WORKFLOW_ID)
         workflows_client.get_workflow(request={"name": workflow_name})
         return True
     except Exception as e:

workflows/cloud-client/noxfile_config.py

@@ -0,0 +1,42 @@
+# Copyright 2022 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.
+
+# Default TEST_CONFIG_OVERRIDE for python repos.
+
+# You can copy this file into your directory, then it will be imported from
+# the noxfile.py.
+
+# The source of truth:
+# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/noxfile_config.py
+
+TEST_CONFIG_OVERRIDE = {
+    # You can opt out from the test for specific Python versions.
+    "ignored_versions": ["2.7"],
+    # Old samples are opted out of enforcing Python type hints
+    # All new samples should feature them
+    "enforce_type_hints": True,
+    # An envvar key for determining the project id to use. Change it
+    # to 'BUILD_SPECIFIC_GCLOUD_PROJECT' if you want to opt in using a
+    # build specific Cloud project. You can also use your own string
+    # to use your own Cloud project.
+    "gcloud_project_env": "GOOGLE_CLOUD_PROJECT",
+    # 'gcloud_project_env': 'BUILD_SPECIFIC_GCLOUD_PROJECT',
+    # If you need to use a specific version of pip,
+    # change pip_version_override to the string representation
+    # of the version number, for example, "20.2.4"
+    "pip_version_override": None,
+    # A dictionary you want to inject into your test. Don't put any
+    # secrets here. These values will override predefined values.
+    "envs": {},
+}