Hi I m trying to create a conditional Flow of Flows that run

Question

Hi I m trying to create a conditional Flow of Flows that runs a flow and then based on that flow s final state either runs `flow if success` or `flow if failure` Basically I want to know how I can take the final state signal from `conditional flow` and decide whether I want to run `flow if success` or `flow if failure` I was having trouble matching the task signal with a value so that the case works properly I was hoping you would be able to help me out Would I need to use the `get task run result` or is there a better way I can just grab the task result Here is a snippet of my current test code ```var1 = Parameter var1 default = 4 var2 = Parameter var2 default = 12 conditional id = StartFlowRun flow name= Conditional Flow project name= Test wait=True flow if success id = StartFlowRun flow name= Flow if success project name= Test wait=True flow if failure id = StartFlowRun flow name= Flow if failure project name= Test wait=True Idea Conditional flow runs first If it succeeds with Success signal run flow if success if it fails with failure signal run flow if failure with Flow Conditional FoF as parent flow conditional run = conditional id with case conditional run Success Conditional run never matches Success even if successful task run flow if success run = flow if success id parameters=dict num=var1 with case conditional run Failed flow if failure run = flow if failure id parameters=dict number=var2 ```

Anna Geller · Answer

You would need to switch to create flow run + wait for flow run and include a case task to implement this condition Here is a bit different use case but you may use a similar syntax and building blocks <https discourse prefect io t can i run a flow of flows that triggers each child flow on a different machine 180>

Nick Hart · Answer

Alright I ll try this out Thank you

Anna Geller · Answer

and if you have any questions about it let us know happy to provide more examples when needed

Nick Hart · Answer

So I m wondering is there a way to not have a case task in my parent flow and just have a way to match the cond flow final signal to something Basically the problem I get is that it skips the case because of the following output `SKIP signal raised SKIP Provided value FlowRunView flow run id= d0bc26fe 9693 4890 832e 182d5035a90e name= Conditional Flow Run 2022 02 07 19 22 47 235833 state= lt Success All reference tasks succeeded gt labels= lt BoxList gt cached task runs=0 did not match lt class prefect engine state Success gt `

Anna Geller · Answer

to get a state from a FlowRunView you need to add ` state` Example ```with Flow parent flow as flow infra = check the infrastructure with case infra Docker with label machine 1 child flow run id = create flow run flow name= child flow name run config=DockerRun labels= machine 1 with a specific condition like image child flowrunview = wait for flow run child flow run id raise final state=True stream logs=True child flowrunview state state of the task you can use in some downstream task to determine what should happen next based on this task```

Nick Hart · Answer

I was trying that before but it gives me `AttributeError FunctionTask object has no attribute state ` error Is there something I need to import to get that to work

Anna Geller · Answer

Can you share your flow

Nick Hart · Answer

```import prefect from prefect import Flow Parameter case from prefect engine state import Failed Success from prefect storage import Docker from prefect tasks prefect import RenameFlowRun create flow run wait for flow run from prefect executors import LocalDaskExecutor import datetime logger = prefect context get logger var1 = Parameter var1 default = 4 var2 = Parameter var2 default = 12 def renameFlowRun obj new state old state d = datetime datetime now if new state is running param value = Conditional FoF format d RenameFlowRun run flow run name=f param value return with Flow Conditional FoF state handlers= renameFlowRun as parent flow conditional id = create flow run flow name= Conditional Flow project name= Test conditional run = wait for flow run conditional id raise final state=True stream logs=True <http logger info|logger info> Conditional task State +str conditional run state with case conditional run state Success flow if success id = create flow run flow name= Flow if success project name= Test parameters=dict num=var1 flow if success run = wait for flow run flow if success id raise final state=True stream logs=True with case conditional run state Failed flow if failure id = create flow run flow name= Flow if failure project name= Test parameters=dict num=var2 flow if failure run = wait for flow run flow if failure id raise final state=True stream logs=True parent flow executor = LocalDaskExecutor parent flow storage = Docker image name= prefect conditionalfof image tag= 7feb2022 files= home test Documents conditional fof py usr local share prefect conditional fof py stored as script=True local image=True path= usr local share prefect conditional fof py prefect directory= usr local share prefect if name == main parent flow run ```

