Dzmitry Aliashkevich

10/30/2022, 2:32 PM
Hi, folks, I’m struggling to find a solution for one particular use case in Prefect 1.0 regarding upstream/downstream dependencies, details are in thread. Would appreciate any help, thanks
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:
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:
        wflr = wait_for_flow_run()
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)

Anna Geller

10/30/2022, 3:37 PM
You mentioned that you are getting started now - why Prefect 1? Prefect 2 is the default - I would highly recommend starting directly with Prefect 2 Reflecting dependencies is much easier in v2 as well
👍 1
for instance for loop and if/else statements are not supported in Prefect 1 but they are fully supported in Prefect 2:

Dzmitry Aliashkevich

10/31/2022, 6:03 AM
Hi, Anna, thanks for quick response. I might been unclear - it’s me who started recently to use Prefect as better Airflow, but this use case came from a customer who started at the time when v2 wasn’t there. Anyway v2 migration is on our roadmap, so we can postpone this usecase until it happens.

Dzmitry Aliashkevich

10/31/2022, 11:42 AM
Great, will take a look, thank you!
🙌 1