@Marvin I'm submitting task in my flow using

task = task_name.submit()

and then i'm awaiting them with

await asyncio.wait(task)

However it looks like the await returns immediately and not when the task is actually finished. How can I await that the task is finished?

It seems you're using

await asyncio.wait(task)

incorrectly for awaiting task completion in Prefect. In Prefect 3.x, the

submit

method of a task returns a

PrefectFuture

object, which you can await directly to ensure the task has completed. Here's how you can modify your code:

python
from prefect import flow, task

@task
def task_name():
    # Task logic here
    pass

@flow
async def my_flow():
    task_future = task_name.submit()
    await task_future

my_flow()

This approach uses the

PrefectFuture

object returned by

submit

, which can be awaited directly to wait for task completion. For more details, you can refer to the task_runners.py and task_runs.py for more insights on task submission and execution. If you're interested in more detailed documentation about asynchronous execution in Prefect, you might find this Prefect 3.x upgrade guide helpful.

I am using Prefect 2.0; and this is my code:

import asyncio
import time

@task
async def my_task():
    time.sleep(10)


@flow()
async def my_flow():
    tasks = []
    for i in range(2):
        tasks.push(asyncio.Task(my_task.submit())
    done, _ = await asyncio.wait(tasks)

I would expect the

await

to wait for any of the tasks to finish (so at least 10 seconds), but it seems to return immediately