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.9 to 3.11.
- An installation of GX Core.
- A preconfigured Data Context. In this guide the variable
context
is 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
context
and 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 or sending alerts when validations fail. The Actions list is executed once for each Validation Definition in a Checkpoint.
Actions can be found in the
great_expectations.checkpoint
module. All Action class names end with*Action
.The following is an example of how to create an Action list that will trigger a
SlackAlertAction
and anUpdateDataDocsAction
:Pythonaction_list = [
# This Action sends a Slack Notification if an Expectation fails.
gx.checkpoint.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.
gx.checkpoint.UpdateDataDocsAction(
name="update_all_data_docs",
),
]In the above example, string substitution is used to pull the value of
slack_token
from an environment variable. For more information on securely storing credentials and access tokens see Configure credentials.If the list of Actions for a Checkpoint is empty, the Checkpoint can still run. Its Validation Results are saved to the Data Context, but no tasks are executed.
-
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
SUMMARY
result 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_expectations
module. 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_name
and executing:Pythoncheckpoint_name = "my_checkpoint"
checkpoint = context.checkpoints.get(checkpoint_name)With a File Data Context or a GX Cloud Data Context you will also be able to retrieve your Checkpoint in future Python sessions.
import great_expectations as gx
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.
gx.checkpoint.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.
gx.checkpoint.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)