Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

With prefect 2, I'm seeing issues when attempting to get flow groups via the api (using the cloud product).
If I try to filter by flow group state, I get a 500 response, but if i omit that the request is fine.

e.g. I'm making the request like:
```    flow_runs = await client.read_flow_runs(
        flow_run_filter=FlowRunFilter(
            state=FlowRunFilterState(type=FlowRunFilterStateType(any_=state_types)),
        )
    )```
And I get this error:
```prefect.exceptions.PrefectHTTPStatusError: Server error '500 Internal Server Error' for url '<https://api.prefect.cloud/api/accounts/c105509e-95a3-458e-ae6c-c944f68863e7/workspaces/1a48caf0-2425-487e-9081-e6a6c36575c0/flow_runs/filter>'
Response: {'exception_message': 'Internal Server Error'}
For more information check: <https://httpstatuses.com/500>```


but if i remove the flow run filter, its fine

Hello Alix, thank you for raising this. We have seen a few reports of users receiving these 500 errors. We currently have an internal ticket open for this and are actively investigating the issue.

Hey <@U03LPUF3HJB> Just to clarify here, what are you passing in to the StateType filter? is this failing for you everytime or just intermittently?

``````

`[StateType.SCHEDULED]` was what `state_types` was set to. I haven't seen that call pass this morning (in like the last 2 hours)

Thanks :smile: You should be able to pass the state types as strings so I'd try that, this code is currently working for me
```test_filter = FlowRunFilter(state=FlowRunFilterState(type=FlowRunFilterStateType(any_=['FAILED', 'CRASHED', 'SCHEDULED'])))

async with get_client() as client:
    flow_runs = await client.read_flow_runs(
    flow_run_filter=test_filter,
    limit=1
        )```

It seems that when i have my filter just have `PENDING`, it fails
```    test_filter = FlowRunFilter(
        state=FlowRunFilterState(type=FlowRunFilterStateType(any_=["PENDING"]))
    )```
but if i also add something else, it works:

```    test_filter = FlowRunFilter(
        state=FlowRunFilterState(
            type=FlowRunFilterStateType(any_=["PENDING", "FAILED"])
        )
    )```


Hmm do you mind elaborating on how it's failing, In my case I'm able to run the code, I don't currently have any Pending flow runs so it just returns None in that case.

```import asyncio
from prefect.client import get_client
from prefect.orion.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateType,
)


async def fetch_matching_flow_runs(
    state_types,
):
    async with get_client() as client:
        flow_runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                state=FlowRunFilterState(type=FlowRunFilterStateType(any_=state_types)),
            )
        )
        return flow_runs


asyncio.run(fetch_matching_flow_runs(["PENDING"]))```


I have 2 pending flows. I can force pending flows by starting a local agent, starting a flow run then ctrl-c-ing before the local agent starts the run

and version:
```prefect --version
2.3.1```
