I’m getting intermittent errors about a Prefect Se...
# prefect-community
j
I’m getting intermittent errors about a Prefect Secret not being set in the cloud for my tenant. It happens every once in a while, and when I re-run the flow, it succeeds. Any idea what is causing the problem and what I can do to make sure it stops being flaky?
Copy code
Exception 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>')
k
I think this is cloud. Is the error intermittent? or always happens?
Are you using
Secret().get()
?
a
if you run it locally, you can do:
Copy code
export 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_URL
if you use a different name than this default Secret name in a
SlackTask
, you would need to make it explicit
j
yeah we named it that because we were already using the default one. In flow initialization, we have
Copy code
state_handlers.append(
    slack_notifier(
        only_states=[Failed],
            webhook_secret="SLACK_PREFECT_NOTIFICATIONS_WEBHOOK_URL",
    )
)
a
ideally, don't use slack_notifier 🙂
SlackTask
is better in a state handler
try this:
Copy code
import 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()