Hey, been using prefect for a while and am a big fan. Is it possible to get the error of an upstream task? I have a task that triggers on any_failed and it receives

prefect.engine.signals.TRIGGERFAIL('Trigger was "all_successful" but some of the upstream tasks failed.')

I would like to get the actual error message of the upstream task, in this case

<Failed: "Unexpected error: NameError("name 'asd' is not defined")">

@task(trigger=prefect.triggers.any_failed)
def any_failed(data):
    logger = prefect.context.get('logger')
    print(data)
    <http://logger.info|logger.info>("Failed! Error message: %s", data)

@task()
def fail():
    asd
    return None

@task
def process(data):
    return True

with Flow('test') as f:
    data = fail()
    result = process(data)
    any_failed(result)
    f.set_reference_tasks([result])

state = f.run()

And I would like to get the upstream error message in any_failed. I was also looking into state_handlers, but I would prefer this solution having a single point to catch any errors.

When an exception is thrown in a task, it should be ending up in a

Failed

state with the exception as the

result

. 😕 As an aside, you can add a state_handler at the flow level instead of the task level, which may do what you want.

Hi @Spencer, thanks for the reply. In my case, it is nested. So a task fails, this is then used in another task that transisitions to the state trigger failed. And this is then passed to the task that triggers on

any_failed

. As in my example, data will have

TriggerFailed

and it is not possible for me to get back the original error message of the

fail

task

Ah, I see the nesting now; makes sense. Not sure if it's possible to get transitive failures like that without threading it explicitly.