wait()

wait(futures)

wait()

future.result()

python
from prefect import flow, task
from prefect.futures import wait

@task
def maybe_fail(x):
    if x == 2:
        raise ValueError("boom")
    return x * 10

@flow
def my_flow():
    futures = [maybe_fail.submit(i) for i in range(4)]

    done, not_done = wait(futures)

    print(f"done={len(done)}, not_done={len(not_done)}")

    for f in futures:
        print(f.state)

    # This is where failure would surface:
    return [f.result() for f in futures]

wait(futures)

[f.result() for f in futures]

python
results = []
for f in futures:
    if f.state.is_failed():
        results.append(None)
    else:
        results.append(f.result())

python
results = [f.result(raise_on_failure=False) for f in futures]

raise_on_failure=False

wait()

wait()

wait()

as_completed()

gather()

timeout=None

10800

timeout=None

python
await wait_for_flow_run(flow_run_id, timeout=None)

wait_for_flow_run()

FlowRun

python
flow_run = await arun_deployment(...)
await wait_for_flow_run(flow_run.id, timeout=None, log_states=True)
await flow_run.state.aresult()

flow_run

arun_deployment

SCHEDULED

wait_for_flow_run()

FlowRun

python
flow_run = await arun_deployment(
    name=flow_name,
    flow_run_name=flow_run_name,
    parameters=dict(
        **custom_args,
        s3_output_folder=s3_output_path.as_uri(),
        index_range=index_range,
    ),
    job_variables=jv,
)

append_uuid(flow_run.id)

flow_run = await wait_for_flow_run(flow_run.id, timeout=None, log_states=True)

assert flow_run.state is not None
try:
    await flow_run.state.aresult()
except Exception as e:
    logger.error("pet-wall-aspect-wall-height failed")
    raise

<http://logger.info|logger.info>("pet-wall-aspect-wall-height finished successfully")

SCHEDULED

arun_deployment()

RUNNING

COMPLETED

FAILED

flow_run

flow_run.state.aresult()

SCHEDULED

UnfinishedRun

python
UnfinishedRun('Run is in SCHEDULED state, its result is not available.')

wait_for_flow_run()

aresult()

wait_for_flow_run(..., timeout=None)

arun_deployment(...)

wait_for_flow_run(...)

arun_deployment

python
flow_run = await wait_for_flow_run(flow_run.id, timeout=None, log_states=True)

python
await wait_for_flow_run(flow_run.id, timeout=None, log_states=True)

wait_for_flow_run()

wait_for_flow_run()

wait_for_flow_run()

timeout=None

flow_run.state

SCHEDULED

aresult()

UnfinishedRun

wait_for_flow_run()

python
import asyncio
from prefect.client.orchestration import get_client

async def wait_until_final(flow_run_id, poll_interval=5):
    async with get_client() as client:
        while True:
            flow_run = await client.read_flow_run(flow_run_id)
            if flow_run.state is not None and flow_run.state.is_final():
                return flow_run
            await asyncio.sleep(poll_interval)

python
with LOCK:
    flow_run = await arun_deployment(
        name=flow_name,
        flow_run_name=flow_run_name,
        parameters=dict(
            **custom_args,
            s3_output_folder=s3_output_path.as_uri(),
            index_range=index_range,
        ),
        job_variables=jv,
    )
    append_uuid(flow_run.id)

flow_run = await wait_until_final(flow_run.id)

try:
    await flow_run.state.aresult()
except Exception:
    logger.error("pet-wall-aspect-wall-height failed")
    raise

<http://logger.info|logger.info>("pet-wall-aspect-wall-height finished successfully")

SCHEDULED

RUNNING

FlowRun

wait_for_flow_run()

run_deployment_and_wait()

SCHEDULED

wait_for_flow_run()

read_flow_run(...)

python
async def wait_until_final(flow_run_id, poll_interval=5):
    async with get_client() as client:
        while True:
            flow_run = await client.read_flow_run(flow_run_id)
            if flow_run.state and flow_run.state.is_final():
                return flow_run
            await asyncio.sleep(poll_interval)

wait_for_flow_run()

wait_for_flow_run_strict()

wait_for_flow_run()

arun_deployment(...)

SCHEDULED

wait_for_flow_run(flow_run.id, timeout=None, log_states=True)

flow_run.state

SCHEDULED

aresult()

UnfinishedRun

wait_for_flow_run()

python
import asyncio
from prefect.client.orchestration import get_client

async def wait_until_final(flow_run_id, poll_interval=5):
    async with get_client() as client:
        while True:
            flow_run = await client.read_flow_run(flow_run_id)
            state = flow_run.state
            if state is not None and state.is_final():
                return flow_run
            await asyncio.sleep(poll_interval)

python
flow_run = await arun_deployment(...)
flow_run = await wait_until_final(flow_run.id)
await flow_run.state.aresult()

python
import logging
logging.getLogger("prefect.events.clients").setLevel(logging.DEBUG)

wait_for_flow_run()

wait_for_flow_run()

