Python
The rule_cascade package. The Python runtime and the reference implementation every other runtime is compared with.
rule_cascade evaluates bundles and compiles source rulesets. It is also the reference
implementation: the other runtimes are ports of it and must agree with it on every case of the
conformance suite. Its only dependency is jsonschema. It reads no YAML: it takes parsed documents.
Supported versions: Python 3.10 or later (requires-python in pyproject.toml). CI tests 3.10
on Linux and 3.12 on Linux, Windows and macOS. No maximum is declared.
Install from the repository
git clone https://github.com/YarlisAISolutions/rule-cascade.git
pip install ./rule-cascade/packages/pythonWithout installing, put rule-cascade/packages/python/src on PYTHONPATH; that is what the
Makefile, CI and tools/rulecheck.py do.
Load
import json
from pathlib import Path
from rule_cascade import RuleSet
rules = RuleSet.from_bundle(json.loads(Path("acme.payments.transfer.bundle.json").read_text(encoding="utf-8")))
print(rules.checksum)
# sha256:c192dd53b5b1d307d52ccbc27fc1674114e8714d53b699b24088a648ae242c7eTo compile source documents, load(document, registry, loader) runs every check of specification
section 5. registry maps ruleset ids to parsed documents; loader(file) returns the document
behind an entity's $ref, or None.
from rule_cascade import load
rules = load(registry["acme.payments.transfer"], registry, schemas.get)Evaluate
result = rules.evaluate({
"entity": "Transfer",
"operation": "create",
"data": {"type": "international", "amount": 12000, "currency": "USD",
"beneficiary": {"name": "Ana", "country": "ES"}},
"actor": {"id": "u-1", "roles": ["teller"]},
})
print(result["decision"])
for finding in result["findings"]:
print(finding["code"], finding["severity"], finding["blocking"], finding["fields"])
# deny
# ORG-TRF-003 warning False ['/memo']
# PAY-TRF-002 error True ['/beneficiary/swiftCode']
# PAY-TRF-003 warning True ['/amount', '/beneficiary/name']The signature is evaluate(request, channel="server", operators=None). operators maps custom
operator names to functions, for example {"x-luhn": luhn}. The result is a dict with
ruleset, version, checksum, decision, findings, effects and commands.
rule_cascade.evaluate(manifest, request, operators=None) evaluates a manifest received from
elsewhere, for example one fetched from a rule server.
Errors
| What | How it surfaces | What to do |
|---|---|---|
| An unusable bundle | LoadError from RuleSet.from_bundle with BUNDLE_UNSUPPORTED or BUNDLE_INVALID | Do not start |
| A source ruleset that fails a check | LoadError from load; .problems is a list of {code, message, rule?}, .codes the codes | Do not start; fix it in CI |
| A request of the wrong shape | ValueError (the subclass RequestError), for example 'data' must be an object | Answer 400 |
| A rule that cannot be evaluated, or a missing operator | No exception: a blocking RULE-EVALUATION-ERROR finding | Alert on it |
from rule_cascade import LoadError, load
try:
load(broken, registry, schemas.get)
except LoadError as err:
print(err.codes)
# ['PARAM_LOOSENED']More
The package README has the runnable versions of these snippets
(against the conformance fixtures), manifests and channels, custom operators, expressions and the
engine protocol (python -m rule_cascade engine). The command-line front end of this package is
tools/rulecheck.py: check, compile, manifest, derive, jsonlogic, conformance, sync.