Leon Kozlowski
09/16/2024, 6:55 PM"""Incoming Webhook Block for #prefect-alerts Slack Channel."""
import os
from prefect.blocks.notifications import SlackWebhook
class PrefectAlerts(SlackWebhook):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._block_document_name = "prefect-alerts-incoming-webhook"
url = os.getenv("PREFECT_INCOMING_WEBHOOK__PREFECT_ALERTS")
is this how i would give it a name?Nate
09/16/2024, 7:12 PMLeon Kozlowski
09/16/2024, 7:14 PMLeon Kozlowski
09/16/2024, 7:15 PM"""Incoming Webhook Block for #prefect-alerts Slack Channel."""
import os
from prefect_slack import SlackWebhook
if __name__ == "__main__":
SlackWebhook(
url=os.getenv("PREFECT_INCOMING_WEBHOOK__PREFECT_ALERTS")
).save(name="incoming-webhook-prefect-alerts")
block^Leon Kozlowski
09/16/2024, 7:15 PMfrom prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect_slack import SlackWebhook
if __name__ == "__main__":
webhook = SlackWebhook.load(name="incoming-webhook-prefect-alerts")
automation = Automation(
name="failure-prefect-alerts",
trigger=EventTrigger(
expect={
"prefect.flow-run.Failed",
},
match={
"prefect.resource.id": "prefect.flow-run.*"
},
posture="Reactive",
for_each={
"prefect.resource.id"
},
threshold=1,
within=0,
),
actions=[webhook]
).create()
automation^Leon Kozlowski
09/16/2024, 7:15 PMLeon Kozlowski
09/16/2024, 7:15 PMNate
09/16/2024, 7:20 PMactions
needs to take a known / defined action, like RunDeployment
/ PauseDeployment
or SendNotification
Leon Kozlowski
09/16/2024, 7:20 PMLeon Kozlowski
09/16/2024, 7:20 PMNate
09/16/2024, 7:22 PM