Yes — in Prefect 3.x you can do this with

allow_failure

.

allow_failure

lets downstream tasks run even if an upstream task failed. The failed task will still show as Failed in the UI, but the flow won’t fail. The downstream task receives the actual Exception object as its input, so you can handle it there without a try/except around the upstream task. Example:

from prefect import flow, task, allow_failure

@task
def might_fail():
    raise ValueError("Boom!")

@task
def handle_result(x):
    if isinstance(x, Exception):
        # upstream failed; decide how to proceed
        return f"handled error: {x}"
    return f"value: {x}"

@flow
def my_flow():
    fut = might_fail.submit()
    # Downstream runs even though upstream failed
    handled = handle_result.submit(allow_failure(fut))
    return handled.result()

Behavior: - The

might_fail

task will be marked Failed in the UI. -

handle_result

runs and is marked Completed. - The flow completes successfully. - The downstream receives the Exception instance; check with

isinstance(x, Exception)

. Notes: - You can wrap lists/mapped futures too:

downstream.map(allow_failure(upstream.map(items)))

. - You can also pass

allow_failure(state)

if you’re working with states (e.g., from

return_state=True

). - Avoid calling

.result()

on the failed upstream future directly; instead pass it via

allow_failure

into a task that decides how to handle it.