darshan darshan
04/13/2023, 12:02 PMSahil Rangwala
04/13/2023, 4:47 PMBianca Hoch
04/13/2023, 5:29 PMfrom prefect import task, flow
from prefect.blocks.notifications import SlackWebhook
from prefect.orion.schemas.states import StateType
import time
#Load your desired block, this example uses a slack notification block.
def notify_function():
slack_webhook_block = SlackWebhook.load("my-test-notification")
#This is the message you want to send to slack
slack_webhook_block.notify("Hello from Prefect!")
#A simple flow
@flow
def my_flow():
time.sleep(10)
raise Exception
if __name__ == "__main__":
#Store the state of your flow
state = my_flow(return_state=True)
'''This if statement checks for a specific flow state.
If the condition is satisfied, the block is invoked.
State type can be set to any available state type: <https://docs.prefect.io/concepts/states/#state-types''>'
if state.type == (StateType.FAILED):
notify_function()
darshan darshan
04/14/2023, 10:22 AM