Chris Gunderson
08/26/2022, 9:36 PMfrom 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:
@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?Matt Conger
08/26/2022, 11:38 PM