Great Expectations：为EphemeralDataContext添加自定义Action

针对你遇到的EphemeralDataContext无法自动加载自定义ValidationAction的问题，可以通过以下两种方法解决：

方法一：手动注册自定义Action到DataContext的ActionRegistry #

EphemeralDataContext作为内存型上下文，不会自动扫描plugins目录并注册插件，你可以直接在创建上下文后手动注册Action类：

from great_expectations.checkpoint.actions import ValidationAction
from great_expectations.data_context import EphemeralDataContext
from great_expectations.data_context.types.base import DataContextConfig, InMemoryStoreBackendDefaults

# 定义自定义Action类
class StoreValidationResultsToDeltaAction(ValidationAction):
    def __init__(self, att):
        self.att = att

    def _run(self, validation_result_suite, *args, **kwargs):
        # 你的业务实现逻辑
        print(f"执行自定义Action，参数att：{self.att}")
        return {"success": True}

# 创建EphemeralDataContext
data_context_config = DataContextConfig(
    store_backend_defaults=InMemoryStoreBackendDefaults(),
)
gx_context = EphemeralDataContext(project_config=data_context_config)

# 手动注册自定义Action到上下文的Action注册表
gx_context.action_registry.register(
    "store_validation_results_to_delta",  # 自定义Action的标识名，用于Checkpoint配置引用
    StoreValidationResultsToDeltaAction
)

后续配置Checkpoint时，即可通过注册的标识名引用该Action：

from great_expectations.checkpoint import SimpleCheckpoint

checkpoint_config = {
    "name": "custom_delta_checkpoint",
    "config_version": 1.0,
    "class_name": "SimpleCheckpoint",
    "expectation_suite_name": "your_expectation_suite",
    "action_list": [
        {
            "name": "store_validation_results_to_delta",
            "action": {
                "class_name": "StoreValidationResultsToDeltaAction",
                "att": "你的参数值"
            }
        }
    ]
}

checkpoint = gx_context.add_checkpoint(**checkpoint_config)

方法二：规范插件目录结构并手动触发插件加载 #

如果仍想通过plugins目录管理插件，需要确保目录结构符合Great Expectations的插件规范，并手动触发加载：

1. 调整plugins目录结构：在你的plugins根目录下创建actions子目录，将自定义Action的.py文件放在该子目录中，例如：

   <你的plugins路径>/
     actions/
       delta_action.py  # 里面定义StoreValidationResultsToDeltaAction类
   

2. 创建上下文后显式调用插件加载方法：

from great_expectations.data_context import EphemeralDataContext
from great_expectations.data_context.types.base import DataContextConfig, InMemoryStoreBackendDefaults

data_context_config = DataContextConfig(
    store_backend_defaults=InMemoryStoreBackendDefaults(),
    plugins_directory="<path_to_plugins_folder>"
)
gx_context = EphemeralDataContext(project_config=data_context_config)

# 手动触发插件加载，扫描plugins目录并注册Action
gx_context.plugins_manager.load_plugins()

完成后，自定义Action会被自动注册到上下文，可直接在Checkpoint配置中引用。

内容的提问来源于stack exchange，提问作者Prassi