import asyncio
from prefect import task, flow

@task
async def print_x(x):
  print(x)
  await asyncio.sleep(1)

def build_subflow(name):
  @flow(name=f"subflow:{name}")
  async def subflow(x):
    await print_x(x)
  return subflow

@flow
async def parent_flow():
  futures = [build_subflow(name=x)(x) for x in ["a", "b", "c"]]
  await asyncio.gather(*futures)

import asyncio
from prefect import task, flow, get_run_logger


@task
async def print_x(x):
    logger = get_run_logger()
    <http://logger.info|logger.info>(x)
    await asyncio.sleep(1)


def build_subflow():
    @flow
    async def subflow(x):
        await print_x(x)

    return subflow


@flow
async def parent_flow():
    futures = [build_subflow()(x) for x in ["x1", "x2", "x3"]]
    await asyncio.gather(*futures)


if __name__ == "__main__":
    asyncio.run(parent_flow())

return_exceptions=True

asyncio.gather

states = await asyncio.gather(*subflows, return_exceptions=True)
    if all(state.is_failed() for state in states):
        raise ValueError("🚒 All subflows failed")

StateGroup

all_completed

any_failed

state.type

<http://StateType.XYZ|StateType.XYZ>

otherwise if any subflow raised an exception, the whole parent flow fails immediately