async

async

sync

async

async def

async

sync

await

async

await async_task.map(return_state=True)

asyncio.gather(*[state.result(fetch=True) for state in task_states])

[state.result() for state in task_states]

• I ran into an error where an

async

flow (flow marked with

async def

) would crash when consuming the result of a an

async

subflow. Fixed this by making the flow

sync

and not using

await

when consuming the result of the subflow

import anyio
import asyncio
from prefect import flow


@flow
async def foo():
    return "foo"

@flow
async def bar():
    return "bar"


@flow
async def asyncio_concurrency():
    results = await asyncio.gather(foo(), bar())
    print("".join(results))

asyncio.run(asyncio_concurrency())

@flow
async def anyio_concurrency():
    results = {}
    async def run_and_capture_result(flow):
        results[flow] = await flow()
        
    async with anyio.create_task_group() as tg:
        tg.start_soon(run_and_capture_result, foo)
        tg.start_soon(run_and_capture_result, bar)

    print("".join([results[foo], results[bar]]))

anyio.run(anyio_concurrency)

run_deployment