[Python] Dynamic DAG: Use expand input(list) in task

I need some help passing the date on to the task generated by the expand function. I am generating dynamic tasks based on the start and end dates. Based on these start/ end dates, a list of dates will be generated within the specified range. Once the list of dates is generated, this list will be passed to the expand() function. A custom operator executes a Python script and I want to provide each date from the generated list to the Python script(custom-date) via the expanded task instances. Below is the code:

def DYNAMIC():

@task
def generate_dates(**kwargs):
params = kwargs.get('params', {})
start_date = params.get('start_date')
end_date = params.get('end_date')

start_date = datetime.strptime(start_date, "%Y-%m-%d")
end_date = datetime.strptime(end_date, "%Y-%m-%d")
# Generate list of dates
return [(start_date + timedelta(days=i)).strftime("%Y-%m-%d") for i in range((end_date - start_date).days + 1)]


start = EmptyOperator(task_id=f'START')
end = EmptyOperator(task_id=f'END')

date_list = generate_dates()


task = CustomOperator.partial(
task_id=f'TASK192',
command=['python', 'xyz.py',
f'--custom-date={d}'], # d should be the date from expanded list


trigger_rule=TriggerRule.ALL_SUCCESS,
).expand(date=date_list)


I have tried to pass the command in the expand function and loop over date_list but it didn't work. Any help would be highly appreciated.

Continue reading...