https://prefect.io logo
Title
s

Sharath Chandra

04/23/2022, 3:01 PM
Hi, I am using prefect master flow to orchestrate multiple flows. I wanted to check if I can have a trigger the downstream flow once my upstream flow is completed irrespective of Success or Failure
register_graph_flow_run_id = create_flow_run(
        flow_name=register_graph.flow.name, project_name=project_name
    )

    wait_for_register_graph_flow_run = wait_for_flow_run(
        flow_run_id=register_graph_flow_run_id, raise_final_state=True
    )

    register_participants_flow_run_id = create_flow_run(
        flow_name=register_participants.flow.name,
        project_name=project_name,
        parameters={"Run date (before offset)": run_date, "Date offset": date_offset},
    )

    wait_for_register_participants_flow_run = wait_for_flow_run(
        flow_run_id=register_participants_flow_run_id, raise_final_state=True
    )

    register_participants_flow_run_id.set_upstream(wait_for_register_graph_flow_run)
k

Kevin Kho

04/23/2022, 3:04 PM
Yes. See Triggers . Specifically
always_run
s

Sharath Chandra

04/23/2022, 3:05 PM
Is this applicable even for
create_flow_run
? I could not see that as parameter on the same
I see that
create_flow_run
returns a task. Is there any way to set this parameter outside the task initialisation
k

Kevin Kho

04/23/2022, 3:06 PM
So there are two things you can do:
from prefect.tasks.prefect import create_flow_run

create_flow_run.trigger = always_run
and the second
with Flow(...) as flow:
    create_flow_run(..., task_args={"trigger": always_run})
task_args lets you change a lot of the init arguments
:upvote: 1
Check task_args here
s

Sharath Chandra

04/23/2022, 3:10 PM
Thanks a lot. Let me try this out
👍 1
@Kevin Kho task_args was a really useful information. Things working as expected now. Thanks.