@Marvin: Given a list of PrefectFutures what is the non-blocking way to get all the results? Something like dask

client.gather()

To get the results of a list of

PrefectFuture

objects in a non-blocking way, similar to Dask's

client.gather()

, you can use asynchronous flows and tasks in Prefect. Here's an example demonstrating how to use async flows and tasks to achieve this:

import asyncio
from prefect import flow, task

@task
async def async_task(n):
    await asyncio.sleep(1)
    return f"Task {n} result"

@flow
async def async_flow():
    futures = [async_task.submit(i) for i in range(5)]
    results = await asyncio.gather(*[future.result() for future in futures])
    return results

if __name__ == "__main__":
    results = asyncio.run(async_flow())
    print(results)

Explanation: 1. Define an asynchronous task using

@task

. 2. In your flow, submit multiple tasks and collect their futures. 3. Use

asyncio.gather()

to collect results in a non-blocking manner. 4. Run the flow with

asyncio.run()

. For further details, you can refer to the Prefect documentation on task runners and results. Additionally, you can check this example for more insights into working with async workflows in Prefect.