Matt Kizaric
05/04/2023, 3:32 PMBianca Hoch
05/04/2023, 4:47 PMFailed
or Canceled
.Matt Kizaric
05/04/2023, 4:49 PMBianca Hoch
05/04/2023, 5:30 PMimport prefect
from 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)
#pass
#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.COMPLETED):
notify_function()
set_flow_run_state
import asyncio
from prefect.client import get_client
from prefect.orion.schemas.states import Cancelled
async def main():
async with get_client() as client:
id = "2331e33e-13aa-4415-a6e5-e6ea86c426a9"
response = await client.set_flow_run_state(flow_run_id=id, state=Cancelled())
print(response)
if __name__ == "__main__":
asyncio.run(main())
Serina
05/04/2023, 8:59 PMon_failure
(and on_crashed
) state change hook that allows you to execute arbitrary Python code in response to the above terminal states. Example usage can be found here:
https://github.com/PrefectHQ/prefect/blob/main/RELEASE-NOTES.md#on_completion-and-on_failure-hooks-for-flows-and-taskson_cancelled
flow run state change hook soon, and its progress can be tracked here:
https://github.com/PrefectHQ/prefect/issues/8446