@Marvin is it possible to set up an automation for flow run errors but only on certain deployments of the flow?

Could you please confirm which version of Prefect you're using? Setting up automation for flow run errors can differ between Prefect 2.x and Prefect 3.x. If you're unsure about the version, any details you provide about your setup may help me determine it.

Prefect 3.x

In both Prefect 2.x and Prefect 3.x, it's possible to set up automation for flow run errors specifically for certain deployments. For Prefect 2.x, you can utilize the Prefect API to create event-driven automations by specifying a deployment filter when setting up automation for dealing with flow run errors. You can define automations programmatically using the Prefect Client, filtering based on deployment names, as shown in this guide. For Prefect 3.x, the process is similar, but you have more flexibility with the Automations API. You can define custom triggers and actions, such as responding to specific state changes for flow runs within certain deployments. The triggers can be based on various conditions including deployment status, and you can programmatically manage these settings through the API. More details can be found in the Prefect 3.x documentation here. Both versions allow you to filter by deployment to ensure that your automation only applies to specific deployments. If you need any specific code examples or further assistance, feel free to ask!

Are there any examples of using the Automations API or using the Automation in the python sdk

Yes, there are examples of using the Automations API with the Python SDK in both Prefect 2.x and Prefect 3.x. Prefect 2.x: You can use the Automations API to create, update, and manage automations using a payload in YAML or JSON format. Here's a snippet of how you might programmatically create or update an automation:

python
import requests
from yaml import safe_load

def create_or_update_automation():
    with open("automation_payload.yaml", "r") as fh:
        payload = safe_load(fh)
    # Example code to find existing automations and update or create one
    automations = <http://requests.post|requests.post>("/automations/filter")
    existing_automation = next((a["id"] for a in automations if a["name"] == payload["name"]), None)
    if existing_automation:
        print(f"Automation '{payload['name']}' already exists and will be updated")
        requests.put(f"/automations/{existing_automation}", payload=payload)
    else:
        print(f"Creating automation '{payload['name']}'")
        <http://requests.post|requests.post>("/automations/", payload=payload)

if __name__ == "__main__":
    create_or_update_automation()

This example demonstrates the basic use of the API via HTTP requests. You can extend this by integrating webhooks to trigger automations based on events. Full details are available here. Prefect 3.x: In Prefect 3.x, you have access to a more refined API and can use the Python SDK's

Automation

class directly. Here is an example:

python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.server.events.actions import CancelFlowRun

# Creating an automation
automation = Automation(
    name="Example Automation",
    trigger=EventTrigger(
        expect={"event.name"},
        match={"attribute": "value"},
        posture="Reactive",
        threshold=3,
    ),
    actions=[CancelFlowRun()]
).create()

print(automation)

In this snippet, the

Automation

class is used to define an automation with specified triggers and actions, such as canceling a flow run when certain conditions are met. The full example and details can be found here. Feel free to modify these examples to fit your specific use case! If you need further assistance, don't hesitate to ask.