Trigger actions

Use Actions to notify the appropriate parties of the results of your Validation runs. Actions can be triggered on failure severity, Validation success, or all Validations. Validations are executed using Checkpoints, which each have a list of Actions that will be executed when each run has finished. By default, GX Cloud creates a Checkpoint for each Data Asset that you create. Optionally, you can also use a Checkpoint that you have created manually. This example will demonstrate how to create a SlackNotificationAction and append it to the list of Actions on a given Checkpoint.

Consider zero-code alerts instead

If you want to send notifications to email addresses, Slack, Microsoft Teams, PagerDuty, or ServiceNow, consider using zero-code alerts instead of coding Actions. While Actions can be scaled programmatically, alerts give you more control over which severities of failures trigger which notifications. Alerts also give you the ability to @mention collaborators in Slack.

Prerequisites

• A GX Cloud account.
• Your Cloud credentials saved in your environment variables.
• A Checkpoint (either an automatically created GX-managed Checkpoint or a manually created Checkpoint).
• Python version 3.10 to 3.13.
• An installation of the Great Expectations Python library.

Procedure

Instructions

1. Import the relevant modules and instantiate your Context.

   Python

   import great_expectations as gx
   from great_expectations.checkpoint import SlackNotificationAction
   
   context = gx.get_context()

2. Retrieve the Checkpoint to append the Action to.

   Python

   checkpoint_name = "my_checkpoint"
   checkpoint = context.checkpoints.get(checkpoint_name)

   The GX-managed Checkpoint name can be found through the UI

   For the Data Asset of interest, go to the Expectations tab. Then, click the code snippet icon next to the Validate button and click Generate snippet.

3. Define the Actions that the Checkpoint will trigger.

   The following is an example of how to define a SlackNotificationAction.

   Python

   action = SlackNotificationAction(
       name="send_slack_notification_on_failed_expectations",
       slack_token="${bot_user_oauth_token}",
       slack_channel="#my_channel",
       notify_on="failure",
       show_failed_expectations=True,
   )

   In this example, notify_on="failure" means that the Slack notification will be triggered when the Validation Results include any severity of Expectation failure. Accepted values for notify_on are as follows:

   • all: Always trigger the Action when Validation Results are received.
   • success: Trigger the Action only when all Expectations succeed.
   • failure: Trigger the Action when any Expectation fails, regardless of failure severity.
   • critical: Trigger the Action when there's a critical Expectation failure. This may be an Expectation configured with critical severity or an Expectation of any severity that failed to execute.
   • warning: Trigger the Action when there's a warning-level Expectation failure and no critical failures.
   • info: Trigger the Action when there's an info-level Expectation failure and no warning or critical failures.
   The highest severity takes precedence

   If a Validation Result includes a mix of warning and info failures, only Actions configured to notify on warning, failure, or all will be triggered. Any Actions configured to run on info will not be triggered.

4. Append the newly-created Action to the Checkpoint Action list and save the Checkpoint.

   Python

   checkpoint.actions.append(action)
   checkpoint.save()

5. Optional. Run a Validation to ensure the newly-created Action is triggered as expected.

Sample code

Python

# Instantiate the Context.
import great_expectations as gx
from great_expectations.checkpoint import SlackNotificationAction

context = gx.get_context()
set_up_context_for_example(context)

# Retrieve the Checkpoint.
checkpoint_name = "my_checkpoint"
checkpoint = context.checkpoints.get(checkpoint_name)

# Create a SlackNotificationAction for the Checkpoint to perform.
action = SlackNotificationAction(
    name="send_slack_notification_on_failed_expectations",
    slack_token="${bot_user_oauth_token}",
    slack_channel="#my_channel",
    notify_on="failure",
    show_failed_expectations=True,
)

# Append the Action to the Checkpoint and save it.
checkpoint.actions.append(action)
checkpoint.save()