https://prefect.io logo
Title
s

Stephen Lloyd

03/29/2023, 8:18 AM
I'm looking to get the flow_run_id of a task running in async.
@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
d

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.
s

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?
@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
e

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
.
s

Stephen Lloyd

04/06/2023, 3:19 AM
Hey @Emil Christensen Thanks for your help on this!
ā¤ļø 1
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?