@Marvin how do I await an async task list from a sync flow with prefect?

To await an asynchronous task list from a synchronous flow in Prefect, you can use the

submit

method to run async tasks within a sync flow. Here's a basic example:

python
from prefect import flow, task

@task
async def my_async_task():
    pass

@flow
def my_sync_flow():
    async_results = [my_async_task.submit() for _ in range(5)]
    for result in async_results:
        # Use `result.result()` to block and get the result of the async task
        result.result()

my_sync_flow()

In this example,

my_async_task

is an asynchronous task that you want to run multiple times. The

submit

method is used to schedule these tasks, and

result.result()

is used to block and retrieve the results synchronously. You can find more about task submission and flow execution in the Prefect tasks documentation. If you have more specific requirements or configurations, feel free to ask!

what is the best way to gather the results from a sync flow? Calling 'result' will block the sync code

some_task.map(range(N)).result()

where yes .result is blocking but you're collecting all of them at once unlike marvin's sequentially blocking example

@Nate as_completed from asyncio?

sorry, i meant

from prefect.futures

https://github.com/prefecthq/prefect/blob/main/src/prefect/futures.py#L343-L381 so the resulting list of futures from map (or list comp of submits) can be provided to that, and then you can process them as they finish

ohhh, generator yielding completed futures! Sweet, thanks!

i'd be happy to look at an MRE!

working on it, thanks!