Version: 1.3.0

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.
Python
```
class MyCustomAction(ValidationAction):
```

Set a unique name for type.

Python
type: Literal["my_custom_action"] = "my_custom_action"

Override the run() method with the logic for the Action.

Python
@override
def run(
    self,
    checkpoint_result: CheckpointResult,
    action_context: ActionContext,  # Contains results from prior Actions in the same Checkpoint run.
) -> dict:
    # Domain-specific logic
    self._do_my_custom_action(checkpoint_result)
    # Return information about the Action
    return {"some": "info"}

def _do_my_custom_action(self, checkpoint_result: CheckpointResult):
    # Perform custom logic based on the validation results.
    ...

Python
from typing import Literal

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. Override the `run()` method to perform the desired task.
    @override
    def run(
        self,
        checkpoint_result: CheckpointResult,
        action_context: ActionContext,  # Contains results from prior Actions in the same Checkpoint run.
    ) -> dict:
        # Domain-specific logic
        self._do_my_custom_action(checkpoint_result)
        # Return information about the Action
        return {"some": "info"}

    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.

Prerequisites​

Procedure​

Prerequisites

Procedure