Skip to main content
Version: 1.3.12

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

  1. Create a new custom Action class that inherits the ValidationAction class.

    Python
    class MyCustomAction(ValidationAction):
  2. Set a unique name for type.

    Python
    type: Literal["my_custom_action"] = "my_custom_action"
  3. 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 custom run method through self.<MY_FIELD_NAME>.

Python
my_custom_str_field: str
  1. 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.
    ...

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.