Getting "invalid list filter" when filtering GCP Workflows using Python

Question

I have a process where I need to list off failed workflows in Google cloud platform so they can be highlighted to be fixed. I have managed to do this quite simply writing a gcloud command and calling it in a shell script, but I need to transfer this to Python.

I have the following shell command as an example where I am able to filter on a specific workflow to pull back any failures using the --filter flag.

gcloud workflows executions list --project test-project "projects/test-project/locations/europe-west4/workflows/test-workflow"  --location europe-west4 --filter STATE:FAILED

In the documentation for filtering it not possible to do this on the executions but on the list workflows instead, which is fine. You can see within the code snippet below I am trying to filter on STATE:FAILED like in the gcloud command.

The example below doesn't work and there are no examples within the Google cloud documents. I have checked in the following webpage:

https://cloud.google.com/python/docs/reference/workflows/latest/google.cloud.workflows_v1.types.ListWorkflowsRequest

from google.cloud import workflows_v1
from google.cloud.workflows import executions_v1

# Create a client
workflow_client = workflows_v1.WorkflowsClient()
execution_client = executions_v1.ExecutionsClient()
project = "test-project"
location = "europe-west4"
# Initialize request argument(s)
request = workflows_v1.ListWorkflowsRequest(
    parent=f"projects/{project}/locations/{location}",
    filter="STATE:FAILED"
)
# Make the request
workflow_page_result = workflow_client.list_workflows(request=request)
# Handle the response
with open("./workflows.txt", "w") as workflow_file:
    for workflow_response in workflow_page_result:
        name = workflow_response.name
        request = executions_v1.ListExecutionsRequest(
            parent=name,
        )
        execution_page_result = execution_client.list_executions(request=request)
        # Handle the response
        for execution_response in execution_page_result:
            print(execution_response)
        workflow_file.write(name)

What is the correct syntax for filtering on the failed state within the Python code? Where would I look to find this information in the Google documentation?

I get the following error message:

/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/bin/python /Users/richard.drury/code/gcp/git/service-level-monitoring/daily_checks/sensor_checks/workflows.py
Traceback (most recent call last):
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/lib/python3.10/site-packages/google/api_core/grpc_helpers.py", line 50, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/lib/python3.10/site-packages/grpc/_channel.py", line 946, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/lib/python3.10/site-packages/grpc/_channel.py", line 849, in _end_unary_response_blocking
    raise _InactiveRpcError(state)
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
    status = StatusCode.INVALID_ARGUMENT
    details = "The request was invalid: invalid list filter: Field 'STATE' not found in 'resource'."
    debug_error_string = "{"created":"@1670416298.880340000","description":"Error received from peer ipv4:216.58.213.10:443","file":"src/core/lib/surface/call.cc","file_line":967,"grpc_message":"The request was invalid: invalid list filter: Field 'STATE' not found in 'resource'.","grpc_status":3}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/daily_checks/sensor_checks/workflows.py", line 15, in <module>
    workflow_page_result = workflow_client.list_workflows(request=request)
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/lib/python3.10/site-packages/google/cloud/workflows_v1/services/workflows/client.py", line 537, in list_workflows
    response = rpc(
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/lib/python3.10/site-packages/google/api_core/gapic_v1/method.py", line 154, in __call__
    return wrapped_func(*args, **kwargs)
  File "/Users/richard.drury/code/gcp/git/service-level-monitoring/venv/lib/python3.10/site-packages/google/api_core/grpc_helpers.py", line 52, in error_remapped_callable
    raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.InvalidArgument: 400 The request was invalid: invalid list filter: Field 'STATE' not found in 'resource'. [field_violations {
  field: "filter"
  description: "invalid list filter: Field \'STATE\' not found in \'resource\'."
}
]

Process finished with exit code 1

Answer 1

In a YAML Workflow using the Workflows Executions API and after a lot of trial and error testing different filter values (because I could not find examples in the documentation) I was able to get a list of active executions using this workflow step:

      - getActiveExecutions:
          call: http.get
          args:
            auth:
              type: OAuth2
            url: ${"https://workflowexecutions.googleapis.com/v1/projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/locations/" + sys.get_env("GOOGLE_CLOUD_LOCATION") + "/workflows/" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_ID") + "/executions"}
            query:
                filter: state="ACTIVE"
          result: activeExecutions

This may work in your python example, but you may need to use the filter value like below because the quotes around the state value were important:

request = workflows_v1.ListWorkflowsRequest(
    parent=f"projects/{project}/locations/{location}",
    filter='state="FAILED"'
)