@Marvin give me an example of using a trigger to deploy a second deployment from python using .deploy function. The decond deplayment is triggered after the flow run on a first deployment completes successfully.

To achieve the triggering of a second deployment upon the successful completion of a flow run in Prefect 3.x, you can use the

on_completion

hook in combination with the

run_deployment

function. Here's an example:

python
from prefect import flow, deploy
from prefect.deployments import run_deployment

@flow
def first_flow():
    # Your flow logic here
    pass

@flow
def second_flow():
    # Logic for the second flow
    pass

# Create a deployment for the second flow
second_flow_deployment = deploy(
    flow=second_flow,
    name="second-flow-deployment"
)

# Define a completion hook to trigger the second deployment
@first_flow.on_completion
def trigger_second_deployment(state):
    if state.is_completed():
        run_deployment(name="second-flow-deployment")

# Deploy the first flow
first_flow_deployment = deploy(
    flow=first_flow,
    name="first-flow-deployment"
)

if __name__ == "__main__":
    first_flow_deployment()

In this example, the

first_flow

has an

on_completion

hook that checks if the flow run completed successfully. If it did, it triggers the

second_flow

deployment using

run_deployment

. Make sure to adjust the deployment names as needed. For more information, you can refer to the Prefect documentation on deployment concepts.

@Marvin modify your code to accomodate for when the flow sources are coming from github

To deploy flows from GitHub using Prefect 3.x, you'll need to configure a GitHub Actions workflow. Here's a modified example incorporating GitHub as the source for your flows: 1. GitHub Workflow Configuration: Create a

.github/workflows/deploy-prefect-flow.yaml

file in your repository with the following content:

yaml
name: Deploy Prefect Flow

on:
  push:
    branches:
      - main

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.x'

      - name: Install Prefect
        run: pip install prefect

      - name: Deploy Flow
        env:
          PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}
          PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
        run: |
          prefect deploy --name first-flow-deployment --entrypoint path/to/first_flow.py:first_flow
          prefect deploy --name second-flow-deployment --entrypoint path/to/second_flow.py:second_flow

2. Python Script: Modify your Python script to use the GitHub-deployed flows:

python
from prefect import flow
from prefect.deployments import run_deployment

@flow
def first_flow():
    # Your flow logic here
    pass

@flow
def second_flow():
    # Logic for the second flow
    pass

# Define a completion hook to trigger the second deployment
@first_flow.on_completion
def trigger_second_deployment(state):
    if state.is_completed():
        run_deployment(name="second-flow-deployment")

if __name__ == "__main__":
    # This will be handled by GitHub Actions
    pass

Notes: - Ensure you have set the

PREFECT_API_KEY

and

PREFECT_API_URL

as secrets in your GitHub repository under Settings -> Secrets and variables -> Actions. - Adjust the

entrypoint

paths in the GitHub Actions file to match the location of your flow scripts in the repository. For more detailed guidance, refer to the Prefect CI/CD deployment documentation.

@Marvin I want to use a trigger passed to .deploy function flow.from_source().deploy(triggers=[sometrigger]) for example

To deploy a flow with triggers using the

.deploy

function in Prefect 3.x, you can define triggers directly in your Python code. Prefect supports different types of triggers like

DeploymentEventTrigger

,

DeploymentMetricTrigger

,

DeploymentCompoundTrigger

, and

DeploymentSequenceTrigger

from the

prefect.events

module. Here's an example of how you might set this up:

python
from prefect import flow, deploy
from prefect.events import DeploymentEventTrigger

@flow
def my_flow():
    # Your flow logic here
    pass

# Define a trigger
trigger = DeploymentEventTrigger(
    event="external.resource.pinged",
    resource="my.external.resource"
)

# Deploy the flow with the trigger
my_flow_deployment = deploy(
    flow=my_flow,
    name="my-flow-deployment",
    triggers=[trigger]
)

if __name__ == "__main__":
    my_flow_deployment()

Notes: - Replace

"external.resource.pinged"

and

"my.external.resource"

with the actual event and resource you want to use as triggers. - The

triggers

parameter in the

deploy

function accepts a list of trigger objects, allowing you to define multiple triggers if needed. You can find more details on defining triggers in the Prefect documentation on automations and triggers.

@Marvin the trigger is for flow run completed

To set up a deployment trigger that activates upon the completion of a flow run, you can use Prefect's event-driven deployment capabilities. Here's how you might configure it using the `DeploymentEventTrigger`:

python
from prefect import flow, deploy
from prefect.events import DeploymentEventTrigger

@flow
def first_flow():
    # Your first flow logic here
    pass

@flow
def second_flow():
    # Logic for the second flow
    pass

