https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I need to configure automations with ...
# marvin-ai
r
@Marvin I need to configure automations with a Python script. I have the following function for generating the action: def autom_fail_mail(): return Automation( name="Send mail on flow error", trigger=EventTrigger( match={ "prefect.resource.id": "prefect.flow-run.*", }, match_related={ "prefect.resource.role": "flow", "prefect.resource.id": [ "prefect.flow.5d361213-6af5-41e4-be62-75308c8295cb", ], }, after=[], expect=[ "prefect.flow-run.Failed", "prefect.flow-run.Crashed", ], for_each=["prefect.resource.id"], posture="Reactive", threshold=1, within=0, ), actions=[RunDeployment(deployment_id="af4b3c2b-9913-4307-8ad4-76de4dc697d5")], ) I need to replace hard-coded deployment_id by finding the deployment by name. How can I do that?
m
✅ thought for 137.0 seconds
Yes — use the Prefect client to look up the deployment by its full name, then pass 
deployment.id
into 
RunDeployment
. In Prefect 3.x, the lookup method is 
read_deployment_by_name(...)
, and the name format is: 
"<flow-name>/<deployment-name>"
Example:
Copy code
python
from prefect.client.orchestration import get_client
from prefect.automations import Automation
from prefect.events.actions import RunDeployment
from prefect.events.schemas.automations import EventTrigger

async def autom_fail_mail():
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

    return Automation(
        name="Send mail on flow error",
        trigger=EventTrigger(
            match={
                "prefect.resource.id": "prefect.flow-run.*",
            },
            match_related={
                "prefect.resource.role": "flow",
                "prefect.resource.id": [
                    "prefect.flow.5d361213-6af5-41e4-be62-75308c8295cb",
                ],
            },
            after=[],
            expect=[
                "prefect.flow-run.Failed",
                "prefect.flow-run.Crashed",
            ],
            for_each=["prefect.resource.id"],
            posture="Reactive",
            threshold=1,
            within=0,
        ),
        actions=[
            RunDeployment(
                source="selected",
                deployment_id=deployment.id,
            )
        ],
    )
If you want a sync version:
Copy code
python
from prefect.client.orchestration import get_client
from prefect.automations import Automation
from prefect.events.actions import RunDeployment
from prefect.events.schemas.automations import EventTrigger

def autom_fail_mail():
    with get_client(sync_client=True) as client:
        deployment = client.read_deployment_by_name(
            "my-flow/my-deployment"
        )

    return Automation(
        name="Send mail on flow error",
        trigger=EventTrigger(
            match={
                "prefect.resource.id": "prefect.flow-run.*",
            },
            match_related={
                "prefect.resource.role": "flow",
                "prefect.resource.id": [
                    "prefect.flow.5d361213-6af5-41e4-be62-75308c8295cb",
                ],
            },
            after=[],
            expect=[
                "prefect.flow-run.Failed",
                "prefect.flow-run.Crashed",
            ],
            for_each=["prefect.resource.id"],
            posture="Reactive",
            threshold=1,
            within=0,
        ),
        actions=[
            RunDeployment(
                source="selected",
                deployment_id=deployment.id,
            )
        ],
    )
A couple of important notes: - 
read_deployment_by_name()
expects 
flow_name/deployment_name
, not just the deployment name - for 
RunDeployment
, if you specify 
deployment_id
, set 
source="selected"
- 
deployment.id
is the UUID you want If useful, I can also show you how to: - look up the flow ID by flow name instead of hard-coding that 
prefect.flow...
UUID - create/update the automation end-to-end from Python Relevant API docs: - Prefect orchestration client - Automations and triggers - Automation actions
r
@Marvin I would like to look up the flow ID by name, as you suggested (instead of hard-coding it).
m
✅ thought for 177.9 seconds
Yes — you can look up the flow by name with the client, then use 
flow.id
when building 
match_related
. The simplest method is: 
read_flow_by_name("your-flow-name")
Example with your automation:
Copy code
python
from prefect.client.orchestration import get_client
from prefect.automations import Automation
from prefect.events.actions import RunDeployment
from prefect.events.schemas.automations import EventTrigger

async def autom_fail_mail():
    async with get_client() as client:
        flow = await client.read_flow_by_name("my-flow-name")
        deployment = await client.read_deployment_by_name(
            "my-flow-name/my-deployment-name"
        )

    return Automation(
        name="Send mail on flow error",
        trigger=EventTrigger(
            match={
                "prefect.resource.id": "prefect.flow-run.*",
            },
            match_related={
                "prefect.resource.role": "flow",
                "prefect.resource.id": [
                    f"prefect.flow.{flow.id}",
                ],
            },
            after=[],
            expect=[
                "prefect.flow-run.Failed",
                "prefect.flow-run.Crashed",
            ],
            for_each=["prefect.resource.id"],
            posture="Reactive",
            threshold=1,
            within=0,
        ),
        actions=[
            RunDeployment(
                source="selected",
                deployment_id=deployment.id,
            )
        ],
    )
If you want the sync version:
Copy code
python
from prefect.client.orchestration import get_client
from prefect.automations import Automation
from prefect.events.actions import RunDeployment
from prefect.events.schemas.automations import EventTrigger

def autom_fail_mail():
    with get_client(sync_client=True) as client:
        flow = client.read_flow_by_name("my-flow-name")
        deployment = client.read_deployment_by_name(
            "my-flow-name/my-deployment-name"
        )

    return Automation(
        name="Send mail on flow error",
        trigger=EventTrigger(
            match={
                "prefect.resource.id": "prefect.flow-run.*",
            },
            match_related={
                "prefect.resource.role": "flow",
                "prefect.resource.id": [
                    f"prefect.flow.{flow.id}",
                ],
            },
            after=[],
            expect=[
                "prefect.flow-run.Failed",
                "prefect.flow-run.Crashed",
            ],
            for_each=["prefect.resource.id"],
            posture="Reactive",
            threshold=1,
            within=0,
        ),
        actions=[
            RunDeployment(
                source="selected",
                deployment_id=deployment.id,
            )
        ],
    )
A few notes: - 
read_flow_by_name()
does an exact name lookup - if the flow does not exist, the client will raise an error - for the deployment lookup, the format is still 
flow-name/deployment-name
If you want, you can factor the resource ID formatting into a helper:
Copy code
python
def flow_resource_id(flow_id):
    return f"prefect.flow.{flow_id}"
Then use:
Copy code
python
"prefect.resource.id": [flow_resource_id(flow.id)]
If you'd like, I can also show a full example that creates or updates the automation in Prefect Cloud from this Python definition.
Open in Slack
PreviousNext
Powered by