Hey folks, I have just started diving into prefect and I am trying to migrate some code from luigi One of the problems I am facing is that we used the dynamic dependency system of luigi, so a task could

yield

other tasks as dependents. I was looking for similar functionality and the closest I found is the

LOOP

but the problem is that this permits only one direct dependency while I need to create multiple. I also tried to represent the problem on the flow level but unfortunately it does not work in my case. The task dependencies depend on the value of parameters so this is not known until a flow is executed. Any ideas here what is the prefect way to tackle this?

Hi @Sque! Welcome to the Prefect Community 😄 You’re looking for Mapping: https://docs.prefect.io/core/concepts/mapping.html

Hi @Dylan! Thanks for finding this amazing community here:) I checked on the mapping but do not understand exactly how it will help me here. So putting this as code it could be something like this if the flow was dynamic e.g.

@task
def process_date(date: datetime):
    # .. do an action on this date

with Flow('dynamic') as flow:
    date = DateTimeParameter('date')
    sub_tasks = [process_date(sub_date) 
                 for sub_date in get_childs_for_this_date(date)]
    main_task = process_date(date, upstream_tasks=sub_tasks)

So the problem with the above is that

get_childs_for_this_date

returns different number of dates depending on the value. I could unfold the whole problem inside

process_date

and process all sub_dates and main date under one task. But I loose all the parallelism here and this process takes a lot of time. Ideally, I would like this flow to be scheduled every X and given the current time pass this as parameter

hmmm