Great questions. In Prefect 3.x: - Calling another

@flow

inside a flow creates a subflow run and is blocking by default. - To keep looping when a subflow fails, either catch exceptions or request the subflow’s state and inspect it. - To run a limited number of subflows in parallel, wrap the subflow call in a task and submit it under a

ThreadPoolTaskRunner(max_workers=N)

. Here are two patterns you can drop into your code. 1) Keep the loop going even if some subflows fail (sequential)

from prefect import flow
from prefect.states import State

@flow(name="AMAZON-FORECAST-DATA", log_prints=True, task_runner=ThreadPoolTaskRunner(max_workers=2))
def run():
    amaz_seller_list = extract_amazon_sellers()

    for amz_seller in amaz_seller_list[:2]:
        # Option A: try/except
        try:
            extract_amazon_sales(amz_seller)  # subflow call, blocks until done
        except Exception as e:
            print(f"extract_amazon_sales failed for {amz_seller}: {e}")
            continue

        # Option B: inspect state (no exception raised at call site)
        # state: State = extract_amazon_sales(amz_seller, return_state=True)
        # if state.is_failed():
        #     print(f"extract_amazon_sales failed for {amz_seller}: {state.message}")
        #     continue

2) Run a limited number of subflows in parallel (at most N at a time) and tolerate failures

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

# Wrap the subflow call in a task so it can run under the task runner
@task
def run_sales_subflow(amz_seller):
    # Any exception here will be captured in the task's state
    return extract_amazon_sales(amz_seller)  # still creates a child flow run

@flow(name="AMAZON-FORECAST-DATA", log_prints=True, task_runner=ThreadPoolTaskRunner(max_workers=2))
def run():
    amaz_seller_list = extract_amazon_sellers()

    futures = [run_sales_subflow.submit(s) for s in amaz_seller_list[:2]]

    # Collect results without crashing the whole flow if one fails
    results = []
    for f in futures:
        res = f.result(raise_on_failure=False)  # returns Exception if the task/subflow failed
        if isinstance(res, Exception):
            print(f"extract_amazon_sales failed: {res}")
            results.append(None)
        else:
            results.append(res)

    return results

Notes: - Direct subflow calls are sequential and will raise on failure unless you use

return_state=True

(and then check

state.is_failed()

), or you wrap them in a

try/except

. - For parallelism, wrapping the subflow in a

@task

and using

.submit()

allows the parent flow’s

ThreadPoolTaskRunner(max_workers=N)

to limit how many run concurrently. - If you later need to pass a failed upstream future into another task without failing it, you can wrap that future with

allow_failure(...)

when submitting the downstream task.

Here’s your flow adapted to: - run subflows with a max of N in parallel (controlled by ThreadPoolTaskRunner) - continue even if some subflows fail - keep success/failure counters and print progress

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import as_completed

# Wrap the subflow call in a task so it can run concurrently
@task
def run_sales_subflow(amz_seller):
    # This still creates a child flow run for the subflow
    return extract_amazon_sales(amz_seller)

@flow(name="AMAZON-FORECAST-DATA", log_prints=True, task_runner=ThreadPoolTaskRunner(max_workers=2))
def run():
    amaz_seller_list = extract_amazon_sellers()
    targets = amaz_seller_list[:5]
    total = len(targets)

    complete_counter = 0
    failed_counter = 0

    # Submit all work; concurrency is limited by max_workers above
    future_to_seller = {}
    for idx, amz_seller in enumerate(targets, start=1):
        f = run_sales_subflow.submit(amz_seller)
        future_to_seller[f] = amz_seller
        if idx % 10 == 0 or idx == total:
            print(f"total submitted {idx}/{total}")

    # Process results as each subflow finishes
    for processed_count, f in enumerate(as_completed(list(future_to_seller.keys())), start=1):
        amz_seller = future_to_seller[f]
        result = f.result(raise_on_failure=False)  # do not raise on failure

        if isinstance(result, Exception):
            failed_counter += 1
            print(f"{amz_seller} - Extraction failed {failed_counter}/{total} | processed {processed_count}")
        else:
            complete_counter += 1
            print(f"{amz_seller} - Extraction completed {complete_counter}/{total} | processed {processed_count}")

    print(f"Finished. Success={complete_counter}, Failed={failed_counter}, Total={total}")

How it works: - The

ThreadPoolTaskRunner(max_workers=2)

caps concurrent subflows at 2. Adjust

max_workers

to change the parallelism. - Wrapping the subflow call in a

@task

and using

.submit()

enables concurrent execution. - Using

f.result(raise_on_failure=False)

lets the loop continue even if a subflow fails. You’ll get either a real result or an Exception instance for each future.