I m looking to get the flow run id of a task running in asyn Prefect Community #ask-community

I'm looking to get the flow_run_id of a task runni...

Stephen Lloyd

03/29/2023, 8:18 AM

I'm looking to get the flow_run_id of a task running in async.

Copy code

@task
def run_single_flow(deployment_name: str):
    return run_deployment(name=deployment_name)

@flow
def orchestrate_flows():
    logger = get_run_logger()
    flow_name = "scheduler"
    <http://logger.info|logger.info>(contexts.build_flow_run_url())
    deployments = [
        "my-flow/some_other_flow",
        "another-flow/some_flow",
        "final-flow/some_final_flow",
    ]
    futures = []
    for d in deployments:
        run = run_single_flow.submit(d)
        #figure out how to get the flow_run_id her
        <http://logger.info|logger.info>(run.something.flow_run_id)
        futures.append(run)

    waiting = [f.wait() for f in futures]
    for x in waiting:
        <http://logger.info|logger.info>(x.result().id)

I can get this if I

wait()

for the result, but that doesn't help me as much.

👀 1

Dominic Tarro

03/29/2023, 1:05 PM

Perhaps if you construct the flow run from the deployment yourself you can know the flow run ID rather than leaving it to autogeneration? Break it into two functions: create_flow_run and run_single_flow.

Stephen Lloyd

03/30/2023, 4:09 AM

Sheepishly, I feel like I'm missing something. It seems like the

run_deployment

function both creates and runs a flow run. If you have a moment, could you flesh out what you mean a little more?

Stephen Lloyd

03/30/2023, 4:17 AM

@Emil Christensen I'd love any insights you have here. I need to give our platform support staff a way to easily process and navigate what is going on in a flow-of-flows situation (not subflows). Things that don't work so far • Logs from a flow run created from

run_deployment

do not show up in the calling flow. • Creating and logging a url-string to a flow run created by

run_deployment()

◦ I cannot seem to find the

flow_run_id

prior to flow run reach a completed sate ◦ This seems like an opportunity for improvement! • Just hoping I will figure it out

👀 1

Emil Christensen

03/30/2023, 5:34 PM

Hey @Stephen Lloyd 👋 when calling

run_deployment

you can pass

timeout=0

which will immediately return the

flow_run

where you can get the ID with

flow_run.id

Stephen Lloyd

04/06/2023, 3:19 AM

Hey @Emil Christensen Thanks for your help on this!

❤️ 1

Stephen Lloyd

04/11/2023, 10:50 AM

I put this in, but there seems to be a downside. After using

timeout=0

the State no longer updates.

my_flow_run.wait()

does not seem to behave as intended. The flow seems to think the wait is over. 🙂 Is this to be expected?

3 Views

Open in Slack

Previous Next