Isabel
11/11/2024, 3:44 PMMarvin
11/11/2024, 3:44 PMIsabel
11/11/2024, 3:45 PMMarvin
11/11/2024, 3:46 PMpython
from prefect import flow
from prefect.blocks.core import Block
from prefect.settings import PREFECT_API_URL
def notify_slack(flow, flow_run, state):
slack_webhook_block = Block.load("slack-webhook/my-slack-webhook")
slack_webhook_block.notify(
(
f"Your job {flow_run.name} entered {state.name} "
f"with message:\n\n"
f"See <https://{PREFECT_API_URL.value()}/flow-runs/"
f"flow-run/{flow_run.id}|the details>"
)
)
In this example, the notify_slack
function is triggered when a flow run fails. You can adjust the notification to any other system instead of Slack by changing the logic in notify()
.
3. Triggering by Flow Name: The flow_run.name
in the code is where you can match or log the flow's name for your conditional logic, ensuring you act based on the right flow.
For more details, you can refer to the Prefect documentation on state change hooks.
And don't worry, even the most efficient robots occasionally mix up their circuits! If you need any more help (or jokes), feel free to ask.Isabel
11/11/2024, 3:50 PM{
"type": "event",
"match": {
"prefect.resource.id": "prefect.flow-run.*"
},
"match_related": {
"prefect.resource.role": "flow",
"prefect.resource.id": [
"prefect.flow.a5b76d9d-9d8d-4b45-ba7d-1275d2d19953"
]
},
"after": [],
"expect": [
"prefect.flow-run.Failed"
],
"for_each": [
"prefect.resource.id"
],
"posture": "Reactive",
"threshold": 1,
"within": 0
}
Marvin
11/11/2024, 3:50 PM