Hey 🙂 I'm trying to send a notification on Slack in case of task failure, but unfortunately I don't receive anything 😕 • First, I followed the instructions on : https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#customizing-your-alerts with

slack_notifier

, without success. • Then I tried with

SlackTask

, this is my piece of code (didn't work) :

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 = 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(
            message=msg,
            webhook_secret="<https://hooks.slack.com/services/*******/*******/*******>"
        ).run()
    return new_state

• Finnally, I tried with a custom solution (didn't work, again) :

def post_to_slack_on_failure(task, old_state, new_state):
    if new_state.is_finished():
        msg = "Task {0} finished in state {1}".format(task, new_state)
        # replace with your Slack webhook URL secret name
        secret_slack = cast(str, Secret("<https://hooks.slack.com/services/*******/*******/*******>").get())
        <http://requests.post|requests.post>(secret_slack, json={"text": msg})
    return new_state

This is my test flow :

@task
def task_error():
    raise Exception("Test")

with Flow("Test Slack", state_handlers=[post_to_slack_on_failure]) as flow:
        task_error()

    flow.run()

(I also tried EmailTask with smtp_type="INSECURE", and once again it didn't work) Need some help please 🙏

@Jean-Baptiste Six you are very close to make it work! Instead of the actual web-hook string, the

webhook_secret

expects a reference to your Prefect Secret. By default, it looks for a secret named SLACK_WEBHOOK_URL.

I tried the second solution (without the secret), but when I print the response of the request I get :

<Response [404]>

404 seems that your webhook is not valid. How did you create it? Can you try creating a new webhook, store it as Prefect Secret with name SLACK_WEBHOOK_URL in the UI, and then try this simple example?

import prefect
from prefect import task, Flow
from prefect.tasks.notifications import SlackTask
from prefect.engine import signals


def post_to_slack_on_failure(task, old_state, new_state):
    if new_state.is_failed():
        msg = (
            f"The task `{prefect.context.task_name}` failed "
            f"in a flow run {prefect.context.flow_run_id} "
            f"with a message `{new_state.result}`"
        )
        SlackTask(message=msg).run()
    return new_state


@task(state_handlers=[post_to_slack_on_failure])
def mission_critical_task(a, b):
    if a == b:
        raise signals.FAIL("a equaled b!")
    else:
        return 1 / (b - a)


with Flow(name="state-inspection-handler") as flow:
    result = mission_critical_task(1, 1)

I think the requests syntax is not entirely correct, it should be:

<http://requests.post|requests.post>(url=webhook_url, data=json.dumps({"text": "your Slack msg"}))

In the UI, secret is disable

Ah storing Secrets in the UI are for Cloud only