hey <@ULVA73B9P> Can you write me me a on_failure ...
# ask-marvin
f
hey @Marvin Can you write me me a on_failure hook function that prints the error message if there’s an exception ?
m
thought for 154.8 seconds
Here’s a simple Prefect 3.x on_failure hook that prints the exception message (and falls back to the state message if needed):
Copy code
from prefect.states import get_state_exception

def print_failure_exception(obj, run, state):
    # obj is a Flow or Task object, run is FlowRun or TaskRun, state is the final State
    try:
        exc = get_state_exception(state)
    except Exception:
        exc = None

    if exc:
        print(f"[{run.name}] failed with error: {exc}")
        # If you want a full traceback instead, uncomment the next two lines:
        # from prefect.states import format_exception
        # print(format_exception(exc))
    else:
        # Fallback: print any state message if available
        msg = getattr(state, "message", None)
        print(f"[{run.name}] failed. {('Message: ' + msg) if msg else ''}")
Example usage on a flow:
Copy code
from prefect import flow

@flow(on_failure=[print_failure_exception], log_prints=True)
def my_flow():
    1 / 0  # trigger a failure

if __name__ == "__main__":
    my_flow()
You can also attach it to a task similarly with
@task(on_failure=[print_failure_exception])
. Docs for reference: State change hooks in Prefect 3