Hi, I have some questions about using the on_failure
hook that was recently released: prefect/RELEASE-NOTES.md at main · PrefectHQ/prefect · GitHub. Apologies for the upcoming lengthy post:
1. How can we pass task parameters to the hook?
I have tried it with this sample code:
from prefect import flow, task
from pprint import pprint
def task_hook(task, task_run, state):
# pprint(task.__dict__, indent=2)
pprint(task_run.__dict__, indent=2)
# pprint(state.__dict__, indent=2)
@task(on_failure=[task_hook])
def failing_task(message):
raise ValueError(message)
@flow
def failing_flow(message):
failing_task(message)
failing_flow("foobars")
In this case, I would like to retrieve the message
param from within task_hook
. After going through the output of task
, task_run
, and state
, it looks like task_run
has what I need:
{
...,
'task_inputs': {'message': []},
...
}
But it is empty. Is there a way to retrieve the actual message
that was passed to the task? It seems possible with flows - if I were to use a flow hook instead, flow_run
will look like this:
{
...,
'parameters': {'message': 'foobars'},
...
}
This leads to a second related question:
2. Is it possible to retrieve the value of a Secret Block in a hook?
With some updated code that passes a Secret instead of a plain string as a param:
from prefect import flow, task
from prefect.blocks.system import Secret
def flow_hook(flow, flow_run, state):
print(flow_run.parameters)
@task
def failing_task(message: Secret):
raise ValueError(message)
@flow(on_failure=[flow_hook])
def failing_flow(message: Secret):
failing_task(message)
failing_flow(Secret(value="foobars"))
The output of flow_run.parameters
in flow_hook
looks like this:
{'message': {'value': '**********', 'block_type_slug': 'secret'}}
Normally, we’d use .get()
to retrieve the value of a Secret block, but in this case, how can we get the unobfuscated value?
Last question:
3. How can we retrieve more of the stack trace from within the hook?
It looks like the error message itself is accessible, by printing state.message
, which gives:
Task run encountered an exception: ValueError: foobars
But can we also get part of the stack trace as well? The Prefect error logs are useful to pinpoint the source of the error since they have file names and line numbers:
...
File "/FILE_PATH/test.py", line 44, in failing_task
raise ValueError(message)
ValueError: foobars
But it does not seem to be present in the data that is sent to the hook.
Thanks for your help!