Hey team! So heres a fun one for you - anonymous/dynamic flows. Heres my use case: 1. We have a bunch of endpoints to request data from and save them to an internal DB 2. The data for these endpoints is available one different schedules 3. Each

task

that fetches data returns it in the same

pd.DataFrame

format, so our

save_to_db

task is very general. Now what I was wanting to do was something like the following:

for data_task in list_of_data_tasks:
   @flow(name=f"{data_task.__name__} flow", task_runner=dask_task_runner)
   def anonymous_flow():
      data = data_task.submit()
      save_to_db_task.submit(data)
   Deployment.build_from_flow(...).apply()

I thought “yeah this is good, it saves tons and tons of duplicated

@flow

declarations and duplicated

save_to_db

lines. However, this won’t work because the flow is dynamic, so an agent wont be able to actually find it. What would be the best practise solution for the above?