I’ve written an example below to demo my scenario. I have a task (

Task1

in demo code) that is upstream of two other tasks (

Task2

and

Task3

).

Task2

only runs on a condition.

Task3

runs, even if that condition fails (

skip_on_upstream_skip=False

). When the condition fails,

Task1

is skipped, such that its value is

None

, breaking the input to

Task3

. I can fix this by explicitly setting

Task1

as downstream of

Task3

(see commented code line) but thought I’d check if this was expected behaviour?

from prefect import Flow, Task, case
from prefect.engine.executors import LocalDaskExecutor


class MyCondition(Task):
    def run(self):
        return False


class Task1(Task):
    def run(self):
        print("task 1 run")
        return "123"


class Task2(Task):
    def run(self, task_1_output: str):
        print(f"task 1 output: {task_1_output}")
        print("task 2 run")


class Task3(Task):
    def run(self, task_1_output: str):
        print(f"task 1 output: {task_1_output}")
        print("task 3 run")


executor = LocalDaskExecutor()

with Flow("test_flow") as flow:
    my_condition = MyCondition()
    task1 = Task1()
    task2 = Task2()
    task3 = Task3(skip_on_upstream_skip=False)

    # task1.set_downstream(task3)

    with case(my_condition, True):
        task2.set_upstream(task1, key="task_1_output")

    task3.set_upstream(task1, key="task_1_output")

flow.run(executor=executor)