@Marvin How do I access a value from an event in the parameters of an automation that's triggered by that event?

To access a value from an event in the parameters of an automation triggered by that event in Prefect, you typically use the event's attributes in your automation's trigger configuration. Here's a step-by-step approach to achieve this: ### 1. Define the Event Trigger You will need to define an event trigger that reacts to the presence or absence of an event. You can customize the triggers based on specific events and conditions in your Prefect workspace. For example, you might use an event trigger with specific labels or metrics. ### 2. Use Event Attributes in Actions When an event triggers your automation, you can access its attributes within the actions associated with the automation. Prefect provides flexibility in defining these automations programmatically. ### Example: The example below introduces the

Automation

class methods and demonstrates the use of an

EventTrigger

. This example will help you understand how to set up an automation that reacts to specific events.

python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from datetime import timedelta
from prefect.server.events.actions import CancelFlowRun

auto_to_create = Automation(
    name="Event Driven Automation Example",
    trigger=EventTrigger(
        expect={"event_type"}, # Replace with your specific event type
        match={
            "attribute_key": "attribute_value", # Replace with your specific event attributes
        },
        posture="Reactive",
        threshold=3,
        within=timedelta(seconds=10),
    ),
    actions=[CancelFlowRun()] # Replace with your specific actions
)

def create_automation():
    auto_to_create.create()

def read_automation(automation_id):
    return Automation.read(id=automation_id)

def update_automation(automation: Automation):
    automation.name = "new_name"
    automation.update()

def enable_automation(automation: Automation):
    automation.enable()

def disable_automation(automation: Automation):
    automation.disable()

### Useful Documentation Links: - Prefect Automations Guide - Prefect Automations Concepts ### Example in a GitHub Issue: A related example issue was discussed on GitHub, involving the implementation of automation with the SDK methods: GitHub Issue #12830. By defining appropriate triggers and actions, you can programmatically create and manage your automations, leveraging event attributes as needed. If you require more detailed guidance or encounter specific errors, these documents will be helpful. If there's something else you'd like to explore, just let me know. No, you probably still won't escape the tedium of existence, but at least you'll manage your Prefect automations more efficiently.

if you want to pass event data as parameters to a deployed flow, you can do it right in your deployment code (this example is from an az blob storage created event trigger)

from prefect import flow
from prefect.events import DeploymentEventTrigger

@flow
def reader_flow(file_url: str):
     # some flow definition here

if __name__=="__main__":
    reader_flow.deploy(
        "reader",
        ..., # other params
        triggers=[
            DeploymentEventTrigger(
                name="Storage Blob Created",
                match={
                    "prefect.resource.id": "<https://kevinstest.blob.core.windows.net/data/*>"
                },
                expect=["Microsoft.Storage.BlobCreated"],
                parameters={
                    "file_url": "{{ event.payload.data.url }}",
                },
            )
        ],
    )