< Marvin> I created an automation through python to cancel f Prefect Community #ask-marvin

<@ULVA73B9P> I created an automation through pytho...

Isabel

06/26/2024, 11:33 AM

@Marvin I created an automation through python to cancel flow runs that have been pending for 5 minutes from flows with a specific tag. When I create the automation, it ends up disabled in the UI and in the UI I cannot see the conditions for this automation. Is this expected or am I doing something wrong?

Marvin

06/26/2024, 11:33 AM

It appears there might be a few things to consider regarding your issue: 1. Creating Automations in Python: - You can create and access automations with the Python SDK's

Automation

class and its methods. Here's an example snippet for creating an automation:

Copy code

python
    from prefect.automations import Automation
    from prefect.events.schemas.automations import EventTrigger
    from prefect.server.events.actions import CancelFlowRun
    
    automation = Automation(
        name="cancel_pending_flows",
        trigger=EventTrigger(
            expect={"flow_run.state": "pending"},
            match={"flow_run.tags": ["your_specific_tag"]}
        ),
        actions=[CancelFlowRun()]
    )

Source 2. Automation Visibility and Management in Prefect Cloud: - The Automations page in Prefect Cloud provides an overview of all configured automations for your workspace. By selecting the name of an automation, you can view its details and relevant events. - It's also mentioned that the toggle next to an automation lets you pause or enable the automation, and you can find commands to edit or delete the automation next to this toggle. 3. Automations End Up Disabled: - There's no specific mention in the documentation about automations ending up disabled by default when created via Python. This might not be expected behavior and may require further investigation or reaching out to Prefect support/community. It might be useful to ensure that your automation has the correct permissions, and there are no restrictions or conditions that could disable it by default. For further actions, you could enable the automation manually in the UI and monitor for any automated changes. For detailed documentation, you can refer to the Prefect Automations Documentation. It looks like there were no specific open issues related to automations being disabled upon creation, so this might be something peculiar to your setup or requiring support. If you need more personalized assistance, consider reaching out to the Prefect community or support for detailed troubleshooting.

Isabel

06/26/2024, 11:36 AM

@Marvin How do I define to only cancel flows that were pending for more than 5 minutes?

Marvin

06/26/2024, 11:37 AM

To define an automation that cancels flow runs pending for more than 5 minutes (300 seconds), you can use the

EventTrigger

and specify a duration condition. Here's an example:

Copy code

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

automation = Automation(
    name="cancel_long_pending_flows",
    trigger=EventTrigger(
        expect={
            "flow_run.state": "pending",
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        },
        match={"flow_run.tags": ["your_specific_tag"]}
    ),
    actions=[CancelFlowRun()]
)

# Save or register the automation as needed

Explanation: - The

expect

key includes a condition

"flow_run.state.timestamp.elapsed": {"greater_than": 300}

which checks if the flow has been in the 'pending' state for more than 300 seconds (5 minutes). - The

match

condition ensures it only applies to flows with your specified tag. You can find more details about configuring automations in the Prefect Automations Documentation. If this doesn't directly apply to your setup or further fine-tuning is needed, consulting the Prefect Community may provide additional tailored assistance.

Isabel

06/26/2024, 11:41 AM

@Marvin creating the automation as this:

Copy code

automation = Automation(
    name="Cancel Pending Flows",
    trigger=EventTrigger(
        xpect={
            "flow_run.state": "pending",
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        },
        match={"flow_run.tags": ["XXXX"]}
    ),
    actions=[CancelFlowRun()],
).create()

Makes the UI show this:

Marvin

06/26/2024, 11:41 AM

It seems you encountered an issue with how the automation is displayed or works within the UI. Let's troubleshoot the potential problems: 1. Correcting Typographical Errors: - Ensure there are no typographical errors in your code. For instance,

xpect

should be

expect

. 2. Creating Automations with Proper Methods: - Confirming the proper method to create and register the automation. Update

Automation

instantiation and method call for creation if needed. Here's a refined version of your code snippet:

Copy code

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

# Define the automation
automation = Automation(
    name="Cancel Pending Flows",
    trigger=EventTrigger(
        expect={
            "flow_run.state": "pending",
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        },
        match={"flow_run.tags": ["XXXX"]}
    ),
    actions=[CancelFlowRun()],
)

