https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Could you provide an example of a Com...
# ask-marvin
a
@Marvin Could you provide an example of a CompoundTrigger written in Python?
m
You want sample code for a CompoundTrigger? Cool 😎 But first, tell me your Prefect version - 2.x or 3.x?
a
@Marvin 2.x
m
Alright, here's an example of a CompoundTrigger in Prefect 2.x:
Copy code
python
from prefect.triggers import all_finished, some_failed, any_successful

# Define a compound trigger
compound_trigger = some_failed(all_finished, any_successful)

@prefect.flow(trigger=compound_trigger)
def my_flow():
    task_result = some_task()
    another_task(upstream_tasks=[task_result])

# Define your tasks as per usual
@prefect.task
def some_task():
    return "Task Completed"

@prefect.task
def another_task():
    print("Another Task Running")

my_flow()
This example uses triggers like 
all_finished
, 
some_failed
, and 
any_successful
. You can combine them to create complex dependencies. You can read more about it here. Let me know if you need anything else! 😃
a
@Marvin that’s not what I meant, this was the object I was referring to https://prefect-python-sdk-docs.netlify.app/prefect/events/?h=compoundtrigger#prefect.events.TriggerTypes
m
Got it, you're looking for CompoundTrigger examples in the context of 
prefect.events.TriggerTypes
. Here's an example:
Copy code
python
from prefect.events import CompoundTrigger

# Create your individual triggers
first_trigger = {
    "type": "event",
    "posture": "Proactive",
    "after": ["prefect.flow-run.Pending"],
    "expect": ["prefect.flow-run.Running", "prefect.flow-run.Crashed"],
    "for_each": ["prefect.resource.id"],
    "match_related": {
        "prefect.resource.name": "daily-customer-export",
        "prefect.resource.role": "flow"
    }
}

second_trigger = {
    "type": "event",
    "posture": "Reactive",
    "expect": ["prefect.work-pool.not-ready"],
    "match": {
        "prefect.resource.name": "kubernetes-workers",
    }
}

third_trigger = {
    "type": "metric",
    "metric": {
        "name": "lateness",
        "operator": ">",
        "threshold": 600,
        "range": 3600,
        "firing_for": 300
    }
}

# Create a compound trigger
compound_trigger = CompoundTrigger(
    require="any",
    triggers=[first_trigger, second_trigger, third_trigger]
)

# Output the compound trigger to verify 
print(compound_trigger)
This example demonstrates how you can combine multiple event and metric triggers into a 
CompoundTrigger
which will activate if any of the specified conditions are met. For more info, check the documentation. If you need more examples or details, let me know! 😃
a
@Marvin what is the 
require
argument in the CompoundTrigger function there?
m
The 
require
parameter in the 
CompoundTrigger
function determines the logical condition that must be met for the compound trigger to fire. Here’s a breakdown: 1. `"any"`: The compound trigger will fire if any of the individual triggers are satisfied. 2. `"all"`: The compound trigger will fire only if all of the individual triggers are satisfied. 3. A number (e.g., 
1
, 
2
): The compound trigger will fire if at least that number of individual triggers are satisfied. So, in the example I provided:
Copy code
python
compound_trigger = CompoundTrigger(
    require="any",  # It will trigger if any one of the individual triggers is satisfied
    triggers=[first_trigger, second_trigger, third_trigger]
)
You can replace 
"any"
with 
"all"
or a specific number based on your requirements. For more info, refer to the Prefect documentation. Let me know if you have more questions! 😃
n
we need more python examples for triggers in the docs
lemme see if i can find an example for you
but essentially its going to be the same "shape" as the yaml, just you'll use
Copy code
from prefect.events import DeploymentCompoundTrigger
a
Thanks @Nate, yea some examples are lacking in the docs. I almost got it working based on what Marvin shared above, I keep hitting
Copy code
pydantic.error_wrappers.ValidationError: 1 validation error for Automation
actions
  field required (type=value_error.missing)
I’m setting actions to 
actions=RunDeployment(deployment_id="e3eb9477-1273-47f0-95ad-d658e01d7972")
which works on a regular 
EventTrigger
but not with the 
CompoundTrigger
functions
n
can you show what you're trying with the whole trigger?
a
Sure
Copy code
first_trigger = {
        "type": "event",
        "posture": "Reactive",
        "match": {
            "prefect.resource.id":  [
                    "prefect.flow-run.*"
                ]},
        "expect": ["prefect.flow-run.Completed"],
        "match_related":{
        "prefect.resource.id": [
            "prefect.deployment.2682b223-75bd-48ac-888f-00f6ec44470f"
        ]}
    }

    second_trigger = {
        "type": "event",
        "posture": "Reactive",
        "match": {
            "prefect.resource.id":  [
                    "prefect.flow-run.*"
                ]},
        "expect": ["prefect.flow-run.Completed"],
        "match_related":{
        "prefect.resource.id": [
            "prefect.deployment.e36a5e8c-0ef7-4a99-b8ec-d385705c5965"
        ]}
    }

    automation = Automation(
        name="test cli compound trigger",
        trigger=DeploymentCompoundTrigger(require="all",
                                triggers=[first_trigger, second_trigger],
                                actions=RunDeployment(deployment_id="e3eb9477-1273-47f0-95ad-d658e01d7972"))

    ).create()
n
hrm should RunDeployment be in a list?
a
same error, I tried unlisting it
👍 1
Is it expecting an action within each of the event triggers?
n
oh wait something is weird, one sec
oh yeah okay so the actions field should go into the constructor of the Automation, not the trigger
Copy code
from prefect.automations import Automation
from prefect.events.actions import RunDeployment
from prefect.events.schemas.deployment_triggers import DeploymentCompoundTrigger

first_trigger = {
    "type": "event",
    "posture": "Reactive",
    "match": {"prefect.resource.id": ["prefect.flow-run.*"]},
    "expect": ["prefect.flow-run.Completed"],
    "match_related": {
        "prefect.resource.id": [
            "prefect.deployment.2682b223-75bd-48ac-888f-00f6ec44470f"
        ]
    },
}

second_trigger = {
    "type": "event",
    "posture": "Reactive",
    "match": {"prefect.resource.id": ["prefect.flow-run.*"]},
    "expect": ["prefect.flow-run.Completed"],
    "match_related": {
        "prefect.resource.id": [
            "prefect.deployment.e36a5e8c-0ef7-4a99-b8ec-d385705c5965"
        ]
    },
}

automation = Automation(
    name="test cli compound trigger",
    description="test cli compound trigger",
    enabled=True,
    trigger=DeploymentCompoundTrigger(
        name="test cli compound trigger",
        description="test cli compound trigger",
        within=None,
        enabled=True,
        parameters=None,
        job_variables=None,
        require="all",
        triggers=[first_trigger, second_trigger],
    ),
    actions=[RunDeployment(deployment_id="e3eb9477-1273-47f0-95ad-d658e01d7972")],
).create()
bc in general an Automation has • a trigger • a set of actions to perform when triggered
if you could, it would be super helpful if you could open an issue explaining that we need more examples of this in the docs, otherwise i can do it later
a
facepalm don’t know how I missed that
that fixed it!
Sure I can go about adding an issue!
n
catjam
appreciate it!
a
While I have you - is there anyway to list all automations/get all automation ids?
n
prefect automation ls
👀 1
a
amazing
thanks again!
n
👍
a
Is there anyway to do it in python? basically I want to create a process to version control and create new automations, and I was hoping to keep everything in a Python script
11 Views
Open in Slack
PreviousNext
Powered by