https://prefect.io logo
m

Moritz Rocholl

08/04/2023, 12:57 PM
@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?
m

Marvin

08/04/2023, 12:58 PM
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:
Copy code
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.
n

Nate

08/04/2023, 2:08 PM
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
m

Moritz Rocholl

08/04/2023, 2:17 PM
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
n

Nate

08/04/2023, 2:25 PM
gotcha, you can always use the client
Copy code
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
or just use run_deployment with timeout=0 to return the the flow run ID immediately and filter for that later
🙌 1
m

Moritz Rocholl

08/04/2023, 2:37 PM
that sounds like the best option as of now. thx for the help 🙂
👍 2
2 Views