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.
Prerequisites
- A GX Cloud account.
- Your Cloud credentials saved in your environment variables.
- A Checkpoint (either an automatically created GX-managed one or a manually created one).
- Python version 3.9 to 3.12.
- An installation of the Great Expectations Python library.
Procedure
- Instructions
- Sample code
-
Import the relevant modules and instantiate your Context.
Pythonimport great_expectations as gx
from great_expectations.checkpoint import SlackNotificationAction
context = gx.get_context() -
Retrieve the Checkpoint to append the Action to.
Pythoncheckpoint_name = "my_checkpoint"
checkpoint = context.checkpoints.get(checkpoint_name)The GX-managed Checkpoint name can be found through the UIFor the Data Asset of interest, go to the Validations tab. If you have more than one Expectation Suite, select the GX-managed one. Then, click the code snippet icon next to the Validate button and click Generate snippet.
-
Define the Actions that the Checkpoint will trigger.
The following is an example of how to define a
SlackNotificationAction
.Pythonaction = 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 fornotify_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 precedenceIf a Validation Result includes a mix of warning and info failures, only Actions configured to notify on
warning
,failure
, orall
will be triggered. Any Actions configured to run oninfo
will not be triggered. -
Append the newly-created Action to the Checkpoint Action list and save the Checkpoint.
Pythoncheckpoint.actions.append(action)
checkpoint.save() -
Optional. Run a Validation to ensure the newly-created Action is triggered as expected.
# Instantiate the Context.
import great_expectations as gx
from great_expectations.checkpoint import SlackNotificationAction
context = gx.get_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()