@Marvin for an

on_failure

hook in a prefect 2 flow, how do I retrieve the error message the flow failed with

To retrieve the error message when your Prefect 2.x flow fails using an

on_failure

hook, you can capture the state information during the state change. State change hooks are designed to execute code when there are changes in flow or task run states. Here's a basic example from the Prefect 2.x documentation that illustrates capturing and notifying when a flow fails:

python
from prefect import flow
from prefect.blocks.core import Block
from prefect.settings import PREFECT_API_URL

def notify_slack(flow, flow_run, state):
    slack_webhook_block = Block.load("slack-webhook/my-slack-webhook")
    slack_webhook_block.notify(
        (
            f"Your job {flow_run.name} entered {state.name} "
            f"with message:\n\n"
            f"See <https://{PREFECT_API_URL.value()}/flow-runs/"
            f"flow-run/{flow_run.id}|the flow run in the UI>\n\n"
            f"Tags: {flow_run.tags}\n\n"
            f"Scheduled start: {flow_run.expected_start_time}"
        )
    )

@flow(on_failure=[notify_slack], retries=1)
def failing_flow():
    raise ValueError("oops!")

if __name__ == "__main__":
    failing_flow()

In this example, the

notify_slack

function acts as an

on_failure

hook, which sends a notification when the flow fails. The error message "oops!" from

raise ValueError("oops!")

would be included in the state object, accessible within your hook function as part of the

state

parameter. You can extract specific details from the state like the error message or trace information to include in your notifications. For more details, you can check their state change hooks guide here. Hope this helps, and remember: sometimes, flows fail just to keep you busy!

@Marvin there's nothing in the sample message about getting the exception. how can I access the exception from the state in prefect 2?

To access the exception from a flow state in Prefect 2.x, you can use the

result()

method on the state object, which will raise the exception. Here's the approach: 1. During the state transition, the exception details are a part of the state object. 2. You can access the exception by calling

state.result()

, which will raise the exception if the state is failed. However, this requires handling exceptions explicitly. Here's an example based on the documentation:

python
from prefect import flow, task

@task
def my_task():
    raise ValueError("Oops, something went wrong!")

@flow
def my_flow():
    try:
        state = my_task(return_state=True)
        result = state.result()
    except ValueError as e:
        error_message = str(e)
        print(f"Oh no! The task failed with error: {error_message}")

my_flow()

In an

on_failure

hook, you can similarly access the exception from

state.result()

within the hook, catching it using try/except to extract the error message. Hopefully, this helps! Even the best of flows encounter turbulence now and then. Check out the full helpful examples from the documentation on handling states.