Josh

    Josh

    5 months ago
    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?
    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>')
    Kevin Kho

    Kevin Kho

    5 months ago
    I think this is cloud. Is the error intermittent? or always happens?
    Are you using
    Secret().get()
    ?
    Anna Geller

    Anna Geller

    5 months ago
    if you run it locally, you can do:
    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
    Josh

    Josh

    5 months ago
    yeah we named it that because we were already using the default one. In flow initialization, we have
    state_handlers.append(
        slack_notifier(
            only_states=[Failed],
                webhook_secret="SLACK_PREFECT_NOTIFICATIONS_WEBHOOK_URL",
        )
    )
    Anna Geller

    Anna Geller

    5 months ago
    ideally, don't use slack_notifier 🙂
    SlackTask
    is better in a state handler
    try this:
    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()