https://prefect.io logo
Title
z

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
n

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.
As for providing just an
if
task, let me test that real quickly
Ok just checked and that should work well, here's some dummy 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()