https://prefect.io logo
#prefect-community
Title
# prefect-community
c

Chris Gunderson

08/26/2022, 9:36 PM
Hi Team - I'd like to add this function in a utility script
Copy code
from prefect.tasks.notifications import SlackTask
class Alert:

    def alert_on_special_failure(task, old_state, new_state):
        if new_state.is_failed():
            if getattr(new_state.result, "flag", False) is True:
                errMsg = '--- LOADER ERROR ---'
                msg = "{}\nTask: `{}` FAILED.\nThe loader process failed: `{}`".format(
                        errMsg, task.name, new_state.result.value)
                SlackTask().run(message = msg, webhook_secret = "SLACK_PREFECT_DEV")
        return new_state
This was previous called like this:
Copy code
@task(name = 'send API request',
      max_retries = 3,
      retry_delay = timedelta(minutes = 5),
      state_handlers = [alert_on_special_failure])
def post_request_process(
If we add this alert on special failure object, will we need to pass in the task, old_state, and new_state? Were these variables global?
👀 1
1
I think this might work. We are seeing what we will need to do to move to Perfect 2.0 https://docs.prefect.io/concepts/states/ @flow def my_flow(): state = add_one(1, return_state=True) # return State result = state.result() # return int
m

Matt Conger

08/26/2022, 11:38 PM
Sweet, glad you got it figured out Chris. If you are switching over to 2.0 definitely check out the slack block and new slack collections integration. https://prefecthq.github.io/prefect-slack/ https://docs.prefect.io/api-ref/prefect/blocks/notifications/
👍 2