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.

Prefect Community

Hello everybody. I have a Flow that create_flow_run (a few) and waits until the finish - wait_for_flow_run. But if one of the sub-flow is Failed parent flow has a Success state anyway. Can I make parent flow Failed if one of the children fails?

Hey <@U01A1PA4F99>, I see, maybe you can try the `StartFlowRun` task with `StartFlowRun(…, wait=True)`? I think this should fail the flow as it returns the state.

Hey, just chatted with the team and the task state should match the child flow run state. But it will only affect the parent for if it’s a reference task so you need to set it as a reference task if you want it to fail the flow

Hi <@U01QEJ9PP53> I am having the same problem as Mark. I tried setting the reference tasks in the parent flow of my fof with "parent_flow.set_reference_tasks([hello_run_id, world_run_id]) but even when the both child flows fail, the parent flow succeeds. Is there something I am missing?

Did you use `create_flow_run` or `StartFlowRun` ?

Yes I used create_flow_run but not StartFlowRun

Ah I see. I think `StartFlowRun` might be better for you with `wait=True`. Could you try that? I think it propagates the state

Code for <https://github.com/PrefectHQ/prefect/blob/8fcb9aa97d739fa25405c60cbb28b649c130bd3e/src/prefect/tasks/prefect/flow_run.py#L438-L445|that>

StartFlowRun is working in terms of getting the parent flow to fail, but now my child flows are running all at once. How can I get them to run in order? <@U01QEJ9PP53>

I'm just using the default local executor I believe

That should be sequential. Maybe you can just set upstream dependencies in the order which you want them to run?

that's the .set_upstream method right? where's the api for that?

You can do
```with Flow(...) as flow:
     a = StartFlowRun(...)()
     b = StartFlowRun(...)(upstream_tasks=[a])```
or
```with Flow(...) as flow:
     a = StartFlowRun(...)()
     b = StartFlowRun(...)()

     b.set_upstream(a)```

perfect thank you <@U01QEJ9PP53> for all your help !

Hey <@U01QEJ9PP53> - following up on this thread as I believe my question is related... in the docs <https://docs.prefect.io/core/idioms/flow-to-flow.html> it provides the example of initiating flow_a--&gt;d StartFlowRun tasks. The one thing I'd like to confirm is that by including flow_a as an upstream task to flow_b means that it will start flow_a even though its not listed as a task within the actual Flow?

```with Flow("parent-flow", schedule=weekday_schedule) as flow:
    b = flow_b(upstream_tasks=[flow_a])
    c = flow_c(upstream_tasks=[flow_a])
    d = flow_d(upstream_tasks=[b, c])```

```with Flow("parent-flow", schedule=weekday_schedule) as flow:
    flow_a = flow_a()
    b = flow_b(upstream_tasks=[flow_a])
    c = flow_c(upstream_tasks=[flow_a])
    d = flow_d(upstream_tasks=[b, c])```

I believe it will because this will work for other tasks as well where you can pass them by the name to `upstream_tasks` and it will take care of adding it to the Flow but I honestly prefer the second because it is explicit