Python 中的 Rule Cascade
套件:pip install rule-cascade(Python 3.11+;匯入 rule_cascade)。
束由 rcas compile --all 產生,置於 build/rules/。
絕不要在執行時期評估 YAML,也絕不要將規則複製到程式碼中。
載入一次,每個請求評估一次
import json
from rule_cascade import RuleSet, RequestError
# 在啟動時(模組層級、FastAPI lifespan、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。它會對格式錯誤的請求拋出 RequestError(回應 400)。每個會改變狀態的端點都應遵循以下四個步驟:
- 在伺服器端評估,使用
{"entity", "operation", "data", "original", "actor", "resolutions"};actor來自驗證機制({"id", "roles"}),絕不來自呼叫端傳送的欄位。 - 拒絕時回應 422 problem details(
application/problem+json,type: "urn:rule-cascade:rule-violation"),其中包含結果,以便客戶端將findings[].fields對應到輸入。 - 持續化,在套用
compute效果(effect["type"] == "value")之後:儲存伺服器端的值。 - 在提交後執行
result["commands"],每個idempotencyKey執行一次(使用 outbox 表格,或帶有去重機制的任務佇列)。
框架整合方式:FastAPI 使用依賴注入(Depends(get_rules))和將 RuleViolation 轉換為 422 的例外處理;Django 使用由每個檢視或實體序列化器 validate() 呼叫的服務函式,以及中介軟體或 DRF 例外處理來處理 422;Flask 則是應用程式工廠中建立的擴充物件。Celery 和批次 job 在每次寫入前以相同方式評估。
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 和 finding 的 code,而非訊息文字。規則本身由它們的 golden test 測試(rcas check)。在 CI 中,在 pytest 之前執行 rcas check 和 rcas compile --all。