声明式规则引擎 —— 在运行时执行表单校验 (Validation Rules)

MetaForm 低代码引擎系列 · 第 4 篇
技术栈：Python asteval + AST 沙箱

一、为什么需要规则引擎

在传统的后端开发中，校验逻辑通常硬编码在接口中：

@router.post("/api/data/survey")
def submit_survey(payload: dict):
    if payload.get("score") < 0 or payload.get("score") > 100:
        raise HTTPException(400, "问卷分数必须在0到100之间！")
    if payload.get("type") == "VIP" and not payload.get("vip_code"):
        raise HTTPException(400, "VIP问卷必须填写邀请码！")

这种写法的致命伤：每新增一个表单、每修改一个校验规则，都需要重写代码、重跑测试、重新发版。 这违背了低代码平台"元数据驱动、动态生效"的最高原则。

真正的 SaaS 化，要求这些 if/else 业务规则必须"降维"成为存储在元数据表中的普通行记录，动态生效，无需编译部署。

二、设计 meta_validation_rules 表

核心结构极其简单——存储"出错条件"的公式和提示信息：

CREATE TABLE meta_validation_rules (
    rule_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    form_id VARCHAR(64) NOT NULL,
    error_condition_formula TEXT NOT NULL,  -- 出错公式，如: "Score < 0 or Score > 100"
    error_message TEXT NOT NULL,            -- 抛给前端的错误提示
    error_display_field VARCHAR(64),        -- 【精细化设计】错误定位：为空表示全局报错，有值则前端对应字段标红
    is_active BOOLEAN DEFAULT TRUE
);

之前硬编码的校验逻辑，变成了两行数据库记录：

三、安全执行沙箱 (Sandbox)

要让字符串 Score < 0 真正生效，需要在第 3 篇的 DML 拦截层中植入一个安全的表达式求值沙箱 (AST Evaluator Sandbox)。

Python 实现：asteval 库

eval() 虽然方便，但使用原生 eval 无异于引火烧身——黑客会利用它执行 __import__('os').system('rm -rf /') 把服务器格式化。

我们使用工业级安全替代方案 asteval，它在限制所有危险内部调用的前提下完美解析公式：

from asteval import Interpreter
from fastapi import HTTPException

def execute_validation_rules(form_id: str, record_payload: dict, db):
    """在 DML 写入前执行所有激活的校验规则"""

    # 1. 查出当前表单激活的所有规则
    rules = db.execute(
        """SELECT error_condition_formula, error_message
           FROM meta_validation_rules
           WHERE form_id = :fid AND is_active = true""",
        {"fid": form_id}
    ).fetchall()

    if not rules:
        return  # 无规则直接放行

    # 2. 初始化安全沙箱
    sandbox = Interpreter(use_numpy=False, builtins_readonly=True)

    # 3. 将完整的表单数据注入沙箱上下文
    # 架构师提示：注入整个 record_payload 可以实现类似于 Start_Date > End_Date 的跨字段联动校验
    for key, value in record_payload.items():
        sandbox.symtable[key] = value

    # 4. 循环评估每条规则
    for rule in rules:
        # 沙箱执行字符串公式，返回 True 或 False
        # 例如 record_payload={"Score": 120}，评估 "120 < 0 or 120 > 100" → True
        is_error = sandbox(rule.error_condition_formula)

        if is_error:
            # 公式成立 = 满足出错条件，阻断写入！
            raise HTTPException(status_code=400, detail=rule.error_message)

四、事务阻断机制

在第 3 篇的 DML 写入流程中，注入 Rule Engine 拦截器。将上面的 execute_validation_rules 嵌入到写入接口中：

@router.post("/api/data/{form_id}")
def insert_record(form_id: str, payload: dict, db: Session = Depends(get_db)):
    # ... 加载元数据、Canonical 编码（第 3 篇内容）...
    canonical_payload = canonical_encode_all(payload, fields_meta)

    # 🛡️ 注入规则引擎拦截器
    execute_validation_rules(form_id, canonical_payload, db)

    # 校验通过，继续落库
    db.execute(
        "INSERT INTO data_heap (id, org_id, form_id, payload) VALUES (...)",
        {...}
    )
    db.commit()
    return {"status": "ok"}

五、Validation Rule 拦截链图解

整个校验流程：

1. 用户点击保存，携带完整的 {Score: -5, Start_Date: "2024-01-01", End_Date: "2023-01-01"} 进入 Save Transaction
2. 引擎拦截，从 meta_validation_rules 取出公式 Score < 0 或 Start_Date > End_Date

3. AST 沙箱评估：

   • ⚠️ 架构师提示（关于上下文注入）：喂入沙箱的绝对不只是当前触发的单个字段，而是完整的 Payload 上下文。这使得平台天然具备了类似 Salesforce Validation Rules 的强大跨字段联动校验能力。

4. 结果分发：

   • True（触发错误）→ 根据 error_display_field 决定抛出全局错误还是前端精准标红特定字段，事务 Rollback
   • False（校验通过）→ 继续执行 INSERT to PostgreSQL

小结

这套机制的革命性在于：产品经理可以直接在后台管理界面输入 Age < 18 并写上"未成年人不得参与"，整个平台的该表单接口从这一刻起立即获得校验防线——无需编译，无需部署，立即生效。

下一篇预告：如果不只是在数据存入前做拦截，还想在落库后自动发邮件、推送消息、调用 Webhook 该怎么做？