Get_scheduled_flow_runs_for_work_pool is returning unexpected results

Help
, ,
1

I am running this simple flow to get upcoming/scheduled flow runs:

import asyncio
from datetime import datetime, timedelta
from prefect import get_client, flow


async def upcoming_runs():
    async with get_client() as client:
        work_pools = await client.read_work_pools()
        flow_runs = []
        for work_pool in work_pools:
            response = await client.get_scheduled_flow_runs_for_work_pool(
                work_pool_name=work_pool.name,
                # scheduled_before=datetime.today() + timedelta(days=1),
            )
            flow_runs.extend(response)
    return flow_runs


@flow
def print_upcoming_flow_runs():
    flow_runs = asyncio.run(upcoming_runs())
    for run in flow_runs:
        print(run)


if __name__ == "__main__":
    print_upcoming_flow_runs()

This used to work until last week, but now it doesn’t return any results. I can see the scheduled flow runs that should be returned in the Prefect UI, but not when I run this flow from my workstation.

We bumped our version of Prefect and restarted our agents last week, but I am not sure how this could have caused the change in behavior.

What should I look into to troubleshoot this? What additional information can I share that might shed light on what’s going on?

2

Seems like there is some kind of server-side bug in get_scheduled_flow_runs_for_work_pool.

We’ve switched to read_flow_runs with the appropriate filters to get scheduled flow runs, and it seems to work as expected. A teammate got the idea to use this method after noting that the Prefect UI uses it under the hood (and shows the flow runs we expect).

1 Like