I'm trying to setup a programatic notification on slack when my flows fail. I'm looking at the docs for SlackWebhook and they say to use it like so

from prefect.blocks.notifications import SlackWebhook

slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
slack_webhook_block.notify("Hello from Prefect!")

this is working for me. I wrote a function to do that as per this thread and added the

on_failure=[notify_slack]

but i'm getting the error (and mypy is showing the same)

packages/flows/notify.py", line 7, in notify_slack
    slack_webhook_block.notify(
    ^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'coroutine' object has no attribute 'notify'

I'm curious why this is working when I just run this code directly, but within prefect flow infrastructure it's failing. I'm seeing that in pylance the variable is of type

slack_webhook_block: SlackWebhook | Coroutine[Any, Any, SlackWebhook]

-- so perhaps because my flow is async, this

notify_slack

function needs to be async as well?

Yea this is a very frustrating piece of tech debt from 2.x lurking in the codebase: essentially there is engine magic that attempts to detect whether there is an available event loop or not, and based on that return a coroutine or make a synchronous call. There's an escape hatch kwarg called

_sync=True/False

that you can use to hardcode deterministic behavior on methods such as this. More info here: https://github.com/PrefectHQ/prefect/issues/15008 It makes type-hinting challenging to say the least!

oh interesting, thanks Chris! so am I "safe" to take this route? Or is there a better practice recommendation? Only asking because I couldn't safely tell from the docs

from prefect import Flow
from prefect.blocks.notifications import SlackWebhook
from prefect.settings import PREFECT_UI_URL
from prefect.server.schemas.core import FlowRun
from prefect.server.schemas.states import State

async def notify_slack(flow: Flow, flow_run: FlowRun, state: State):
    slack_webhook_block = await SlackWebhook.aload("my-slack-webhook")
    await slack_webhook_block.notify(  # type: ignore
        (
            f"Your job {flow_run.name} entered {state.name} "
            f"with message:\n\n"
            f"See <{PREFECT_UI_URL.value()}/flow-runs/"
            f"flow-run/{flow_run.id}|the flow run in the UI>\n\n"
            f"Tags: {flow_run.tags}\n\n"
            f"Scheduled start: {flow_run.expected_start_time}"
        )
    )

yup that's exactly right! At some point we'll introduce

anotify

as well but until then this is the right way to proceed

Thanks Chris -- feel free to steal this if you'd like, or I can open another PR to your docs 🙂 https://gist.github.com/Ben-Epstein/782ba98dcae837ab633cfb0e5aa9eecf

i meannnnnnnn we do love docs contributions 😉 but no pressure, I'll set a reminder for this and if you haven't gotten around to it I can include it. Thank you!

Happy to do it! where would you like to add this? Maybe under this page https://docs.prefect.io/v3/tutorials/alerts#send-alerts-on-failure in the

Platform Engineering

section. LMK if you like that and ill add it

Yup that seems like the best place to me 👍

feel free to edit as needed 🙂 https://github.com/PrefectHQ/prefect/pull/17589