Create a custom Action
Great Expectations provides Actions for common workflows such as sending emails and updating Data Docs. If these don't meet your needs, you can create a custom Action to integrate with different tools or apply custom business logic based on Validation Results. Example use cases for custom Actions include:
- Opening tickets in an issue tracker when Validation runs fail.
- Triggering different webhooks depending on which Expectations fail.
- Running follow-up ETL jobs to fill in missing values.
A custom Action can do anything that can be done with Python code.
To create a custom Action, you subclass the ValidationAction
class, overriding the type
attribute with a unique name and the run()
method with custom logic.
Prerequisites
Procedure
- Instructions
- Sample code
-
Create a new custom Action class that inherits the
ValidationAction
class.Pythonclass MyCustomAction(ValidationAction):
-
Set a unique name for
type
.Pythontype: Literal["my_custom_action"] = "my_custom_action"
-
Optional. Add any additional fields your Action requires at runtime. Actions are built on Pydantic models. Define the field name as a class-level attribute on your Action, and annotate it with the correct type. When you instantiate the Action, pass the field value into the Action
init
method. Your Action will have access to these values within your customrun
method throughself.<MY_FIELD_NAME>
.
my_custom_str_field: str
-
Override the
run()
method with the logic for the Action.Python@override
def run(
self,
checkpoint_result: CheckpointResult,
action_context: Union[
ActionContext, None
], # Contains results from prior Actions in the same Checkpoint run.
) -> dict:
# Domain-specific logic
self._do_my_custom_action(checkpoint_result)
# Optional. Access custom fields you provide the Action at runtime.
extra_context = self.my_custom_str_field
# Return information about the Action
return {"some": "info", "extra_context": extra_context}
def _do_my_custom_action(self, checkpoint_result: CheckpointResult):
# Perform custom logic based on the validation results.
...
from typing import Literal, Union
from typing_extensions import override
from great_expectations.checkpoint import (
ActionContext,
CheckpointResult,
ValidationAction,
)
# 1. Extend the `ValidationAction` class.
class MyCustomAction(ValidationAction):
# 2. Set the `type` attribute to a unique string that identifies the Action.
type: Literal["my_custom_action"] = "my_custom_action"
# 3. Optional. Add any additional fields your Action requires at runtime.
my_custom_str_field: str
# 4. Override the `run()` method to perform the desired task.
@override
def run(
self,
checkpoint_result: CheckpointResult,
action_context: Union[
ActionContext, None
], # Contains results from prior Actions in the same Checkpoint run.
) -> dict:
# Domain-specific logic
self._do_my_custom_action(checkpoint_result)
# Optional. Access custom fields you provide the Action at runtime.
extra_context = self.my_custom_str_field
# Return information about the Action
return {"some": "info", "extra_context": extra_context}
def _do_my_custom_action(self, checkpoint_result: CheckpointResult):
# Perform custom logic based on the validation results.
...
Now you can use your custom Action like you would any built-in Action. Create a Checkpoint with Actions to start automating responses to Validation Results.