https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I'm using this code to set up an auto...
# ask-marvin
p
@Marvin I'm using this code to set up an automation that send notifications to my Slack when there are flow in errors. But sometime some flows can be in error for some days and I receive too much notifications... Will it be possible to raise only one notification when the flow is in error, and one when it comes back to success ?
Copy code
from prefect import get_client
from prefect.blocks.notifications import SlackWebhook
from prefect.automations import AutomationCore, EventTrigger, SendNotification
import asyncio

AUTOMATION_NAME = "Notify on Flow Failure"
trigger = EventTrigger(
    type="event",
    match={"prefect.resource.id": "prefect.flow-run.*"},
    match_related={},
    after=[],
    expect=[
        "prefect.flow-run.TimedOut",
        "prefect.flow-run.Failed",
        "prefect.flow-run.Crashed"
    ],
    for_each=["prefect.resource.id"],
    posture="Reactive",
    threshold=1,
    within=0
)

async def main():
    # Initialize client
    client = get_client()
    slack_block_name = "slack-notify"

    # Create the Slack block
    block = SlackWebhook(
        url="<https://hooks.slack.com/services/XXXXX>"
    )
    await block.save(name=slack_block_name, overwrite=True)

    # Load block to get its block_document_id
    loaded_block = await SlackWebhook.load(slack_block_name)
    block_document_id = loaded_block._block_document_id

    # Check if automation already exists
    existing = await client.read_automations()
    if any(a.name == AUTOMATION_NAME for a in existing):
        print(f"Automation '{AUTOMATION_NAME}' already exists. Skipping creation.")
        return

    # Define the automation
    automation = AutomationCore(
        name=AUTOMATION_NAME,
        description="Sends a Slack notification when any flow fails or crashes.",
        trigger=trigger,
        actions=[
            SendNotification(
                block_document_id=block_document_id,
                subject="Error",
                body=(
                "Deployment: {{  deployment.name }} observed in state `{{  flow_run.state.name }}` at {{  flow_run.state.timestamp }}.\n"
                "Flow run URL: <http://XXX:4200/runs/flow-run/{{> flow_run.id }}\n"
                "State message: {{  flow_run.state.message  }}"
                )
            )
        ]
    )

    # Register the automation
    await client.create_automation(automation)

# Run async main
asyncio.run(main())
m
✅ thought for 522.2 seconds
You’re getting spam because your automation is keyed per flow run (
for_each=["prefect.resource.id"]
), so every failing run triggers another message. To “latch” notifications so you only get: - one alert when a deployment starts failing, and - one alert when it recovers (first success after failures), create two automations and key them per deployment with 
for_each=["related:deployment:id"]
, then use 
after
to gate firing: - Failure automation: fires on the first failure after a success for that deployment. - Recovery automation: fires on the first success after a failure for that deployment. Below is a drop‑in version of your script that creates both automations. Adjust the Slack webhook URL and Prefect UI base URL as needed. ``` from prefect import get_client from prefect.blocks.notifications import SlackWebhook from prefect.automations import AutomationCore, EventTrigger, SendNotification import asyncio FAIL_AUTOMATION_NAME = "Notify on first failure per deployment" RECOVERY_AUTOMATION_NAME = "Notify on recovery per deployment" # Shared Slack block name SLACK_BLOCK_NAME = "slack-notify" PREFECT_UI_BASE = "http://XXX:4200" # change to your Prefect UI base URL # Triggers: # 1) Failure: first failure AFTER a success for the same deployment failure_trigger = EventTrigger( type="event", match={"prefect.resource.id": "prefect.flow-run.*"}, match_related={}, # optionally filter specific deployments/work pools via match_related after=["prefect.flow-run.Completed"], # gate on a prior success expect=[ "prefect.flow-run.TimedOut", "prefect.flow-run.Failed", "prefect.flow-run.Crashed", ], for_each=["relateddeploymentid"], # key state per deployment posture="Reactive", threshold=1, within=0, ) # 2) Recovery: first success AFTER a failure for the same deployment recovery_trigger = EventTrigger( type="event", match={"prefect.resource.id": "prefect.flow-run.*"}, match_related={}, after=[ "prefect.flow-run.TimedOut", "prefect.flow-run.Failed", "prefect.flow-run.Crashed", ], # gate on a prior failure expect=["prefect.flow-run.Completed"], for_each=["relateddeploymentid"], posture="Reactive", threshold=1, within=0, ) async def main(): client = get_client() # Create/update Slack block block = SlackWebhook(url="https://hooks.slack.com/services/XXXXX") await block.save(name=SLACK_BLOCK_NAME, overwrite=True) loaded_block = await SlackWebhook.load(SLACK_BLOCK_NAME) block_document_id = loaded_block._block_document_id existing = await client.read_automations() existing_by_name = {a.name: a for a in existing} # Failure automation if FAIL_AUTOMATION_NAME not in existing_by_name: fail_automation = AutomationCore( name=FAIL_AUTOMATION_NAME, description="Send a Slack notification on the first failure after a success per deployment.", trigger=failure_trigger, actions=[ SendNotification( block_document_id=block_document_id, subject="Flow failures detected", body=( "Deployment: {{ deployment.name }} started failing. " "Observed state: 
{{ flow_run.state.name }}
at {{ flow_run.state.timestamp }}.\n" f"Flow run: <{PREFECT_UI_BASE}/runs/flow-run/{{{{ flow_run.id }}}}|Open in Prefect>\n" "State message: {{ flow_run.state.message }}" ), ) ], ) await client.create_automation(fail_automation) print(f"Created automation: {FAIL_AUTOMATION_NAME}") else: print(f"Automation '{FAIL_AUTOMATION_NAME}' already exists. Skipping creation.") # Recovery automation if RECOVERY_AUTOMATION_NAME not in existing_by_name: recovery_automation = AutomationCore(
name=RECOVERY_AUTOMATION_NAME, description="Send a Slack notification on the first success after failures per deployment.", trigger=recovery_trigger, actions=[ SendNotification( block_document_id=block_document_id, subject="Flow recovered", body=( "Deployment: {{ deployment.name }} recovered. " "Observed state: 
{{ flow_run.state.name }}
at {{ flow_run.state.timestamp }}.\n" f"Flow run: <{PREFECT_UI_BASE}/runs/flow-run/{{{{ flow_run.id }}}}|Open in Prefect>\n" "State message: {{ flow_run.state.message }}" ), ) ], ) await client.create_automation(recovery_automation) print(f"Created automation: {RECOVERY_AUTOMATION_NAME}") else: print(f"Automation '{RECOVERY_AUTOMATION_NAME}' already exists. Skipping creation.") asyncio.run(main()) ``` Notes and options: - The key change is 
for_each=["related:deployment:id"]
which deduplicates per deployment instead of per flow run. - The 
after
gates make each automation fire only once per “streak” (failure or recovery) for that deployment. - If a deployment has never succeeded, the “first failure after success” automation won’t alert. If you want to also alert on the first-ever failure, we can discuss a variant. - You can further reduce noise by scoping with 
match_related
(e.g., only specific deployments or work pools). Docs for reference: - Event triggers and their fields: Event triggers - Creating automations: Creating automations - Events model (labels and related resources): Events concept Would you like this latching behavior per deployment, or per flow (non-deployment runs) instead? If per flow, we can switch 
for_each
back to 
["prefect.resource.id"]
.
7 Views
Open in Slack
PreviousNext
Powered by