Josh
04/11/2022, 4:22 PMException raised while calling state handlers: KeyError('The secret SLACK_PREFECT_NOTIFICATIONS_WEBHOOK_URL was not found. Please ensure that it was set correctly in your tenant: <https://docs.prefect.io/orchestration/concepts/secrets.html>')
Kevin Kho
04/11/2022, 4:25 PMSecret().get()
?Anna Geller
04/11/2022, 5:58 PMexport PREFECT__CLOUD__USE_LOCAL_SECRETS=false
btw are you sure this is how you named your webhook? the default Slack secret name is SLACK_WEBHOOK_URLSlackTask
, you would need to make it explicitJosh
04/11/2022, 6:08 PMstate_handlers.append(
slack_notifier(
only_states=[Failed],
webhook_secret="SLACK_PREFECT_NOTIFICATIONS_WEBHOOK_URL",
)
)
Anna Geller
04/11/2022, 6:13 PMSlackTask
is better in a state handlerimport prefect
from prefect import task, Flow
from prefect.tasks.notifications import SlackTask
from typing import cast
def post_to_slack_on_failure(task, old_state, new_state):
if new_state.is_failed():
if isinstance(new_state.result, Exception):
value = "```{}```".format(repr(new_state.result))
else:
value = cast(str, new_state.message)
msg = (
f"The task `{prefect.context.task_name}` failed "
f"in a flow run {prefect.context.flow_run_id} "
f"with an exception {value}"
)
SlackTask(webhook_secret="SLACK_PREFECT_NOTIFICATIONS_WEBHOOK_URL", message=msg).run()
return new_state
@task(state_handlers=[post_to_slack_on_failure])
def divide_numbers(a, b):
return 1 / (b - a)
with Flow(name="state-inspection-handler") as flow:
result = divide_numbers(1, 1)
if __name__ == "__main__":
flow.run()