YarlisAISolutions/rules-cascade-python

PythonでのRule Cascade

パッケージ: pip install rule-cascade (Python 3.11以上; import 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")   # ルートデコレーターの下
@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）。状態を変更するすべてのエンドポイントは以下の4つの手順に従います:

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. コミット後に result["commands"] を実行 各 idempotencyKey につき1回（アウトボックステーブル、または 重複排除付きのタスクキューを使用）。

フレームワークの配置: 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スキーマ、 Djangoバリデーター、clean() メソッドとパーミッション、手書きの raise チェック、名前付きリミット（MAX_QTY = 10）、数値との比較、および match ブランチをリストアップします。drift がある候補は、プロジェクトの別の言語（多くの場合フロントエンド）が同じ名前を別の値で決定していることを意味します。1つのルールセットが両方を決定する必要があります。型と形状のバリデーション（Pydanticモデル）はそのままにし、ビジネス上の決定をルールに移します。同じ入力に対して古いチェックとエンジンの両方を実行するパリティテストを追加し、両者が一致した場合にのみ古いチェックを削除します。

テスト

decision と finding の code をアサートし、メッセージテキストはアサートしないでください。ルール自体はゴールデンテスト（rcas check）によってテストされます。CIでは pytest の前に rcas check と rcas compile --all を実行してください。

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