Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

So say I have some flows in a class:

```class SomeClass:

    def init(self):
           

               …
@flow(task_runner=DaskTaskRunner())
def __runmodel(self, *args):
     code to run model on dask 

@flow
def gen_args_run_model(self):
    args_list = #code to gen arg_list
    #arg_list  is a list of tuples which have args
    try:
         results=[self._runmodel(*args) for args in args_list]
   except:
           ….```
I want to create a deployment for gen_args_run_model that I can then run and it will kick off the run_model sub flow  on dask with each iteration of args in arg list, how would I go about this? 

Is there a more efficient way to do this? 

Im guessing I would do 
```asyncio.gather(*results) ```
And make the self._run_model function async? 

I had async before implementing prefect and anything greater than 20 being submitted would timeout and fail

So I added batching and subflow deployment hopefully that’ll work

i guess i have to seperate out my class logic a bit