One more question, does prefect support looping of a subflow? that a part of the flow is repeated? It seems loop only works on a single task, what I have in mind is that I could loop through a chain of tasks.

I believe you could map over a list of flows (that you could treat as sub-flows) by combining map operator and StartFlowRun: https://docs.prefect.io/core/idioms/flow-to-flow.html#scheduling-a-flow-of-flows You could also have a look at this article, which explains how to map over a chain of tasks: https://medium.com/the-prefect-blog/map-faster-mapping-improvements-in-prefect-0-12-0-7cacc3f14e16

Thanks Anna

I believe Anna is right that will work, but you can also loop through tasks with something like this:

sql_list = ["1.sql", "2.sql", "3.sql"]
shell_task = ShellTask(
    helper_script="",
    shell="bash",
    log_stderr=True,
    return_all=True,
    stream_output=True,
)
with Flow(name="Example") as flow:
    tasks = [
        shell_task(
            command="echo '{}'".format(
                statement
            )
        )
        for statement in sql_list
    ]
    for i in range(1, len(tasks)):
        tasks[i].set_upstream(tasks[i - 1])

flow.run()