Hi all! I am wondering what is the best practice, ...
# best-practices
d
Hi all! I am wondering what is the best practice, in prefect 2.0 orion, to trigger a flow run if a run of another flow fails. My concrete use case is the following: I would like to send a slack message as an alert in case a periodic flow (let's call it _flow_A_) fails. I have already set up _flow_B_ that sends a slack message using prefect-slack collection. I just need to trigger it with some input if a run of _flow_A_ fails
a
I wanted to tell you to check Prefect 1.0 Automations, but given you mentioned the
prefect-slack
collection, I assume you rather ask for Prefect 2.0?
👏 1
🙂 1
d
yes sorry, I am using prefect 2.0 . I've just updated the message.
a
thanks, I think this should work:
Copy code
from prefect import task, flow
from prefect.futures import PrefectFuture
from prefect_slack import SlackWebhook
from prefect_slack.messages import send_incoming_webhook_message


def send_slack_alert_on_failure(task_future: PrefectFuture):
    task_future.wait()  # block until completion
    if task_future.get_state().is_failed():
        name_ = task_future.task_run.name
        id_ = task_future.task_run.flow_run_id
        send_incoming_webhook_message(
            slack_webhook=SlackWebhook("<https://hooks.slack.com/services/xx/xx/xxx>"),
            text=f"The task `{name_}` failed in a flow run `{id_}`",
        )


@task
def fail_successfully(x):
    return 1 / x


@flow
def main_flow(nr: int):
    future_obj = fail_successfully(nr)
    send_slack_alert_on_failure(future_obj)


if __name__ == "__main__":
    main_flow(nr=0)
We plan to release a first-class feature allowing you to send failure alerts on failed runs automatically without having to write custom code as here. To give you more transparency here, the target ETA for this is the end of July afaik
yup, this works
d
amazing! many thanks @Anna Geller