Let's say I have three tasks that have absolutely no data dependency but must run in sequential order. Of course I could just make one function wrapping those three instructions. But let's assume they could fail due to external dependencies and I wanted to profit from

prefect

's reporting on failed tasks. Is this the best way to describe the relation?

from prefect import Flow, task


@task
def task1():
    pass


@task
def task2():
    pass


@task
def task3():
    pass


with Flow("Linear Flow") as flow:
    task1().set_downstream(task2().set_downstream(task3()))

flow.run()

This does run the tasks in order, but it is not very readable/scalable - in my "real" example, I have around 8 logical steps which should run in order.

And it becomes even more complicated when we have some data dependencies in between. Here is an example of four tasks where task two should fan out into instantiations of task three but task four should wait for all task three examples to finish:

from prefect import Flow, task


@task
def task1():
    pass


@task
def task2():
    return [0, 1, 2, 3]


@task
def task3(number):
    print(number)


@task
def task4():
    pass


with Flow("Linear Flow") as flow:
    intermediate_output = task2().set_upstream(task1)
    task4().set_upstream(task3.map(intermediate_output))

flow.run()

Again - is this the way to go? It seems somewhat convoluted and in the example of 8 sequential steps with one step fanning out, I guess my reviewers would not like this approach very much 🤔

You can also call

flow.get_tasks

with a filter condition to get all the tasks added to the flow as a list if you don't want to track them manually