With prefect 2, I'm seeing issues when attempting to get flow groups via the api (using the cloud pr...

Alix Cook

09/19/2022, 3:09 PM

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.

✅ 1

Alix Cook

09/19/2022, 3:10 PM

e.g. I'm making the request like:

Copy code

flow_runs = await client.read_flow_runs(
        flow_run_filter=FlowRunFilter(
            state=FlowRunFilterState(type=FlowRunFilterStateType(any_=state_types)),
        )
    )

And I get this error:

Copy code

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>

Alix Cook

09/19/2022, 3:10 PM

but if i remove the flow run filter, its fine

Bianca Hoch

09/19/2022, 3:23 PM

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.

Alix Cook

09/19/2022, 3:24 PM

👍

Mason Menges

09/19/2022, 4:11 PM

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

Alix Cook

09/19/2022, 4:14 PM

[StateType.SCHEDULED]

was what

state_types

was set to. I haven't seen that call pass this morning (in like the last 2 hours)

Mason Menges

09/19/2022, 4:18 PM

Thanks 😄 You should be able to pass the state types as strings so I'd try that, this code is currently working for me

Copy code

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
        )

gratitude thank you 1

🚀 1

Alix Cook

09/19/2022, 4:22 PM

ic, thanks!

🙌 1

Alix Cook

09/19/2022, 5:17 PM

It seems that when i have my filter just have

PENDING

, it fails

Copy code

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

but if i also add something else, it works:

Alix Cook

09/19/2022, 5:17 PM

Copy code

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

Mason Menges

09/19/2022, 6:46 PM

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.

Alix Cook

09/19/2022, 7:45 PM

Copy code

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"]))

Alix Cook

09/19/2022, 7:46 PM

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

Alix Cook

09/19/2022, 7:47 PM

and version:

Copy code

prefect --version
2.3.1

8 Views

Open in Slack

Previous Next

Prefect Community

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.