wait_for_flow_run()

python
import logging
logging.getLogger("prefect.events.clients").setLevel(logging.DEBUG)

python
<http://logger.info|logger.info>(f"waiting for child flow run {flow_run.id}")
updated_flow_run = await wait_for_flow_run(flow_run.id, timeout=None, log_states=True)
<http://logger.info|logger.info>(f"wait returned with state={updated_flow_run.state}")

python
async with get_client() as client:
    fresh = await client.read_flow_run(flow_run.id)
<http://logger.info|logger.info>(f"fresh state after wait: {fresh.state}")

wait_for_flow_run()

wait_for_flow_run()

run_deployment(..., timeout=0)

on_failure

as_completed(tasks)

timeout=0

SCHEDULED

RUNNING

on_failure

wait_for_tasks()

python
async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        await task.state.aresult()

task.state.aresult()

state

task.state

as_completed(tasks)

state

await task.result()

task.result()

task.state.aresult()

python
async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        await task.result()

python
async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        state = await task.wait()
        await task.result()

python
async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        await task.result()

PrefectFuture.state

python
await task.state.aresult()

aresult()

State

task.result()

on_failure

wait_for_flow_run(child_id, timeout=None)

python
async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        await task.state.aresult()

2. Do not depend on `wait_for_flow_run()` as the only child-wait primitive for critical logic; use polling.

3. Avoid reading `.state.aresult()` on stale flow-run objects; refresh first or use the future/result abstraction.

4. Reconsider global bookkeeping for child run IDs.

A more robust task pattern would look like:

