Python中如何将YAML中的Lambda字符串转为可调用对象传入Deequ校验方法

我在用Deequ做数据校验，为提升易用性，编写了遍历YAML配置自动添加校验的代码：

for exp in checks[table]:
    params = checks[table][exp]
    validate = getattr(check, exp)
    result = result.addCheck(
        validate(*params)
    )

对应的YAML配置如下：

checks = """
table:
  hasSize:
   - "lambda x: x < 55000"
  isUnique: 
   - "customer"
"""

但YAML中的"lambda x: x < 55000"会被识别为字符串，调用校验时触发错误：

Can't execute the assertion: An exception was raised by the Python Proxy. Return Message: Traceback (most recent call last):   File "/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 2442, in _call_proxy     return_value = getattr(self.pool[obj_id], method)(*params)   File "/databricks/python/lib/python3.8/site-packages/pydeequ/scala_utils.py", line 37, in apply     return self.lambda_function(arg) TypeError: 'str' object is not callable !

需要将该Lambda字符串转换为可调用对象后传入校验函数，以下是几种可行方案：

解决方案 #

方法1：使用eval直接转换（简单但需注意安全） #

eval可直接将字符串形式的Lambda转为可调用对象，修改循环代码添加判断逻辑：

import re

# 匹配Lambda格式字符串的正则
lambda_pattern = re.compile(r'^lambda\s+.+:')

for exp in checks[table]:
    params = checks[table][exp]
    processed_params = []
    for param in params:
        # 判断是否为Lambda字符串
        if isinstance(param, str) and lambda_pattern.match(param):
            processed_params.append(eval(param))
        else:
            processed_params.append(param)
    
    validate = getattr(check, exp)
    result = result.addCheck(
        validate(*processed_params)
    )

注意：eval存在安全风险，若YAML配置来自不可信渠道，请勿使用此方法，防止恶意代码注入。

方法2：用ast模块安全解析Lambda表达式 #

ast模块可先验证表达式合法性再解析，比直接用eval更安全：

import ast

for exp in checks[table]:
    params = checks[table][exp]
    processed_params = []
    for param in params:
        if isinstance(param, str) and param.strip().startswith('lambda'):
            # 解析表达式
            expr = ast.parse(param, mode='eval')
            # 确保解析结果是Lambda表达式
            if isinstance(expr.body, ast.Lambda):
                processed_params.append(eval(compile(expr, '<string>', 'eval')))
            else:
                processed_params.append(param)
        else:
            processed_params.append(param)
    
    validate = getattr(check, exp)
    result = result.addCheck(
        validate(*processed_params)
    )

方法3：自定义YAML标记区分Lambda参数 #

在YAML中给Lambda参数添加自定义标记，解析时自动转换：
修改后的YAML配置：

checks = """
table:
  hasSize:
   - !lambda "x: x < 55000"
  isUnique: 
   - "customer"
"""

解析YAML时注册自定义构造器：

import yaml

# 自定义Lambda构造器
def lambda_constructor(loader, node):
    lambda_body = loader.construct_scalar(node)
    return eval(f'lambda {lambda_body}')

# 注册构造器
yaml.add_constructor('!lambda', lambda_constructor)

# 解析YAML（checks_yaml_str为原始YAML字符串）
checks = yaml.load(checks_yaml_str, Loader=yaml.FullLoader)

之后循环代码无需额外处理，可直接使用参数。

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