Florian Giroud
10/24/2025, 10:02 AMMarvin
10/24/2025, 10:02 AMMarvin
10/24/2025, 10:05 AMfrom 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:
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