```python
import asyncio
from prefect import task
from prefect.client.orchestration import get_client
from prefect.deployments.flow_runs import run_deployment

async def wait_until_final(flow_run_id, poll_interval=5):
    async with get_client() as client:
        while True:
            flow_run = await client.read_flow_run(flow_run_id)
            if flow_run.state and flow_run.state.is_final():
                return flow_run
            await asyncio.sleep(poll_interval)

@task
async def launch_and_wait(...):
    flow_run = await run_deployment(
        name="my-deployment",
        parameters={...},
        timeout=0,
    )

    append_uuid(flow_run.id)

    flow_run = await wait_until_final(flow_run.id)

    await flow_run.state.aresult()
    return flow_run.id

python
async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        await task.result()

wait_for_flow_run()

.state

task.state.aresult()

result()

prefect.futures.as_completed()

.result()

.result()

python
from prefect.futures import as_completed

async def wait_for_tasks_fail_fast(tasks):
    for task in as_completed(tasks):
        await task.result()

await task.result()

python
from prefect.futures import as_completed

def wait_for_tasks_fail_fast(tasks):
    for task in as_completed(tasks):
        task.result()

as_completed(tasks)

task.result()

wait(tasks)

.result()

python
from prefect.futures import as_completed, PrefectFuture

async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        await task.result()

.result()

await task.result()

python
from prefect.futures import as_completed, PrefectFuture

async def wait_for_tasks(tasks: list[PrefectFuture]):
    for task in as_completed(tasks):
        task.result()

python
await task.result()

as_completed(tasks)

task.result()

task.result()

python
from prefect.futures import as_completed

async def wait_for_tasks_fail_fast(tasks):
    for task in as_completed(tasks):
        task.result()

await

.result()

task.result()

as_completed()

try/except

python
from prefect.futures import as_completed, PrefectFuture
from prefect.client.orchestration import get_client

# however you store these
CHILD_FLOW_RUN_IDS: set[str] = set()

async def cancel_child_flow_runs(flow_run_ids):
    async with get_client() as client:
        for flow_run_id in flow_run_ids:
            try:
                await client.cancel_flow_run(flow_run_id)
            except Exception:
                # optional: log and continue
                pass

async def wait_for_tasks_fail_fast(tasks: list[PrefectFuture]):
    try:
        for task in as_completed(tasks):
            task.result()  # raises immediately if this task failed
    except Exception:
        await cancel_child_flow_runs(CHILD_FLOW_RUN_IDS)
        raise

python
from prefect.futures import as_completed, PrefectFuture
from prefect.client.orchestration import get_client

CHILD_FLOW_RUN_IDS: set = set()

async def cancel_child_flow_runs(flow_run_ids):
    async with get_client() as client:
        for flow_run_id in flow_run_ids:
            try:
                await client.cancel_flow_run(flow_run_id)
            except Exception:
                pass

async def wait_for_tasks_fail_fast(tasks: list[PrefectFuture]):
    try:
        for task in as_completed(tasks):
            task.result()
    except Exception:
        await cancel_child_flow_runs(CHILD_FLOW_RUN_IDS)
        raise

python
from prefect.futures import as_completed
from prefect.client.orchestration import get_client

async def cancel_child_flow_runs(flow_run_ids, logger):
    async with get_client() as client:
        for flow_run_id in flow_run_ids:
            try:
                <http://logger.info|logger.info>(f"Cancelling child flow run {flow_run_id}")
                await client.cancel_flow_run(flow_run_id)
            except Exception as exc:
                logger.warning(f"Failed to cancel child flow run {flow_run_id}: {exc}")

async def wait_for_tasks_fail_fast(tasks, child_flow_run_ids, logger):
    try:
        for task in as_completed(tasks):
            task.result()
    except Exception as exc:
        logger.error(f"A task failed, cancelling remaining child flow runs: {exc}")
        await cancel_child_flow_runs(child_flow_run_ids, logger)
        raise

task.result()

as_completed()

python
await cancel_child_flow_runs(set(CHILD_FLOW_RUN_IDS), logger)

client.cancel_flow_run

Cancelling

set_flow_run_state(...)

python
from prefect.client.orchestration import get_client
from prefect.states import Cancelling

async def cancel_flow_run(flow_run_id):
    async with get_client() as client:
        await client.set_flow_run_state(
            flow_run_id,
            Cancelling(message="Cancelled because a sibling task failed"),
        )

client.cancel_flow_run(...)

python
from prefect.client.orchestration import get_client
from prefect.states import Cancelling

async def cancel_child_flow_runs(flow_run_ids):
    async with get_client() as client:
        for flow_run_id in flow_run_ids:
            try:
                await client.set_flow_run_state(
                    flow_run_id,
                    Cancelling(message="Cancelled because a sibling task failed"),
                )
            except Exception:
                pass

Cancelling

.submit(wait_for=...)

PrefectFuture[CoroutineType[Any, Any, None]]

submit()

async def

Task.submit(...)

Coroutine[...]

list[PrefectFuture[CoroutineType[Any, Any, None]]]

return_state=False

python
next_task.submit("arg", wait_for=list_of_tasks, return_state=False)

python
from typing import cast
from prefect.futures import PrefectFuture

deps = cast(list[PrefectFuture[None]], list_of_tasks)
next_task.submit("arg", wait_for=deps)

python
next_task.submit("arg", wait_for=list_of_tasks)  # type: ignore[arg-type]

return_state=False

cast(...)

submit()

future.task_name

PrefectFuture

task_run_id

task_run.name

.submit()

future.task_run_id

python
from prefect.client.orchestration import get_client
from prefect.futures import as_completed
from prefect.logging import get_run_logger

async def wait_for_tasks_with_logging(tasks):
    logger = get_run_logger()

    async with get_client() as client:
        for future in as_completed(tasks):
            task_run = await client.read_task_run(future.task_run_id)

            try:
                future.result()
                <http://logger.info|logger.info>(f"Task completed: {task_run.name}")
            except Exception:
                logger.error(f"Task failed: {task_run.name}")
                raise

future.state.name

Completed

Failed

future.task_run_id

future.result()

as_completed(...)

python
futures_to_names = {
    my_task.submit(x): f"my_task({x})"
    for x in items
}

for future in as_completed(list(futures_to_names)):
    name = futures_to_names[future]
    future.result()
    <http://logger.info|logger.info>(f"Task completed: {name}")

client.read_task_run(future.task_run_id).name

future -> label

wait_for_tasks_fail_fast_with_logging()

as_completed

task_run_id

PrefectFuture

python
from prefect.futures import as_completed
from prefect.logging import get_run_logger

async def wait_for_tasks_with_logging(tasks_by_name):
    logger = get_run_logger()

    for future in as_completed(list(tasks_by_name.keys())):
        name = tasks_by_name[future]
        try:
            future.result()
            <http://logger.info|logger.info>(f"Task completed: {name}")
        except Exception:
            logger.error(f"Task failed: {name}")
            raise

python
tasks_by_name = {}

for i in range(10):
    future = my_task.submit(i)
    tasks_by_name[future] = f"my_task({i})"

await wait_for_tasks_with_logging(tasks_by_name)

as_completed

dict[PrefectFuture, str]

python
from prefect.futures import as_completed
from prefect.logging import get_run_logger

async def wait_for_tasks_fail_fast(tasks_by_name):
    logger = get_run_logger()

    for future in as_completed(list(tasks_by_name.keys())):
        name = tasks_by_name[future]
        try:
            future.result()
            <http://logger.info|logger.info>(f"Task completed: {name}")
        except Exception:
            logger.exception(f"Task failed: {name}")
            raise

list[PrefectFuture]

list[str]

Flow

prefect.tags(...)

python
from prefect import flow, tags

@flow
def my_flow():
    pass

with tags("prod", "important"):
    my_flow()

tags=

@flow

with tags(...)

@task(tags={...})

python
from prefect import tags

with tags("foo", "bar"):
    flow_obj()

serve()

with tags(...)

python
from prefect import flow, tags
from prefect.runtime import flow_run

@flow
def level_3():
    print(flow_run.tags)

@flow
def level_2():
    print(flow_run.tags)
    level_3()

@flow
def level_1():
    print(flow_run.tags)
    level_2()

with tags("prod", "batch"):
    level_1()

python
with tags("outer"):
    with tags("inner"):
        level_1()

outer

inner