Hello, I am trying to add slack notifications for a prefect flow using this: https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#installation-instructions I have a slack webhook URL that I've saved as a prefect secret. I am importing this:

from prefect.utilities.notifications import slack_notifier

and my tasks now look like this :

task(log_stdout=True, state_handlers=[slack_notifier])

install_snowflake_task = ShellTask(helper_script="pip install boto3 \
    snowflake-connector-python[pandas] \
    snowflake-ingest && pip install PyJWT==1.7.1", shell="bash", stream_output=True, return_all=True, state_handlers=[slack_notifier])

install_dbt_task = ShellTask(helper_script="pip install dbt==0.18.0", \
                             shell="bash", stream_output=True, return_all=True, state_handlers=[slack_notifier])

My flow looks like

with Flow("name-of-flow", state_handlers=[slack_notifier]) as flow:

The flow runs successfully but I see no notifications on the slack channel that I have authorized for this. What am I doing wrong?

This page is quite old tbh 😄 it's much easier to create a Slack webhook, store it as Secret in your Prefect Cloud account and use it in the

SlackTask

Would love an example!

yup, let me write it first 😄

Thank you!

Run this first:

export PREFECT__CLOUD__USE_LOCAL_SECRETS=false

This should be in the prefect flow.py file and should be run each time?

No, you need to set it only once and this will be stored in your ~/.prefect/config.toml

So I am running prefect on a k8 cluster and I'm using sealedsecrets for prefect secrets. Wouldn't this make it a local secret?

if you need to set it on your agent:

prefect agent kubernetes start --env PREFECT__CLOUD__USE_LOCAL_SECRETS=false

@Anna Geller Thank you for your help yesterday. I have Prefect now posting messages to slack. I ended up having to initialize the SlackTask like this:

slack = SlackTask()

and passed in a

webhook_url

instead of secret in the flow like this:

slack(message="Hi from a flow!", webhook_url=SLACK_WEBHOOK_URL_PREFECT)

. This works and I see the message on the chosen channel. However, I still get a failure from SlackTask at some point in the flow with this message:

Task 'SlackTask': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 863, in get_task_run_state
    logger=self.logger,
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 454, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/tasks.py", line 445, in method
    return run_method(self, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/tasks/notifications/slack_task.py", line 61, in run
    webhook_url = webhook_url or cast(str, Secret(webhook_secret).get())
  File "/usr/local/lib/python3.7/site-packages/prefect/client/secrets.py", line 142, in get
    ) from None
ValueError: Local Secret "SLACK_WEBHOOK_URL" was not found.

Also, the state handler function you shared fails too:

Exception raised while calling state handlers: AttributeError("'Context' object has no attribute 'task_name'")
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/cloud/flow_runner.py", line 120, in call_runner_target_handlers
    old_state=old_state, new_state=new_state
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 116, in call_runner_target_handlers
    new_state = handler(self.flow, old_state, new_state) or new_state
  File "<string>", line 25, in post_to_slack_on_failure
AttributeError: 'Context' object has no attribute 'task_name'

Are there some imports I am missing? Maybe something for old_state, new_state?

SlackTask does not take in

webhook_url

. It takes in a

webhook_secret

that is a string that is used to fetch the Secret with that name from Prefect Cloud. Because you used

webhook_url

, it looked for the default secret name which is

SLACK_WEBHOOK_URL

can you share your flow and how do you call your flow? do you run your flow just locally?

@Lon Nix just confirmed that we're on Prefect Server 😬

upsss

I am already doing that, this is my flow:

Kevin already told you that you should set

webhook_secret

rather than

webhook_url

here:

SlackTask(message=msg, webhook_url=SLACK_WEBHOOK_URL_PREFECT).run()

also, the state handler will fire only when your task will fail. The way you set this up at the moment it's super unlikely to fail because it does nothing other than printing something to the console 😄

@task(log_stdout=True, state_handlers=[post_to_slack_on_failure])
def print_output(output):
    logger = prefect.context.get("logger")
    for l in output:
        <http://logger.info|logger.info>(l)

Thanks @Anna Geller. We found we were doing things one way in one file and another way in another file so we're adjusting from our side to make sure all the changes you've suggested got implemented properly across the board. Hopefully that takes care of our issues and we'll force a failure somewhere to test that state handler

Yeah, thanks @Anna Geller 🙂