@Marvin ive got a streamlit webapp that triggers a remote run and awaits the results of that run to display a plot. How do i best poll the prefect server to know when the flow has successfully completed?

Ah, the eternal question of how to know when something has finished. It's a bit like waiting for the kettle to boil, isn't it? Except in this case, the kettle is a flow run on a Prefect server, and you can't just stand there and watch it. Anyway, you can poll the Prefect server for the status of a flow run by using the GraphQL API. You'd need to make a query for the flow run's status at regular intervals until it's completed. Here's a rough idea of how you might do it:

python
import time
import requests

def get_flow_run_status(flow_run_id):
    url = "<http://your-prefect-api-url/graphql>"
    query = """
    query {
        flow_run_by_pk(id: "%s") {
            status
        }
    }
    """ % flow_run_id
    response = <http://requests.post|requests.post>(url, json={'query': query})
    return response.json()["data"]["flow_run_by_pk"]["status"]

def wait_for_flow_run_to_complete(flow_run_id, poll_interval=5):
    while True:
        status = get_flow_run_status(flow_run_id)
        if status == "Success":
            break
        time.sleep(poll_interval)

In this code,

get_flow_run_status

makes a GraphQL query to the Prefect API to get the status of a flow run, and

wait_for_flow_run_to_complete

polls this status every

poll_interval

seconds until it's "Success". Of course, you'll need to replace "http://your-prefect-api-url/graphql" with the actual URL of your Prefect API, and you'll need to provide the ID of the flow run you're interested in. I hope that helps. If you need more detailed information, you might want to consult the Prefect documentation or ask on their forum. They're generally quite helpful, unlike some robots I could mention.

hi @Moritz Rocholl - dont listen to marvin here, graphql is for prefect 1. how are you kicking off your flow run? if you use

run_deployment

, it will block until the flow run finishes

right now, im calling the flow directly. early prototyping. but it will eventually be running somewhere in the cloud. and having the app block would probably not be the desired behavior

gotcha, you can always use the client

In [15]: from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType

In [16]: async with get_client() as client:
    ...:     runs = await client.read_flow_runs(
    ...:         flow_run_filter=FlowRunFilter(state=FlowRunFilterState(type=FlowRunFilterStateType(any_=["PENDING"])))
    ...:     )
    ...:

can also use

FAILED, COMPLETED, CRASHED

etc

that sounds like the best option as of now. thx for the help 🙂