https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> in this pattern how do I get the pare...
# ask-marvin
m
@Marvin in this pattern how do I get the parent flow to fail if one the sub flows failed?
Copy code
for storage_key in storage_keys:
        <http://logger.info|logger.info>(f"Starting global pipeline for storage_key: {storage_key} and date: {partition_date}")
        futures.append(global_pipeline.submit_to_work_pool(
            prefect_env=prefect_env,
            tasks=tasks if tasks else ['global_tasks'],
            storage_key=storage_key,
            partition_date=partition_date,
            other_args=other_args,
        ))
    for future in futures:
        future.wait()
m
✅ thought for 123.9 seconds
Yes — 
future.wait()
alone is not enough. 
wait()
just blocks until the subflow run reaches a final state. It does not raise if that subflow failed. If you want the parent flow to fail when any submitted subflow fails, call 
future.result()
after waiting, or just call 
result()
directly. Example:
Copy code
for storage_key in storage_keys:
    <http://logger.info|logger.info>(
        f"Starting global pipeline for storage_key: {storage_key} and date: {partition_date}"
    )
    futures.append(
        global_pipeline.submit_to_work_pool(
            prefect_env=prefect_env,
            tasks=tasks if tasks else ["global_tasks"],
            storage_key=storage_key,
            partition_date=partition_date,
            other_args=other_args,
        )
    )

for future in futures:
    future.result()
Why this works: - 
future.wait()
→ waits for completion only - 
future.result()
→ waits if needed and raises the underlying exception if the subflow failed So if any subflow run ends in 
Failed
, 
future.result()
will raise, and the parent flow will fail too. If you want to log which one failed:
Copy code
for storage_key, future in zip(storage_keys, futures):
    try:
        future.result()
    except Exception as exc:
        logger.error(f"Subflow failed for storage_key={storage_key}: {exc}")
        raise
If you want to wait for all of them first and then fail if any failed, you can inspect state:
Copy code
failed = []

for storage_key, future in zip(storage_keys, futures):
    future.wait()
    if future.state.is_failed():
        failed.append((storage_key, future.state))

if failed:
    raise RuntimeError(
        "One or more subflows failed: "
        + ", ".join(storage_key for storage_key, _ in failed)
    )
But in most cases, the simplest and best pattern is just:
Copy code
for future in futures:
    future.result()
A couple of references: - Prefect futures behavior and 
result()
/ 
wait()
are covered in the Python API docs: prefect.futures - General Prefect 3 docs: Prefect docs If you want, I can also show you the fail-fast pattern vs the collect all failures first pattern for these subflows.
m
@Marvin they are all independant processes
m
✅ thought for 9.4 seconds
That’s fine — even if they’re independent processes, the parent flow can still fail based on their final states. The key point is: - 
submit_to_work_pool(...)
starts separate flow runs - those runs execute independently - the returned 
future
still tracks each run - 
future.result()
will raise in the parent if that child flow run finished in a failed state So for independent subprocess/subflow runs, this is still the right pattern:
Copy code
for storage_key in storage_keys:
    <http://logger.info|logger.info>(
        f"Starting global pipeline for storage_key: {storage_key} and date: {partition_date}"
    )
    futures.append(
        global_pipeline.submit_to_work_pool(
            prefect_env=prefect_env,
            tasks=tasks if tasks else ["global_tasks"],
            storage_key=storage_key,
            partition_date=partition_date,
            other_args=other_args,
        )
    )

for storage_key, future in zip(storage_keys, futures):
    try:
        future.result()
    except Exception as exc:
        logger.error(f"Pipeline failed for storage_key={storage_key}: {exc}")
        raise
A few important nuances: If you only use 
wait()
 - the parent waits until each child run finishes - but it will not fail automatically if one child failed If you use 
result()
 - the parent waits for the child run to finish if needed - then raises if that run failed - which causes the parent flow to fail If your goal is: Run them all in parallel, then fail parent if any failed this is a good pattern:
Copy code
futures = []
keys = []

for storage_key in storage_keys:
    keys.append(storage_key)
    futures.append(
        global_pipeline.submit_to_work_pool(
            prefect_env=prefect_env,
            tasks=tasks if tasks else ["global_tasks"],
            storage_key=storage_key,
            partition_date=partition_date,
            other_args=other_args,
        )
    )

errors = []

for storage_key, future in zip(keys, futures):
    try:
        future.result()
    except Exception as exc:
        logger.error(f"Pipeline failed for storage_key={storage_key}: {exc}")
        errors.append((storage_key, exc))

if errors:
    raise RuntimeError(
        "One or more subflows failed: "
        + ", ".join(storage_key for storage_key, _ in errors)
    )
That gives you: - full parallel execution - all runs get a chance to finish - parent fails at the end if any child failed If instead you want: fail parent immediately when the first child failure is observed, use:
Copy code
for storage_key, future in zip(storage_keys, futures):
    future.result()
Since you mentioned they’re independent processes, I suspect the collect all failures, then raise once version is probably the better fit.
Open in Slack
PreviousNext
Powered by