I have been trying to deploy an async flow, but I am pretty sure I'm missing something. Using this snippet from Anna:

import asyncio
import random
import time

from prefect import flow, task, get_run_logger


async def async_func():
    if random.random() > 0.3:
        time.sleep(1)
    await asyncio.sleep(random.randint(1, 5))
    return 1


@task
async def my_task():
    result = await async_func()
    return result


@flow()
async def my_flow():
    tasks = []
    for _ in range(500):
        t = my_task()
        tasks.append(t)

    await asyncio.gather(*tasks, return_exceptions=True)


if __name__ == "__main__":
    asyncio.run(my_flow())

How can one create a deployment for this? Would you still be able to use

Deployment.build_from_flow()

with this? And if so, would this still work with deployment parameters?

hi @Aaron Gonzalez - there's nothing that strikes me as out of the ordinary here, besides the large number of tasks. did you run into any specific troubles trying to create a deployment? I was able to using your code with

prefect deployment build --name testasync testing-prefect/testasync.py:my_flow -a

and i assume

Deployment.build_from_flow()

would work just the same

🤔

hmm i'd guess that has to do with the fact that its trying to spin up 500 tasks at the same time - I'd think its a matter of resources on the machine trying to run that flow. do you anticipate having a use case similar to this? or are you stress testing

Something similar but probably much larger. https://prefect-community.slack.com/archives/CM28LL405/p1676909231085509

gotcha - I'd probably recommend chunking the tasks in groups of N (like you suggested) and distributing them across worker flows that each have their own infrastructure i have a recipe like that here i like that pattern because it works even if you don't know what your total number of tasks is going to be, you can just figure out what resources a worker flow needs to handle N tasks and then create

(total number of tasks) / N

worker flows

oh nice!

woo!