# Define a trigger for when the first flow run completes successfully
trigger = DeploymentEventTrigger(
    event="prefect.flow_run.completed",
    resource="first-flow-deployment"
)

# Deploy the second flow with the trigger
second_flow_deployment = deploy(
    flow=second_flow,
    name="second-flow-deployment",
    triggers=[trigger]
)

if __name__ == "__main__":
    # Deploy the first flow
    first_flow_deployment = deploy(
        flow=first_flow,
        name="first-flow-deployment"
    )
    first_flow_deployment()

Explanation: - The

DeploymentEventTrigger

is set to listen for the

prefect.flow_run.completed

event, which occurs when a flow run completes. - You specify the resource as the deployment name of the first flow,

"first-flow-deployment"

. - When the first flow completes, the trigger will activate the deployment of the second flow. This setup assumes that your Prefect environment is correctly configured to handle event-driven automations. For more details, refer to the Prefect documentation on automations and triggers.

from prefect import flow
from prefect.events import DeploymentCompoundTrigger


@flow(log_prints=True)
def decorated_fn(param_1: str):
    print(param_1)


if __name__=="__main__":
    decorated_fn.deploy(
        name="my-deployment",
        image="my-image-registry/my-image:my-tag",
        triggers=[
            DeploymentCompoundTrigger(
                enabled=True,
                name="my-compound-trigger",
                require="all",
                triggers=[
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my.external.resource"},
                      "expect": ["external.resource.pinged"],
                    },
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my.external.resource"},
                      "expect": ["external.resource.replied"],
                    },
                ],
                parameters={
                    "param_1": "{{ event }}",
                },
            )
        ],
        work_pool_name="my-work-pool",
    )

@Marvin look up documentation in https://docs.prefect.io/v3/automate/events/automations-triggers#define-triggers-in-serve-and-deploy

I cannot directly browse external websites, but I can guide you on how to access the documentation for defining triggers in Prefect. To find the relevant documentation for defining triggers in the

.deploy

function, you can visit the Prefect documentation site and navigate to the section on automations and triggers. Specifically, look for sections that discuss defining triggers in

.serve

and

.deploy

, which will provide detailed examples and explanations on how to configure your deployment with triggers. For direct access to the specific section you mentioned, you can use the following link: Define triggers in serve and deploy. This should provide you with the necessary information to understand and implement deployment triggers effectively in your Prefect workflows.

@Marvin

Using from prefect import flow
from prefect.events import DeploymentCompoundTrigger


@flow(log_prints=True)
def decorated_fn(param_1: str):
    print(param_1)


if __name__=="__main__":
    decorated_fn.deploy(
        name="my-deployment",
        image="my-image-registry/my-image:my-tag",
        triggers=[
            DeploymentCompoundTrigger(
                enabled=True,
                name="my-compound-trigger",
                require="all",
                triggers=[
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my.external.resource"},
                      "expect": ["external.resource.pinged"],
                    },
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my.external.resource"},
                      "expect": ["external.resource.replied"],
                    },
                ],
                parameters={
                    "param_1": "{{ event }}",
                },
            )
        ],
        work_pool_name="my-work-pool",
    )

what are the different options for type, match and expect parameters if I want a trigger when flow run complete

To configure a trigger for when a flow run completes, you can use the

DeploymentEventTrigger

with the appropriate event type and match parameters. Here's how you can set it up:

python
from prefect import flow
from prefect.events import DeploymentCompoundTrigger

@flow(log_prints=True)
def decorated_fn(param_1: str):
    print(param_1)

if __name__ == "__main__":
    decorated_fn.deploy(
        name="my-deployment",
        image="my-image-registry/my-image:my-tag",
        triggers=[
            DeploymentCompoundTrigger(
                enabled=True,
                name="my-compound-trigger",
                require="all",
                triggers=[
                    {
                      "type": "event",
                      "match": {"prefect.resource.id": "my-flow-run-id"},
                      "expect": ["prefect.flow-run.completed"],
                    }
                ],
                parameters={
                    "param_1": "{{ event }}",
                },
            )
        ],
        work_pool_name="my-work-pool",
    )

### Key Parameters: - type: Use

"event"

to specify that the trigger is based on an event. - match: Set this to match the specific flow run you are interested in, using

"prefect.resource.id": "my-flow-run-id"

. Replace

"my-flow-run-id"

with the actual ID of the flow run. - expect: Use

["prefect.flow-run.completed"]

to trigger when the flow run completes. This setup ensures that the deployment is triggered when the specified flow run completes successfully. Adjust the

match

parameter to target the specific flow run or resource you're interested in. For more detailed information, refer to the Prefect documentation on custom triggers.