<@ULVA73B9P> I have a task that my throw an exception, but I want to continue the other tasks anyway...

Adrien Besnard

06/22/2026, 3:56 PM

@Marvin I have a task that my throw an exception, but I want to continue the other tasks anyway. Is there an idomatic way to do that?

Marvin

06/22/2026, 3:56 PM

✅ thought for 107.0 seconds

Marvin

06/22/2026, 3:58 PM

Yes — in Prefect 3.x, the idiomatic approach is usually: submit tasks concurrently, then inspect each future/state individually instead of immediately calling
.result()
in a way that raises and stops your flow logic. A few good patterns: 1. Submit tasks and handle failures per future

Copy code

python
from prefect import flow, task

@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(5)]

    results = []
    for fut in futures:
        try:
            results.append(fut.result())
        except Exception as exc:
            print(f"Task failed, continuing anyway: {exc}")

    return results

This lets all submitted tasks run, even if one fails. 2. Use
wait()
and inspect states This is especially nice for batches/mapped tasks.

Copy code

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

@task
def maybe_fail(x):
    if x % 2 == 0:
        raise ValueError(f"bad value: {x}")
    return x * 10

@flow
def my_flow():
    futures = [maybe_fail.submit(i) for i in range(5)]
    done, not_done = wait(futures)

    successes = []
    failures = []

    for fut in done:
        if fut.state.is_completed():
            successes.append(fut.result())
        else:
            failures.append(fut.state)

    print(f"successes={successes}")
    print(f"failures={len(failures)}")

3. Use
.result(raise_on_failure=False)
If you want to avoid try/except:

Copy code

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

    for fut in futures:
        value = fut.result(raise_on_failure=False)
        print(value)

For failed tasks, that returns the exception/result payload instead of raising. 4. Catch expected errors inside the task If failure is expected business logic, this is often the cleanest design:

Copy code

python
@task
def maybe_fail(x):
    try:
        if x == 2:
            raise ValueError("boom")
        return {"ok": True, "value": x * 10}
    except ValueError as exc:
        return {"ok": False, "error": str(exc)}

That keeps the flow itself simple and avoids failed task states for expected cases. Important note If you call a task directly like

maybe_fail(2)

inside a flow, that exception is raised inline and can interrupt downstream flow code unless you catch it. If you want independent execution, prefer

.submit()

.map()

and then inspect the futures. So the short answer is: Yes: use `.submit()`/`.map()` and then either
try/except
around each
future.result()
, or inspect futures/states with
wait()
. If you want, I can show the most idiomatic version for: - independent parallel tasks - mapped tasks - downstream tasks that should still run when upstream fails

Adrien Besnard

06/22/2026, 4:00 PM

I like the 3., actually. Will the flow fail anyway? I want it to fail

Adrien Besnard

06/22/2026, 4:00 PM

I like the 3., actually. Will the flow fail anyway? I want it to fail @Marvin

Marvin

06/22/2026, 4:00 PM

✅ thought for 6.8 seconds

Marvin

06/22/2026, 4:00 PM

Yes — the flow can still fail even if you use

future.result(raise_on_failure=False)

. That flag only changes whether your Python code raises immediately when reading the future. It does not magically mark the task as successful. So you can do both: - let all tasks run - inspect all results without stopping early - then explicitly fail the flow at the end if any task failed Example:

Copy code

python
from prefect import flow, task

@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(5)]

    results = []
    failures = []

    for fut in futures:
        value = fut.result(raise_on_failure=False)
        if fut.state.is_failed():
            failures.append(value)
        else:
            results.append(value)

    if failures:
        raise RuntimeError(f"{len(failures)} task(s) failed: {failures}")

    return results

In this pattern: - all tasks are allowed to finish - failed tasks do not interrupt collection - the flow still fails at the end if any failed If you want a cleaner version, you can inspect states directly:

Copy code

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(5)]
    wait(futures)

    failed = [fut for fut in futures if fut.state.is_failed()]
    results = [fut.result() for fut in futures if fut.state.is_completed()]

    if failed:
        raise RuntimeError(f"{len(failed)} tasks failed")

    return results

So the answer is: No,
raise_on_failure=False
does not prevent the flow from failing — but only if you choose to fail it yourself afterward. If you don’t raise at the end, the flow may complete successfully even though some tasks failed. Docs: - Tasks - Flows If you want, I can give you a small helper like

gather_allowing_failures()

to make this pattern reusable.

Adrien Besnard

06/22/2026, 6:48 PM

@Marvin How to make a task fail without throwing an exception?

Marvin

06/22/2026, 6:48 PM

✅ thought for 145.4 seconds

Marvin

06/22/2026, 6:51 PM

Yes — in Prefect 3.x, the way to do that is to return a failed state from the task.

Copy code

python
from prefect import task, flow
from prefect.states import Failed

@task
def my_task(x):
    if x < 0:
        return Failed(message=f"Invalid value: {x}")
    return x * 10

@flow
def my_flow():
    state = my_task(-1, return_state=True)
    print(state.type)  # FAILED

A few important notes: 1. Returning
Failed(...)
marks the task as failed You do not need to raise an exception in your task code. 2. If you want to inspect it without raising, use
return_state=True

Copy code

python
@flow
def my_flow():
    state = my_task(-1, return_state=True)

    if state.is_failed():
        print(f"Task failed: {state.message}")
        raise RuntimeError("Failing flow because task failed")

3. If you call
.result()
on that failed state/future, it can still raise So “fail without throwing an exception” really means: - no uncaught exception inside the task body - but the task still ends in a FAILED state - and downstream/result access may still raise unless you handle state explicitly If your goal is: - task should fail - other tasks should continue - flow should fail at the end then this pattern works well:

Copy code

python
from prefect import flow, task
from prefect.states import Failed

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

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

    failed = []
    results = []

    for fut in futures:
        state = fut.wait()
        if state.is_failed():
            failed.append(state.message)
        else:
            results.append(fut.result())

    if failed:
        raise RuntimeError(f"Some tasks failed: {failed}")

    return results

Relevant docs: - Tasks - Flows If you want, I can show the cleanest batch pattern for “collect all task outcomes, then fail the flow once at the end.”

Open in Slack

Previous Next

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.