@Marvin, can I run

asyncio.run(async_function())

in a Prefect 2 sub flow?

Yes, you can run

asyncio.run(async_function())

in a Prefect 2 subflow. Asynchronous subflows can be run in parallel by using AnyIO task groups or asyncio.gather. Here is an example:

import asyncio
from prefect import flow

@flow
async def subflow_1():
    print("Subflow 1 started!")
    await asyncio.sleep(1)

@flow
async def subflow_2():
    print("Subflow 2 started!")
    await asyncio.sleep(1)

@flow
async def subflow_3():
    print("Subflow 3 started!")
    await asyncio.sleep(1)

For more details, you can refer to this discussion.

@Marvin what would the main flow look like that calls those sub flows?

The main flow that calls these subflows would look something like this:

from prefect import Flow

@Flow
async def main_flow():
    await subflow_1.run()
    await subflow_2.run()
    await subflow_3.run()

This will run the subflows sequentially. If you want to run them in parallel, you can use `asyncio.gather`:

from prefect import Flow
import asyncio

@Flow
async def main_flow():
    await asyncio.gather(subflow_1.run(), subflow_2.run(), subflow_3.run())

Please note that the

run

method is used to execute the subflows.

@Marvin "the

run

method is used to execute the subflows." untrue friend

I apologize for the confusion. In Prefect 2, you don't use the

run

method to execute flows or subflows. Instead, you just call the flow like a function. Here's how you would do it:

from prefect import Flow

@Flow
async def main_flow():
    await subflow_1()
    await subflow_2()
    await subflow_3()

And for parallel execution:

from prefect import Flow
import asyncio

@Flow
async def main_flow():
    await asyncio.gather(subflow_1(), subflow_2(), subflow_3())

I hope this clears up the misunderstanding.

Thanks Nate 😄