Trying to get a conditional workflow written and having trou Prefect Community #ask-community

Trying to get a conditional workflow written and h...

Zach

04/29/2020, 10:39 PM

Trying to get a conditional workflow written and having trouble. At three points in the workflow, I have a task that queries the database for a URI. If the tasks finds a URI then there are 3 subsequent tasks that need to run. Each of the tasks has output that the following tasks needs as input. I was looking at the

ifelse

flow conditional but it doesn't really make sense how I would use it when I have multiple "true_tasks" and they have output that feed into eachother

nicholas

04/29/2020, 10:45 PM

Hi @Zach! I would suggest using a single task in the conditional which you can set as the upstream (kwarg

upstream_tasks=

) for your 3 subequent tasks. If that task doesn't run because of the conditional, the subsequent tasks will be skipped.

nicholas

04/29/2020, 10:46 PM

As for providing just an

if

task, let me test that real quickly

nicholas

04/29/2020, 10:56 PM

Ok just checked and that should work well, here's some dummy code:

Copy code

import prefect
from prefect import Flow, task
from prefect.tasks.control_flow.conditional import ifelse


@task
def proceed():
    # Will proceed on true
    return False


@task
def spawn_branches():
    return "Now THIS is podracing!"


@task
def branch1(x):
    print(x)


@task
def branch2(x):
    print(x)


@task
def branch3(x):
    print(x)


with Flow("Ifelse Flow") as flow:
    ifelse(proceed, spawn_branches, None)

    branch1(spawn_branches)
    branch2(spawn_branches)
    branch3(spawn_branches)

flow.run()

Open in Slack

Previous Next