Hi all, I’m trying to set a scheduled dependency b...
# prefect-community
j
Hi all, I’m trying to set a scheduled dependency between flows so that flow B starts when flow A finishes (after being scheduled by time). I found this: https://docs.prefect.io/core/idioms/flow-to-flow.html but it’s not very clear to me how I can implement it.
a
I actually built many examples for this on Discourse - check this tag. Especially this one dives deeper into various options on how you may approach it. Here is a simple example:
Copy code
from datetime import timedelta
from prefect import Flow
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run

PARENT_FLOW_NAME = "parent_flow_example"
PREFECT_PROJECT_NAME = "community"


with Flow(PARENT_FLOW_NAME) as parent_flow:
    flow_a_run_id = create_flow_run(
        flow_name="Flow_A",
        project_name=PREFECT_PROJECT_NAME,
        task_args=dict(name="Flow A"),
    )
    flow_a_flowrunview = wait_for_flow_run(
        flow_a_run_id,
        raise_final_state=True,
        stream_logs=True,
        task_args=dict(name="Wait for Flow A"),
    )

    flow_b_run_id = create_flow_run(
        flow_name="Flow_B",
        project_name=PREFECT_PROJECT_NAME,
        upstream_tasks=[flow_a_flowrunview],
        task_args=dict(name="Flow B"),
    )
    flow_b_flowrunview = wait_for_flow_run(
        flow_b_run_id,
        raise_final_state=True,
        stream_logs=True,
        task_args=dict(name="Wait for Flow B"),
    )
    flow_c_run_id = create_flow_run(
        flow_name="Flow_C",
        project_name=PREFECT_PROJECT_NAME,
        upstream_tasks=[flow_b_flowrunview],
        task_args=dict(name="Flow C"),
    )
    flow_c_flowrunview = wait_for_flow_run(
        flow_c_run_id,
        raise_final_state=True,
        stream_logs=True,
        task_args=dict(name="Wait for Flow C"),
    )
if __name__ == "__main__":
    parent_flow.visualize()
150 Views