Create a Checkpoint with Actions
A Checkpoint executes one or more Validation Definitions and then performs a set of Actions based on the Validation Results each Validation Definition returns.
Prerequisites
- Python version 3.10 to 3.13.
- An installation of GX Core.
- A preconfigured Data Context. In this guide the variable
contextis assumed to contain your Data Context. - A Validation Definition.
Procedure
- Instructions
- Sample code
-
Retrieve the Validation Definitions the Checkpoint will run.
In the following example your Data Context has already been stored in the variable
contextand an existing Validation Definition is retrieved from it:Pythonvalidation_definitions = [
context.validation_definitions.get("my_validation_definition")
] -
Determine the Actions that the Checkpoint will automate.
After a Checkpoint receives Validation Results from running a Validation Definition, it executes a list of Actions. The returned Validation Results determine what task is performed for each Action. Actions can include updating Data Docs with the new Validation Results, sending different notifications for different failure severities, or your own custom logic. The Actions list is executed once for each Validation Definition in a Checkpoint.
Actions can be found in the
great_expectations.checkpointmodule. All Action class names end with*Action.The following is an example of how to create an Action list that will trigger a
SlackNotificationActionand anUpdateDataDocsAction:Pythonaction_list = [
# This Action sends a Slack Notification if an Expectation fails.
SlackNotificationAction(
name="send_slack_notification_on_failed_expectations",
slack_token="${validation_notification_slack_webhook}",
slack_channel="${validation_notification_slack_channel}",
notify_on="failure",
show_failed_expectations=True,
),
# This Action updates the Data Docs static website with the Validation
# Results after the Checkpoint is run.
UpdateDataDocsAction(
name="update_all_data_docs",
),
]In the above example, string substitution is used to pull the value of
slack_tokenfrom an environment variable. For more information on securely storing credentials and access tokens see Configure credentials.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_onare 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, orallwill be triggered. Any Actions configured to run oninfowill not be triggered. -
Optional. Choose the Result Format
When a Checkpoint is created you can adjust the verbosity of the Validation Results it generates by setting a Result Format. A Checkpoint's Result Format will be applied to all Validation Results in the Checkpoint every time they are run. By default, a Checkpoint uses a
SUMMARYresult format: it indicates the success or failure of each Expectation in a Validation Definition, along with a partial set of the observed values and metrics that indicate why the Expectation succeeded or failed.For more information on configuring a Result Format, see Choose a Result Format.
-
Create the Checkpoint.
The Checkpoint class is available from the
great_expectationsmodule. You instantiate a Checkpoint by providing the lists of Validation Definitions and Actions that you previously created, as well as a unique name for the Checkpoint, to the Checkpoint class. The Checkpoint's Result Format can optionally be set, as well:Pythoncheckpoint_name = "my_checkpoint"
checkpoint = gx.Checkpoint(
name=checkpoint_name,
validation_definitions=validation_definitions,
actions=action_list,
result_format={"result_format": "COMPLETE"},
) -
Add the Checkpoint to your Data Context.
Once you create a Checkpoint you should save it to your Data Context for future use:
Pythoncontext.checkpoints.add(checkpoint)When you add your Checkpoint to your Data Context you will be able to retrieve it elsewhere in your code by replacing the value for
checkpoint_nameand executing:Pythoncheckpoint_name = "my_checkpoint"
checkpoint = context.checkpoints.get(checkpoint_name)With a File Data Context you will also be able to retrieve your Checkpoint in future Python sessions.
import great_expectations as gx
from great_expectations.checkpoint import (
SlackNotificationAction,
UpdateDataDocsAction,
)
context = gx.get_context()
# Create a list of one or more Validation Definitions for the Checkpoint to run
validation_definitions = [
context.validation_definitions.get("my_validation_definition")
]
# Create a list of Actions for the Checkpoint to perform
action_list = [
# This Action sends a Slack Notification if an Expectation fails.
SlackNotificationAction(
name="send_slack_notification_on_failed_expectations",
slack_token="${validation_notification_slack_webhook}",
slack_channel="${validation_notification_slack_channel}",
notify_on="failure",
show_failed_expectations=True,
),
# This Action updates the Data Docs static website with the Validation
# Results after the Checkpoint is run.
UpdateDataDocsAction(
name="update_all_data_docs",
),
]
# Create the Checkpoint
checkpoint_name = "my_checkpoint"
checkpoint = gx.Checkpoint(
name=checkpoint_name,
validation_definitions=validation_definitions,
actions=action_list,
result_format={"result_format": "COMPLETE"},
)
# Save the Checkpoint to the Data Context
context.checkpoints.add(checkpoint)
# Retrieve the Checkpoint later
checkpoint_name = "my_checkpoint"
checkpoint = context.checkpoints.get(checkpoint_name)