Community编程与开发github.com

Python 规则级联(Rule Cascade)

在 Python 服务与任务(FastAPI、Django、Flask、Celery、批处理脚本)中使用 rule-cascade 包强制执行规则级联。适用于将规则接入 Python 代码路径,取代 Pydantic 校验器或手写检查,以及批量评估大量记录。

Python 规则级联(Rule Cascade) 是什么?

本技能指导开发者在 Python 项目中使用 `rule-cascade` 包集成规则级联引擎。它支持 FastAPI、Django、Flask、Celery 及批处理脚本等常见框架,并提供装饰器、中间件或显式调用的方式来评估规则。该方案强调启动时一次性加载从 YAML 编译好的规则包(bundle),避免运行时解析 YAML 或复制规则到代码中。对于每条请求,引擎会构建包含实体、操作、数据和角色等信息的请求对象,执行评估后返回 `allow`/`deny` 及发现项、计算值、命令等结果。拒绝时会返回符合 RFC 9457 的 422 错误详情。此外,该技能还介绍了批量处理多条记录的优化方法、如何用 `rcas analyze` 自动识别可替换的手工校验逻辑,以及使用黄金测试和 CI 流程确保规则质量。

兼容平台~Claude Code~Codex CLI~Cursor
npx skills add https://github.com/YarlisAISolutions/rule-cascade-skills/tree/main/plugins/rules-cascade/skills/rules-cascade-python

Installed? Explore more 编程与开发 skills: steipete/bluebubbles, steipete/eightctl, steipete/blucli · View all 6 →

在你喜欢的 AI 中提问

打开一个已预加载此 Agent Skill 的新对话。

文档

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。每个改变状态的操作端点遵循以下四个步骤:

  1. 在服务器端评估,使用 {"entity", "operation", "data", "original", "actor", "resolutions"};actor 来自身份认证({"id", "roles"}),绝不应来自调用者发送的字段。
  2. 拒绝时返回 422,详细说明问题(application/problem+json,type: "urn:rule-cascade:rule-violation"),包含评估结果,以便客户端能将 findings[].fields 映射到输入。
  3. 持久化,在应用 compute 效果(effect["type"] == "value")之后:存储服务器端计算的值。
  4. 在提交后,每个 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。

参考文档:https://rulescascade.com/usage/python/

Individual skills in this repo

This repo contains 2 individual skills — each has its own dedicated page.

相关技能