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
Anna Geller
05/29/2022, 10:56 AM
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
Domenico Di Gangi
05/29/2022, 10:57 AM
yes sorry, I am using prefect 2.0 . I've just updated the message.
a
Anna Geller
05/29/2022, 11:02 AM
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)
Anna Geller
05/29/2022, 11:05 AM
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
Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.