Creating Custom Checkpoint Actions in GX Core

GX Core Support
1

Hey,

Prior GX versions, 0.18.21 and earlier we were able to create custom checkpoint actions Action | Great Expectations as long as it conformed to the ValidationAction API it worked.

We had a custom action plugin working and we are now trying to port it to GX 1.x however it looks like ValidationActions now use pydantic to validate Checkpoint Actions via “CheckpointAction” which does not allow us to extend it with a custom ValidationAction.

CheckpointAction = Annotated[
    Union[
        EmailAction,
        MicrosoftTeamsNotificationAction,
        OpsgenieAlertAction,
        PagerdutyAlertAction,
        SlackNotificationAction,
        SNSNotificationAction,
        UpdateDataDocsAction,
    ],
    Field(discriminator="type"),
]

So when we try to call our plugin it throws a validation error, rightly so,

pydantic.v1.error_wrappers.ValidationError: 1 validation error for Checkpoint
actions -> 0
  No match for discriminator 'type' and value 'my_custom_action' (allowed values: 'email', 'microsoft', 'opsgenie', 'pagerduty', 'slack', 'sns', 'update_data_docs') (type=value_error.discriminated_union.invalid_discriminator; discriminator_key=type; discriminator_value=my_custom_action; allowed_values='email', 'microsoft', 'opsgenie', 'pagerduty', 'slack', 'sns', 'update_data_docs')

Is there a custom action implementation guide or suggestions? As we tried monkey patching and subclassing but seems like “CheckpointAction” is used in many places.

New documentation suggests no way to customise actions now.

1 Like
2

hey @schen welcome to our community and thanks for reaching out. I’m going to speak to a few folks about this and will come back with an update

1 Like
3

just following up here! thank you for highlighting this issue.

You can use your previous actions by redefining them with our new Pydantic syntax and updated APIs. Unfortunately, we don’t have specific documentation for this at the moment. I escalated the matter today, and it will be prioritized for resolution shortly.

2 Likes
4

Hey @adeola thanks for following up.

It looks like defining with the new Pydantic syntax wasn’t enough, but we managed to get it working by subclassing the Checkpoint class and overwriting the action_list to accept our new ValidationAction class as a valid CheckpointAction which pydantic was blocking us on.

We then monkey patched the Checkpoint class in a few areas to stop pydantic complaining.

We’ll look to fix the approach once the proper documentation/solution for this comes out. Thanks for the help.

2 Likes
5

@schen Is there any chance you could post example code for how you got this working?

I’d like to play around with building custom checkpoint actions as well and having an example would be helpful.

1 Like
6

That’s great to hear, and I’m glad you were able to get unblocked! I’m sorry I couldn’t be of more help, but I did strongly advocate for the release of relevant documentation, which should be released soon. I’ll make sure to follow up with you once it’s available.

In the meantime, I’d love to see how your solution worked. If it’s not too much trouble to share an example of how you got it working

7

Hey @adeola / @ToivoMattila , I’ve attached a scrubbed sample code block

This is more a hack rather than a solution to avoid all the pydantic errors, so hopefully a proper solution is released soon.

from typing import Literal, Union, Annotated, List, get_args
from typing_extensions import override
from great_expectations.checkpoint.actions import (
    ActionContext,
    CheckpointAction,
    ValidationAction,
)
from great_expectations.compatibility.pydantic import Field
from great_expectations.checkpoint.checkpoint import CheckpointResult


class ExampleCustomAction(ValidationAction):
    type: Literal["mycustomaction"] = "mycustomaction"  # Pydantic model

    @override
    def run(
        self, checkpoint_result: CheckpointResult, action_context: ActionContext | None = None
    ) -> dict:
        print('Do your custom action here')

# Monkey Patch all relevant Checkpoint actions
CustomCheckpointAction = Annotated[
    Union[
        *get_args(get_args(CheckpointAction)[0]),  # Ensure we carry over original actions
        ExampleCustomAction,
    ],
    Field(discriminator="type"),
]

# Monkey Patch Checkpoint to include the new action
from great_expectations import Checkpoint
class CustomCheckpoint(Checkpoint):
    actions: List[CustomCheckpointAction] = Field(default_factory=list)


Checkpoint = CustomCheckpoint

# Monkey Patch CheckpointStore to include the new deserialization method
from great_expectations.data_context.store import CheckpointStore
def deserialize_override(self, value):
    if self.cloud_mode:
        return CustomCheckpoint.parse_obj(value)
    return CustomCheckpoint.parse_raw(value)


CheckpointStore.deserialize = deserialize_override
2 Likes
8

Thank you so much for providing this! As promised earlier, I’ll be sure to follow up about the status of the proper documentation

9

Awesome!

Thank you very much!