So I have a task which outputs the list of ranges for which I’d like to create multiple flow runs. What I want to do is: a). map those outputs to flow runs, where each of them have upstream dependency on previous, something like:

@task
def prepare_request(batch_size=200):
    # querying two databases to get last update ids, then

    return list(range(first_id, second_id))

with Flow('import-scheduled') as flow:

    load_ids = prepare_request()
    wflr = None

    for id in load_ids:
        flr = create_flow_run()
        if wflr:
            flr.set_upstream(wflr)
        wflr = wait_for_flow_run()
        wflr.set_upstream(flr)

the problem here is that it seems I can’t just iterate task result, and I can’t use just method without task decorator as using db connection secrets from context

Another use case is even trickier - ideally I would like to launch 3-4 flows, and set their wait_for_flow_run tasks as upstream dependencies for another 3-4 flows, and repeat for all range ids

This is how it looks on UI (on use cases where range ids are hardcoded)

for instance for loop and if/else statements are not supported in Prefect 1 but they are fully supported in Prefect 2: docs.prefect.io