Anonymous / Dynamic / Runtime Flows in Prefect2

[Samreay]

Hey team!

So heres a fun one for you - anonymous/dynamic flows in Prefect 2.8.3+.

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? I notice this was asked in #6752, but remains unanswered.