Hey @Marvin how does one asynchronously await a flow run state result in Prefect 2.x?

In Prefect 2.x, to asynchronously await the result of a flow run state, you can use the

State.result()

method with async/await syntax. This allows you to wait for the task or flow to complete and then retrieve its result. Here's a concise example:

python
import asyncio
from prefect import flow, task

@task
async def my_task():
    return 1

@flow
async def my_flow():
    state = await my_task(return_state=True)
    result = await state.result(fetch=True)
    return result + 1

async def main():
    state = await my_flow(return_state=True)
    assert await state.result(fetch=True) == 2

asyncio.run(main())

### Key Points: - When inside a flow, you can retrieve the state result of a task asynchronously by using

await state.result(fetch=True)

. -

fetch=True

ensures the retrieval of persisted results, which was introduced in Prefect 2.6.0. - Make sure you run the main async function using

asyncio.run()

to properly execute the async flow. For more detailed information, you can refer to the Prefect 2.x documentation.