https://prefect.io logo
Title
h

Hawkar Mahmod

08/19/2022, 2:28 PM
What is the idiomatically Prefect (2.0) way of handling failed tasks? My task raises an exception that bubbles all the way when I run my flow locally. I’d like to just be able to have the flow finish with a failed state and not see the whole stack trace?
a

Anna Geller

08/19/2022, 2:34 PM
That's totally possible! You can raise a custom exception directly in the flow, which will result in a
Failed
task run:
from prefect import task

@task
def signal_task(message):
    if message == 'stop_immediately!':
        raise RuntimeError(message='End the task run!')
Alternatively, your flow may return a specific state:
from prefect import task
from prefect.orion.schemas.states import Failed

@task
def signal_task(message):
    if message == 'stop_immediately!':
        return Failed(message='Stopping the task run immediately!')
🤩 1