hi all, i'm trying to write a flow where all the tasks should execute in a sequential order but some of them should optionally be skipped. basic example in normal python:

print("hello world 1")
if not skip_task2:
    print("hello world 2")
if not skip_task3:
    print("hello world 3")
print("hello world 4")

here's an attempt in Prefect:

@task()
def print_task(value: str) -> str:
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(value)
    return value


with Flow("case_skip_flow") as flow:
    skip_task2 = Parameter("skip_task2", True)
    skip_task3 = Parameter("skip_task3", True)

    task1 = print_task("hello world 1")
    with case(skip_task2, False):
        task2 = print_task("hello world 2", upstream_tasks=[task1])
    with case(skip_task3, False):
        task3 = print_task(
            "hello world 3", upstream_tasks=[task1, task2], task_args={"skip_on_upstream_skip": False}
        )
    task4 = print_task("hello world 4", upstream_tasks=[task3], task_args={"skip_on_upstream_skip": False})

i'm getting into trouble with the upstream skip status for

task3

. • if I set

skip_on_upstream_skip

to

True

, then when

task2

is skipped, so is

task3

regardless of the value for

skip_task3

• if I set

skip_on_upstream_skip

to

False

, then

task3

runs regardless of the value for

skip_task3

how can i implement a pattern like this?

Hi @Dominick Olivito, looking now

thanks for taking a look. i tried a bit more, and it looks like if I put in an intermediate dummy task, then I can hack this and make this work:

@task()
def dummy_task() -> None:
    pass


with Flow("case_skip_flow") as flow:
    skip_task2 = Parameter("skip_task2", True)
    skip_task3 = Parameter("skip_task3", False)

    task1 = print_task("hello world 1")
    with case(skip_task2, False):
        task2 = print_task("hello world 2", upstream_tasks=[task1])
    dummy_result = dummy_task(upstream_tasks=[task2], task_args={"skip_on_upstream_skip": False})

    with case(skip_task3, False):
        task3 = print_task(
            "hello world 3", upstream_tasks=[dummy_result]
        )
        task4 = print_task("hello world 4", upstream_tasks=[task3])

yep definitely will be because you can just call

.wait()

and use the native Python if