# Create and register the automation
automation.create()

### Debugging Display Issues in the UI 1. Check Prefect Cloud Configuration: - Ensure that your Prefect Cloud account has the necessary permissions and configurations to support custom automations. 2. UI Update Lag: - Sometimes, the UI may not immediately reflect changes. Refresh the page or log out and back in again to see updates. 3. Logs and Feedback: - Review any logs or error messages in the Prefect dashboard to see if there are hints about why the automation might not be displayed correctly. 4. Support and Community Help: - If the problem persists, consider reaching out on the Prefect Discourse Community or contacting Prefect Support for detailed troubleshooting. For detailed information, refer to the Prefect Automations Documentation. If further customization or debugging is needed, consulting the Prefect community can also provide insights.

Isabel

06/26/2024, 11:41 AM

@Marvin it is not clear that the automation is canceling pending flows after 5 minutes

Isabel

06/26/2024, 11:45 AM

@Marvin the code you sent fails:

Copy code

Traceback (most recent call last):
  File "/Users/isabel/Documents/Git/GitHub/data-platform-workflows/prefect2/automations/dataanalyticsengineering/dae_automations.py", line 9, in <module>
    trigger=EventTrigger(
  File "pydantic/main.py", line 341, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for EventTrigger
expect
  value is not a valid set (type=type_error.set)

Marvin

06/26/2024, 11:45 AM

It looks like the issue is caused by the

expect

field needing to be a set rather than a dictionary. Let's refine the code to meet the expected schema requirements. Here’s how you can properly define the automation trigger for flows pending for more than 5 minutes: ### Updated Code:

Copy code

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

# Define the automation
automation = Automation(
    name="Cancel Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},
        match={
            "flow_run.tags": ["XXXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[CancelFlowRun()],
)

# Create and register the automation
automation.create()

### Explanation: - EventTrigger.expect: This should be a set (denoted with curly braces

{}

). Here, it specifically looks for the

"flow_run.state.pending"

event. - EventTrigger.match: Includes additional criteria such as

"flow_run.tags"

and

"flow_run.state.timestamp.elapsed"

. ### Implementation Tips: 1. Ensure Valid Syntax: Note the use of the set format for

expect

. 2. Testing: Test by creating a flow and tagging it

"XXXX"

, then ensuring it gets canceled after 5 minutes. For further documentation and examples, refer to the Prefect Automations Documentation. If you continue to face issues, the Prefect community or support can provide more personalized guidance.

Isabel

06/27/2024, 12:27 PM

@Marvin could you show me an example of a python automation which sends a notification to slack which is defined by a block in prefect?

Marvin

06/27/2024, 12:27 PM

Here's an example of how you can define an automation in Python that sends a notification to Slack using a Prefect block. Make sure you have the Slack block configured in Prefect. ### Example Code: 1. Configure the Slack Block: - First, create a Slack block in the Prefect UI, which you will use in your automation. 2. Create the Automation Script:

Copy code

python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.server.events.actions import SendSlackMessage
from prefect.blocks.notifications import SlackWebhook

# Load your Slack webhook block
slack_webhook_block = SlackWebhook.load("your-slack-block-name")

# Define the message content
message = "Flow run has been pending for more than 5 minutes!"

# Define the automation
automation = Automation(
    name="Notify Slack for Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},
        match={
            "flow_run.tags": ["XXXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[SendSlackMessage(webhook=slack_webhook_block.url, message=message)]
)

# Create and register the automation
automation.create()

### Explanation: - SlackWebhook Block: This is where you load the Slack webhook block using

SlackWebhook.load("your-slack-block-name")

. - Message Content: Defining the notification message you want to send. - Automation Setup: Similar to the previous examples, but this time the action is

SendSlackMessage

. ### Running the Automation: - Save this script and run it in your environment. - Ensure the automation has been created by checking the Prefect UI. ### Additional Resources: - Prefect Automations Documentation - Prefect Slack Notification Documentation If you need further customization, review these documents or reach out to the Prefect Community for more tailored support.

Isabel

06/27/2024, 12:29 PM

@Marvin it seems like SendSlackMessage doesn’t exist in prefect 2. Also the documentation under https://docs.prefect.io/latest/notifications/slack/ does not exist

Marvin

06/27/2024, 12:30 PM

It looks like the previous approach using

SendSlackMessage

is not applicable to Prefect 2.0. Instead, you can use the

prefect-slack

integration to send a Slack message. Here's an updated example based on the latest available details: ### Steps: 1. Install Prefect Slack Integration:

Copy code

bash
   pip install prefect-slack

2. Set Up the Slack Block: - Create a Slack block in the Prefect UI with the relevant credentials for your Slack webhook. 3. Create the Automation Script: Here’s a complete example:

Copy code

python
from prefect import flow, task
from prefect.blocks.notifications import SlackWebhook
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import ShellAction

# Define a task to send a Slack message
@task
async def send_slack_message(message: str):
    slack_webhook_block = SlackWebhook.load("your-slack-block-name")
    await slack_webhook_block.notify(message)

# Define the automation
automation = Automation(
    name="Notify Slack for Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},
        match={
            "flow_run.tags": ["XXXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[
        ShellAction(
            command="python -c 'from your_script import send_slack_message; import asyncio; asyncio.run(send_slack_message(\"Flow run has been pending for more than 5 minutes!\"))'"
        )
    ]
)

# Create and register the automation
automation.create()

# Example Flow to demonstrate
@flow
def example_flow():
    # Your flow implementation
    pass

# Register the flow (adapt as necessary for your registration mechanism)
example_flow()

### Explanation: - SlackWebhook Block: Use

SlackWebhook.load("your-slack-block-name")

to load your Slack webhook block. - Async Task: The

send_slack_message

task uses the block to send a notification. - Automation Setup: Creates an automation with a

ShellAction

to call the

send_slack_message

function. ### Key Points: - Adjust

"your_slack_block_name"

to the name of your actual Slack block. - Ensure the

ShellAction

command correctly references the Python file and function. ### Useful Links: - Prefect Slack Integration Documentation - Prefect Automations Documentation This setup should correctly send Slack notifications for flow runs pending longer than 5 minutes. If issues persist, consult the Prefect Community or Prefect Support for additional assistance.

Isabel

06/27/2024, 12:31 PM

@Marvin is there no way of using

from prefect.server.events.actions import SendNotification

to send notifications using a prefect block?

Marvin

06/27/2024, 12:31 PM

Yes, you can use

SendNotification

to send notifications using a Prefect block. Here's an example of how you can define an automation to send a Slack notification: ### Example Code: 1. Install Prefect Slack Integration:

Copy code

bash
   pip install prefect-slack

2. Set Up the Slack Block: - Create a Slack block in the Prefect UI and save the block's name for use in the script. 3. Create the Automation Script:

Copy code

python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import SendNotification
from prefect_slack import SlackWebhook
from prefect import flow

# Define the Slack block name you have created in the Prefect UI
SLACK_BLOCK_NAME = "your-slack-block-name"

# Load the Slack block
slack_webhook_block = SlackWebhook.load(SLACK_BLOCK_NAME)

# Define the message content
message_subject = "Alert: Pending Flow Run"
message_body = "A flow run has been pending for more than 5 minutes!"

# Define the automation
automation = Automation(
    name="Notify Slack for Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},
        match={
            "flow_run.tags": ["XXXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[
        SendNotification(
            block_document_id=slack_webhook_block.id,
            subject=message_subject,
            body=message_body,
        )
    ]
)

# Create and register the automation
automation.create()

# Example Flow to demonstrate
@flow
def example_flow():
    # Your flow implementation
    pass

# Run the example flow to test (you can also register this flow if required)
example_flow()

### Explanation: - SlackWebhook Block: Use

SlackWebhook.load(SLACK_BLOCK_NAME)

to load the Slack webhook block. - Message Content: Define the Slack message subject and body. - Automation Setup: Creates an automation with

SendNotification

action using the loaded Slack block. - EventTrigger: Specifies the condition to match flow runs pending for more than 5 minutes with specific tags. - SendNotification Action: Leverages the block's ID to send the notification. This script defines and creates an automation that sends a Slack notification via a Slack webhook block when a flow run is pending for more than 5 minutes. ### Resources: - For further details and configuration, refer to the Prefect Slack Integration Documentation. Should you run into any issues, the Prefect Community and Prefect Support are excellent resources for assistance.

Isabel

06/27/2024, 12:34 PM

@Marvin also in the UI I can add into the message the flow name and run ID etc. How can I add this to the message through python?

Marvin

06/27/2024, 12:35 PM

To add dynamic information such as the flow name and run ID to the notification message, you can use Jinja templating. Prefect supports this to allow messages to include information about the flow run context. Here’s how you can modify your script to include dynamic content in the message using Jinja templates: ### Example Code: 1. Install Prefect Slack Integration:

Copy code

bash
   pip install prefect-slack

2. Set Up the Slack Block: - Create a Slack block in the Prefect UI and save the block's name for use in the script. 3. Create the Automation Script:

Copy code

python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import SendNotification
from prefect_slack import SlackWebhook
from prefect import flow

# Define the Slack block name you have created in the Prefect UI
SLACK_BLOCK_NAME = "your-slack-block-name"

# Load the Slack block
slack_webhook_block = SlackWebhook.load(SLACK_BLOCK_NAME)

# Define the message subject and body using Jinja templates
message_subject = "Alert: Pending Flow Run"
message_body = """
Flow {{ event.payload.flow_run.name }} has been pending for more than 5 minutes.
Run ID: {{ event.payload.flow_run.id }}
Start Time: {{ event.payload.flow_run.start_time }}
"""

# Define the automation
automation = Automation(
    name="Notify Slack for Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},
        match={
            "flow_run.tags": ["XXXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[
        SendNotification(
            block_document_id=slack_webhook_block.id,
            subject=message_subject,
            body=message_body,
        )
    ]
)

# Create and register the automation
automation.create()

# Example Flow to demonstrate
@flow
def example_flow():
    # Your flow implementation
    pass

# Run the example flow to test (you can also register this flow if required)
example_flow()

### Explanation: - Jinja Templating: The

message_body

uses Jinja template syntax to include dynamic content such as flow name, run ID, and start time. - `{{ event.payload.flow_run.name }}`: The name of the flow run. - `{{ event.payload.flow_run.id }}`: The unique ID of the flow run. - `{{ event.payload.flow_run.start_time }}`: The start time of the flow run. - SendNotification Action: Utilizes the Jinja templated message body. ### Useful Links: - Jinja Templating Documentation - Prefect Automations Documentation - Prefect Slack Integration ### Testing and Validation: - Ensure the Slack block name is correct and tests the automation by running a flow with the specified conditions. - Monitor the Slack channel to verify that messages are being sent with the correct dynamic content. This should help you include flow-specific information in your Slack notifications, making them more informative and actionable. If you encounter issues, the Prefect Community and Prefect Support are good resources for assistance.

Isabel

06/27/2024, 12:38 PM

@Marvin using the slack block name is not working

Marvin

06/27/2024, 12:38 PM

Let's ensure we're correctly referencing and using the Slack Block for sending notifications. Here’s an adjusted version that ensures the block is correctly referenced and properly passes the block document ID to the

SendNotification

action. ### Pre-requisites 1. Install Prefect Slack Integration:

Copy code

bash
   pip install prefect-slack

2. Create and Configure Slack Block: - Ensure you have created a Slack block in the Prefect UI and take note of its name. ### Adjusted Script:

Copy code

python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import SendNotification
from prefect.blocks.notifications import SlackWebhook
from prefect import flow

# Define the Slack block name you've created in the Prefect UI
SLACK_BLOCK_NAME = "your-slack-block-name"

# Load the SlackWebhook block
slack_webhook_block = SlackWebhook.load(SLACK_BLOCK_NAME)

# Ensure that the correct block document ID is utilized
slack_block_id = slack_webhook_block._block_document_id

# Define the message subject and body using Jinja templates
message_subject = "Alert: Pending Flow Run"
message_body = """
Flow {{ event.payload.flow_run.name }} has been pending for more than 5 minutes.
Run ID: {{ event.payload.flow_run.id }}
Start Time: {{ event.payload.flow_run.start_time }}
"""

# Define the automation
automation = Automation(
    name="Notify Slack for Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},
        match={
            "flow_run.tags": ["XXXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[
        SendNotification(
            block_document_id=slack_block_id,  # Use the block document ID
            subject=message_subject,
            body=message_body,
        )
    ]
)

# Create and register the automation
automation.create()

# Example Flow to demonstrate
@flow
def example_flow():
    # Your flow implementation
    pass

# Run the example flow to test (you can also register this flow if required)
example_flow()

### Key Adjustments: - Using Block Document ID: The critical change here is ensuring that we correctly reference the block document ID when using

SendNotification

. This is done by accessing the

_block_document_id

attribute of the

SlackWebhook

block instance. ### Testing and Troubleshooting: - Verify Block Name: Double-check that

SLACK_BLOCK_NAME

matches exactly with the block name configured in the Prefect UI. - Run Test Flow: Execute the script and ensure that the automation triggers the Slack notification as expected. ### Additional Resources: - Prefect Documentation - Prefect Slack Integration - Jinja Templating If further issues arise, consulting the Prefect Community on Discourse or contacting Prefect Support would be beneficial.

Isabel

06/27/2024, 12:49 PM

@Marvin another question, when setting up this automation:

Copy code

pending_automation = Automation(
    name="Cancel Pending Flows",
    trigger=EventTrigger(
        expect={
            "flow_run.state": "pending",
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        },
        match={"flow_run.tags": ["XXX"]}
    ),
    actions=[CancelFlowRun()],
).create()

I get the following error message:

Copy code

trigger=EventTrigger(
  File "pydantic/main.py", line 341, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for EventTrigger
expect
  value is not a valid set (type=type_error.set)

What is the issue?

Marvin

06/27/2024, 12:50 PM

The error you are encountering indicates that the

expect

field of the

EventTrigger

class is expecting a set, not a dictionary. Let's correct the

EventTrigger

instantiation accordingly. ### Corrected Script: Here is the corrected example demonstrating how to properly create and register the automation:

Copy code

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

# Define the automation
pending_automation = Automation(
    name="Cancel Pending Flows",
    trigger=EventTrigger(
        expect={"flow_run.state.pending"},  # This should be a set, using curly braces
        match={
            "flow_run.tags": ["XXX"],
            "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
        }
    ),
    actions=[CancelFlowRun()],
)

# Create and register the automation
pending_automation.create()

### Explanation: - Expect Field as a Set: The

expect

field must be defined as a set. In this case,

"flow_run.state.pending"

is placed inside curly braces to form a set. ### Example Flow to Demonstrate: Adding an example flow to test the automation:

Copy code

python
from prefect import flow

@flow
def example_flow():
    # Example flow implementation
    pass

# Run the example flow to test (or register it if needed)
example_flow()

### Key Points: 1. Ensure
expect
is a Set: This is a critical part of the correction. The

expect

field should be defined as a set using curly braces

{}

. 2. Monitor Prefect UI for Automation Creation: After running the script, check the Prefect UI to ensure that the automation is created correctly. ### Additional Checklist: - Prefect Version: Ensure you are using a compatible version of Prefect and its dependencies. - Prefect Cloud Credentials: Ensure that you have proper access and credentials configured for Prefect Cloud. With these corrections, the automation should be successfully created, and you should no longer see the

ValidationError

. If further issues arise, checking the Prefect Documentation and reaching out to the Prefect Community could provide additional insights.

Isabel

06/27/2024, 12:52 PM

@Marvin if you run the create code more than once, will it create multiple identical automations or will it know they are duplicates?

Marvin

06/27/2024, 12:52 PM

Prefect does not inherently deduplicate automations created using the Python SDK. If you run the create code more than once, it will likely create multiple identical automations. Prefect Cloud does not automatically check for duplicates before creating a new automation. To ensure idempotent behavior (creating an automation only if it doesn't already exist), you can implement a check in your code. Here is an example of how you can adapt your script to check for existing automations before creating a new one: ### Example Script to Check for Existing Automation:

Copy code

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

async def get_existing_automations():
    async with get_client() as client:
        response = await client.read_automations()
        return response

# Define the automation details
automation_name = "Cancel Pending Flows"
slack_message_subject = "Alert: Pending Flow Run"
slack_message_body = """
Flow {{ event.payload.flow_run.name }} has been pending for more than 5 minutes.
Run ID: {{ event.payload.flow_run.id }}
Start Time: {{ event.payload.flow_run.start_time }}
"""

# Check if an automation with the same name already exists
existing_automations = await get_existing_automations()
automation_exists = any(a['name'] == automation_name for a in existing_automations)

# Define and create the automation if it doesn't exist
if not automation_exists:
    # Define the automation
    pending_automation = Automation(
        name=automation_name,
        trigger=EventTrigger(
            expect={"flow_run.state.pending"},  # Must be a set
            match={
                "flow_run.tags": ["XXX"],
                "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
            }
        ),
        actions=[CancelFlowRun()],
    )

    # Create and register the automation
    pending_automation.create()
else:
    print(f"Automation '{automation_name}' already exists.")

### Explanation: 1. Get Existing Automations: The

get_existing_automations

function queries the Prefect API to get a list of current automations. 2. Check for Duplicates: Before creating a new automation, the script checks if an automation with the same name already exists. 3. Conditional Creation: The automation is created only if it doesn't already exist. ### Running this Script: 1. Ensure the Prefect client is authenticated and has appropriate permissions. 2. Execute the script in your environment. ### Additional Resources: - Prefect API Documentation - Prefect Automations Documentation By incorporating this check, you can prevent creating multiple identical automations and ensure cleaner and more manageable automation setups. If you encounter issues or need more advanced deduplication strategies, consult the Prefect Community and support channels.

Isabel

06/27/2024, 12:54 PM

@Marvin does the code above work in prefect 2?

Marvin

06/27/2024, 12:54 PM

Prefect 2.x does offer API methods to manage and read automations, although the specific example code I provided above assumes some API client methods from Prefect 1.x that may not be directly applicable or available. Here's how you can achieve the same result using compatible methods in Prefect 2.x: ### Revised Script for Prefect 2.x: 1. Install Prefect and PREFECT-SLACK Integration:

Copy code

bash
   pip install prefect prefect-slack

2. Ensure you have your Prefect Cloud or Server set up properly. 3. Create and Use Automations in Prefect 2.x:

Copy code

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

# Define the Prefect Cloud API endpoint and your API token
API_URL = "<https://api.prefect.io>"
API_TOKEN = "your_api_token"

async def get_existing_automations(name):
    async with httpx.AsyncClient() as client:
        headers = {"Authorization": f"Bearer {API_TOKEN}"}
        response = await client.get(f"{API_URL}/automations", headers=headers)
        response.raise_for_status()
        automations = response.json()
        return any(auto['name'] == name for auto in automations)

# Define the automation details
automation_name = "Cancel Pending Flows"

# Check if an automation with the same name already exists
automation_exists = await get_existing_automations(automation_name)

# Define and create the automation if it doesn't exist
if not automation_exists:
    # Define the automation
    pending_automation = Automation(
        name=automation_name,
        trigger=EventTrigger(
            expect={"flow_run.state.pending"},  # Must be a set
            match={
                "flow_run.tags": ["XXX"],
                "flow_run.state.timestamp.elapsed": {"greater_than": 300}  # greater than 5 minutes
            }
        ),
        actions=[CancelFlowRun()],
    )

    # Create and register the automation
    await pending_automation.create()
else:
    print(f"Automation '{automation_name}' already exists.")

### Explanation: 1. API Token and Endpoint: Set up your Prefect Cloud API token and endpoint. 2. HTTPX Client: Using

httpx

, an asynchronous HTTP client for making HTTP/1.1 requests. 3. Check Existing Automations: Function

get_existing_automations

fetches the list of existing automations and checks if an automation with the same name already exists. 4. Create Automation: Creates and registers the automation only if it does not already exist. ### Notes: - This script relies on Prefect's API for listing automations. Make sure you have the correct API endpoint and authorization token. - It's an async/await script, so ensure your environment supports asynchronous execution (e.g., an event loop in Jupyter, or running in an async-compatible environment). ### Testing: - Run Script: Execute the script in your environment and confirm that it behaves as expected, preventing duplicate automations. - Prefect UI: Check your Prefect Cloud or Server UI to ensure automations are listed correctly. If this script requires further tuning or support, please refer to the Prefect API documentation and engage with the Prefect Community for more specific guidance related to your setup.

5 Views

Open in Slack

Previous Next