https://prefect.io
Join Slack
is it possible to create a helper function/task that implements both create_flow_run and wait_for_fl...
# ask-community
h
is it possible to create a helper function/task that implements both create_flow_run and wait_for_flow_run? I tried:
Copy code
@task()
def flow_run_helper(flow_name, project_name):
    cfr = create_flow_run(flow_name=flow_name, project_name=project_name)
    wfr = wait_for_flow_run(cfr, stream_logs=True, raise_final_state=True)
    return wfr
but its yelling at me about "ValueError: Could not infer an active Flow context while creating edge". my app does a lot of chaining of flows, so i wanted to streamline this
a
we have many examples using that pattern here https://discourse.prefect.io/tag/wait_for_flow_run
k
StartFlowRun(..., wait=True)
does it in one go
upvote 2
a
Copy code
from prefect import Flow
from prefect.tasks.prefect import StartFlowRun

start_flow_run = StartFlowRun(project_name="PROJECT_NAME", wait=True)

with Flow("FLOW_NAME") as flow:
    staging = start_flow_run(flow_name="child_flow_name")
h
i dont think i can use StartFlowRun and still get all of my child flow logs appearing in the main flow
I currently have to do all my chaining like this to get the behavior i want out of the dashboard:
Copy code
sd_id = create_flow_run(flow_name='sync_data', project_name="proj")
sd_wait = wait_for_flow_run(sd_id, stream_logs=True, raise_final_state=True)

pd_id = create_flow_run(flow_name='process_data', project_name="proj")
pd_wait = wait_for_flow_run(pd_id, stream_logs=True, raise_final_state=True)
pd_id.set_upstream(sd_wait)

ed_id = create_flow_run(flow_name='export_data', project_name="proj")
ed_wait = wait_for_flow_run(ed_id, stream_logs=True, raise_final_state=True)
ed_id.set_upstream(pd_wait)
a
check this Discourse page, we have 6 examples showing how you can do that https://discourse.prefect.io/tag/wait_for_flow_run
k
Ah I see. You function will though if you call 
.run()
. To use a task inside another task:
Copy code
@task
def abc():
    bcd.run(...)
    cde.run(...)
but note it’s not a task. it’s just the Python function underneath
✅ 1
h
awesome, i think that's what i was looking for. ty
59 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by