Python 中的规则级联
包:pip install rule-cascade(Python 3.11+;导入 rule_cascade)。规则包由 rcas compile --all 编译到 build/rules/ 目录。切勿在运行时评估 YAML,也切勿将规则复制到代码中。
一次性加载,每次请求评估
import json
from rule_cascade import RuleSet, RequestError
# 在启动时(模块级别、FastAPI 生命周期、Django AppConfig.ready)。错误的规则包会引发异常:不要提供服务。
with open("build/rules/<ruleset-id>.bundle.json", encoding="utf-8") as f:
rules = RuleSet.from_bundle(json.load(f))
优先使用装饰器(自 1.0.0-alpha.7 起):它会构建请求,在函数执行前强制执行,并抛出 RuleViolation,其 problem() 就是下面的 422 响应。
from rule_cascade import enforce_rules, current_evaluation
from rule_cascade.fastapi import install # 或者 rule_cascade.flask.init_app(app),或 Django 中间件
install(app) # RuleViolation -> 422, RequestError -> 400
@app.post("/orders", status_code=201)
@enforce_rules(rules, entity="Order", operation="create", data="order", actor="user") # 位于路由装饰器之下
def create_order(order: Order, user: User = Depends(current_user)):
evaluation = current_evaluation() # 允许的结果:警告、计算值、命令
rules.enforce(request, "server") 无需装饰器即可完成相同操作。底层 rules.evaluate(request, "server") 返回一个字典:decision(allow 或 deny)、findings、effects、commands。对于格式错误的请求(返回 400),它会抛出 RequestError。每个改变状态的操作端点遵循以下四个步骤:
- 在服务器端评估,使用
{"entity", "operation", "data", "original", "actor", "resolutions"};actor来自身份认证({"id", "roles"}),绝不应来自调用者发送的字段。 - 拒绝时返回 422,详细说明问题(
application/problem+json,type: "urn:rule-cascade:rule-violation"),包含评估结果,以便客户端能将findings[].fields映射到输入。 - 持久化,在应用
compute效果(effect["type"] == "value")之后:存储服务器端计算的值。 - 在提交后,每个
idempotencyKey(使用 outbox 表,或具有去重功能的任务队列)执行result["commands"]。
框架集成方式:FastAPI 使用依赖项(Depends(get_rules))和将 RuleViolation 转换为 422 的异常处理器;Django 使用每个视图或序列化器 validate() 方法调用的服务函数,外加将 422 转换为响应的中间件或 DRF 异常处理器;Flask 使用在应用工厂中创建的扩展对象。Celery 和批处理作业在每次写入前以相同方式评估。
RuleSetHolder(load_bundle, interval=300)(或 cron="0 * * * *")后台重新加载规则包,并在重新加载失败时保留上一次正确的规则。
处理大量记录
对于文件和队列,可以启动一次 rcas engine,然后通过它流式传输 JSON Lines 请求,或循环调用 rules.evaluate:两种方式都只需加载一次规则包。参见 https://rulescascade.com/examples/batch/。
替换现有校验
rcas analyze --module <module> 可以列出 Pydantic Field(le=...) 和校验器、marshmallow schema、Django 校验器、clean() 方法和权限、手写的 raise 检查、命名限制(MAX_QTY = 10)、数字比较以及 match 分支。带有 drift 标记的候选意味着同一名称在项目的另一种语言(通常是前端)中有不同取值:应由一个规则集为两者都做决定。保留类型和形状校验(如 Pydantic 模型)在原处;将业务决策移至规则中。添加一次回归测试,让旧校验和新引擎处理相同的输入,只有当两者结果一致时才删除旧校验。
测试
断言 decision 和 findings 的 code,而不是消息文本。规则本身由黄金测试(rcas check)验证。在 CI 中,在 pytest 之前运行 rcas check 和 rcas compile --all。