Anna Geller · Answer

This is not as straightforward so I can totally understand why this is difficult to configure Here is one idea of how you can approach this ```import prefect from prefect import Flow Parameter task Task from prefect engine state import Success from prefect storage import Docker from prefect tasks prefect import RenameFlowRun create flow run wait for flow run from prefect executors import LocalDaskExecutor import datetime def rename flow run obj new state old state d = datetime datetime now if new state is running param value = Conditional FoF format d RenameFlowRun run flow run name=f param value return task def trigger conditional flow runs flow run view if isinstance flow run view state Success flow if success id = create flow run flow name= Flow if success project name= Test parameters=dict num=prefect context parameters get var1 run flow if success run = wait for flow run flow if success id raise final state=True stream logs=True run else flow if failure id = create flow run flow name= Flow if failure project name= Test parameters=dict num=prefect context parameters get var2 flow if failure run = wait for flow run flow if failure id raise final state=True stream logs=True with Flow Conditional FoF state handlers= rename flow run storage=Docker image name= prefect conditionalfof image tag= 7feb2022 files= home test Documents conditional fof py usr local share prefect conditional fof py stored as script=True local image=True path= usr local share prefect conditional fof py prefect directory= usr local share prefect executor=LocalDaskExecutor as parent flow var1 = Parameter var1 default=4 var2 = Parameter var2 default=12 conditional id = create flow run flow name= Conditional Flow project name= Test conditional run = wait for flow run conditional id raise final state=True stream logs=True trigger conditional flow runs conditional run parent flow add task var1 parent flow add task var2 if name == main parent flow visualize ``` Does this pattern work for you

Nick Hart · Answer

It looks like it should but for some reason it gave me `ValueError Could not infer an active Flow context while creating edge to lt Task create flow run gt This often means you called a task outside a `with Flow ` block If you re trying to run this task outside of a Flow context you need to call `create flow run run `` and I tried putting run in different spots in the trigger conditional flows task but it didn t seem to help I ll keep trying again tomorrow morning Thank you for all your help i appreciate it I ll let you know if I get it working

Anna Geller · Answer

< Nick Hart> with a help from < Kevin Kho> I found a much better solution Here is an example flow you may try ```from prefect import Flow task from prefect tasks prefect import create flow run wait for flow run from prefect triggers import all successful any failed from random import random task log stdout=True def say hi flow name print f Hi from flow name task log stdout=True def randomly fail nr = random print f Number nr if nr gt 0 5 raise ValueError Big number with Flow child1 as child1 say hello = say hi child1 randomly fail upstream tasks= say hello with Flow child2 as child2 say hi child2 with Flow child3 as child3 say hi child3 with Flow conditional flow of flows as parent flow hi = say hi Parent flow child1 id = create flow run flow name= child1 project name= xyz task args=dict name= Main Flow Run upstream tasks= hi conditional run = wait for flow run child1 id raise final state=True stream logs=True task args=dict name= Condition on success of conditional run child2 id = create flow run flow name= child2 project name= xyz upstream tasks= conditional run task args=dict name= Flow run on Success trigger=all successful conditional run child2 = wait for flow run child2 id raise final state=True stream logs=True on failure of conditional run child3 id = create flow run flow name= child3 project name= xyz upstream tasks= conditional run task args=dict name= Flow run on Failure trigger=any failed conditional run child3 = wait for flow run child3 id raise final state=True stream logs=True parent flow set reference tasks hi if name == main parent flow visualize child1 register xyz child2 register xyz child3 register xyz parent flow register xyz ``` It works exactly as you wanted based on a condition it runs the respective child flow

Anna Geller · Answer

btw here is a <https discourse prefect io t how to build a conditional flow of flows i e trigger a different child flow depending on the upstream flows state 202|transcription> of this discussion if you need it for a later reference

Nick Hart · Answer

oh no way this is awesome Yeah that s exactly what I m looking for I just tested it on my machine and it works Thank you again I appreciate all the help