Repository navigation
Conversation
|
Here is the summary of changes. You are about to add 1 region tag.
This comment is generated by snippet-bot.
|
There was a problem hiding this comment.
Code Review
This pull request introduces a new sample script get_execution.py and its corresponding tests to retrieve Google Cloud Workflows execution details. The feedback suggests using the client library's built-in execution_path helper method instead of manual string formatting, mocking this method in the unit tests, and conditionally printing the Result and Error fields to prevent confusing output when they are empty.
| name = ( | ||
| f"projects/{project_id}/locations/{location_id}/workflows/{workflow_id}" | ||
| f"/executions/{execution_id}" | ||
| ) |
There was a problem hiding this comment.
Use the client library's built-in execution_path helper method to construct the resource name instead of manual string formatting. This is more robust, idiomatic, and prevents potential formatting errors.
name = client.execution_path(
project=project_id,
location=location_id,
workflow=workflow_id,
execution=execution_id,
)| print(f"Resource ID: {execution.name}") | ||
| # The result is set when the execution succeeds; the error when it fails. | ||
| print(f"Error: {execution.error.payload}") | ||
| print(f"Result: {execution.result}") | ||
| print(f"State: {execution.state.name}") |
There was a problem hiding this comment.
Conditionally print the Result and Error fields only when they are populated. Printing empty/unset fields (e.g., printing Error: for a successful execution or Result: for a failed execution) can be confusing to users running the sample.
| print(f"Resource ID: {execution.name}") | |
| # The result is set when the execution succeeds; the error when it fails. | |
| print(f"Error: {execution.error.payload}") | |
| print(f"Result: {execution.result}") | |
| print(f"State: {execution.state.name}") | |
| print(f"Resource ID: {execution.name}") | |
| print(f"State: {execution.state.name}") | |
| if execution.result: | |
| print(f"Result: {execution.result}") | |
| if execution.error and execution.error.payload: | |
| print(f"Error: {execution.error.payload}") |
| client = mock_client_class.return_value | ||
| client.get_execution.return_value = executions_v1.Execution( | ||
| name=EXECUTION_NAME, | ||
| state=executions_v1.Execution.State.SUCCEEDED, | ||
| result='"done"', | ||
| ) |
There was a problem hiding this comment.
Mock the execution_path method on the client to return the expected execution name. This ensures the unit test remains correct and passes when using the client's built-in path helper.
| client = mock_client_class.return_value | |
| client.get_execution.return_value = executions_v1.Execution( | |
| name=EXECUTION_NAME, | |
| state=executions_v1.Execution.State.SUCCEEDED, | |
| result='"done"', | |
| ) | |
| client = mock_client_class.return_value | |
| client.execution_path.return_value = EXECUTION_NAME | |
| client.get_execution.return_value = executions_v1.Execution( | |
| name=EXECUTION_NAME, | |
| state=executions_v1.Execution.State.SUCCEEDED, | |
| result='"done"', | |
| ) |
Polling or auditing an execution created asynchronously requires fetching its state, result, and error payload by execution ID. Adding `workflows_get_execution` covers `executions_v1.ExecutionsClient.get_execution` with `NotFound` handling.
2857542 to
80308e3
Compare
Description
Callers that trigger workflows asynchronously or inspect past runs need to fetch execution state and output by execution ID.
Adds
get_execution.py(workflows_get_execution) usingexecutions_v1.ExecutionsClient.get_execution, with unit and gated live integration tests.Checklist
Testing
Compliance & Style
workflows/cloud-client/.Post-Approval Actions