rule-cascade (Python)

The Rule Cascade runtime for Python, and the reference implementation of the specification. It implements both conformance levels: it reads a bundle and evaluates (evaluator), and it loads source documents and produces bundles (compiler).

Requires Python 3.10 or later. The only dependency is jsonschema, which validates source documents against the ruleset schema.

pip install rule-cascade               # add the extra "rule-cascade[yaml]" to read YAML sources

The command-line tool for every language is rcas.

Module	What it is
rule_cascade.expressions	Expressions, operators and portable patterns: specification section 4
rule_cascade.ruleset	Loading, inheritance, load-time checks, checksum, manifests, bundles: sections 5 to 7
rule_cascade.evaluate	Evaluation of a request against a manifest: section 8
rule_cascade.values	Numbers, equality, canonical JSON and rendering, shared by the others
rule_cascade.engine	The engine protocol: section 13

The snippets below run from the repository root. They read the JSON fixtures of the conformance suite, which are the example rulesets in examples/contracts and examples/catalog converted to JSON.

Compile and evaluate

load(document, registry, loader) runs every step of specification section 5 and returns a RuleSet, or raises LoadError. It never returns a ruleset that failed a check.

import json
from pathlib import Path

from rule_cascade import LoadError, load

fixtures = Path("conformance/fixtures")


def read(name):
    return json.loads((fixtures / name).read_text(encoding="utf-8"))


registry = {                                    # ruleset id -> document, so `extends` can be resolved
    "acme.org.base": read("acme-org-base.ruleset.json"),
    "acme.payments.transfer": read("payments-transfer.ruleset.json"),
}
schemas = {"./payments.openapi.yaml": read("payments.openapi.json")}

rules = load(registry["acme.payments.transfer"], registry, schemas.get)
print(rules.id, rules.version, rules.checksum)
# acme.payments.transfer 1.0.0 sha256:c192dd53b5b1d307d52ccbc27fc1674114e8714d53b699b24088a648ae242c7e

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']
print(result["effects"])
# [{'type': 'value', 'field': '/fee', 'value': 180, 'rule': 'transfer.fee.international'}]

• registry maps ruleset ids to source documents. The parent named by extends must be in it.
• loader(file) returns the parsed document behind the file part of an entity's $ref, or None. Without a loader the PATH_UNKNOWN and SCHEMA_REF_UNRESOLVED checks are skipped; every other check still runs.
• The result is a dict with ruleset, version, checksum, decision, findings, effects and commands, as specification section 8 defines them.

evaluate(request, channel="server", operators=None) checks the shape of the request before it evaluates anything and raises ValueError for a request that is not well formed:

rules.evaluate({"entity": "Transfer", "operation": "create", "data": []})
# ValueError: 'data' must be an object

A ruleset that breaks a rule of section 5 does not load. LoadError.problems is a list of {code, message, rule?} and LoadError.codes lists the codes:

broken = json.loads(json.dumps(registry["acme.payments.transfer"]))
broken["overrides"]["params"]["maxTransferAmount"] = 60000      # the parent allows only lower values
try:
    load(broken, registry, schemas.get)
except LoadError as err:
    print(err.codes)
# ['PARAM_LOOSENED']

YAML

The package reads no YAML: it takes parsed documents. read() in tools/rulecheck.py parses a ruleset file by the YAML 1.2 core schema, as specification section 12 requires, and python tools/rulecheck.py check <file> reports every scalar that YAML 1.1 and YAML 1.2 parsers read differently. A file that passes that check means the same to yaml.safe_load.

Manifests and channels

A loaded ruleset holds two manifests. The server manifest contains everything. The client manifest contains only what a browser may see.

print(len(rules.manifest("server")["rules"]), len(rules.manifest("client")["rules"]))
# 14 9
print(sorted(rules.manifest("client")["params"]))
# ['internationalFeeRate', 'largeTransferThreshold', 'maxTransferAmount']

blocked = {"entity": "Transfer", "operation": "create",
           "data": {"type": "international", "amount": 500, "currency": "USD", "memo": "gift",
                    "beneficiary": {"name": "X", "country": "KP", "swiftCode": "ABCDKPPY"}}}
print(rules.evaluate(blocked)["decision"], rules.evaluate(blocked, "client")["decision"])
# deny allow

The rule that blocks the country is enforcement: server, so the client channel does not know it. A client evaluation is advice; the server evaluation is the decision.

rule_cascade.evaluate(manifest, request, operators=None) evaluates a manifest received from elsewhere, for example one fetched from a rule server.

Bundles

A bundle is the compiled form of a ruleset: one JSON document with both manifests. Compile once, in CI, and evaluate the same bundle in every runtime.

from rule_cascade import RuleSet

bundle = rules.bundle()
print(list(bundle))
# ['ruleCascadeBundle', 'id', 'version', 'checksum', 'manifests']
Path("acme.payments.transfer.bundle.json").write_text(json.dumps(bundle), encoding="utf-8")

loaded = RuleSet.from_bundle(json.loads(Path("acme.payments.transfer.bundle.json").read_text(encoding="utf-8")))
print(loaded.checksum == rules.checksum, loaded.resolved)
# True None

RuleSet.from_bundle raises LoadError with BUNDLE_UNSUPPORTED for a bundle whose format is not version 1.x and BUNDLE_INVALID for one without a usable server and client manifest. It runs no other check, because the compiler already did. resolved is None for a ruleset read from a bundle: a bundle holds manifests, not the resolved source. A bundle contains the server manifest, so it is never sent to a browser.

From the command line, python tools/rulecheck.py compile <file> -o <id>.bundle.json writes the bundle of a ruleset file.

Custom operators

A ruleset declares the custom operators it uses under operators and calls them as {"op": "x-<name>", "args": [...]}. The host supplies each one as a function, by name:

def luhn(text):
    if not isinstance(text, str) or len(text) < 2 or not all("0" <= c <= "9" for c in text):
        return False
    total = 0
    for i, c in enumerate(reversed(text)):
        d = int(c) * (2 if i % 2 else 1)
        total += d - 9 if d > 9 else d
    return total % 10 == 0


operators = {"x-luhn": luhn}

customer = RuleSet.from_bundle(json.loads(
    Path("conformance/bundles/acme.onboarding.customer.bundle.json").read_text(encoding="utf-8")))
print(customer.manifest("server")["operators"])      # what this manifest needs; check it at start-up
# ['x-luhn']

request = {"entity": "Customer", "operation": "update", "original": {},
           "data": {"loyaltyNumber": "79927398710"}, "view": {"section": "membership"}}
finding = customer.evaluate(request, "server", operators)["findings"][0]
print(finding["code"], finding["message"])
# ONB-CUS-001 This loyalty number is not valid. Check the digits.
print(finding["location"])
# {'page': 'onboarding', 'screen': 'profile', 'section': 'membership'}

finding = customer.evaluate(request, "server")["findings"][0]      # not registered: fails closed
print(finding["code"], finding["blocking"], finding["detail"])
# RULE-EVALUATION-ERROR True custom operator x-luhn is not registered

An operator receives its arguments as positional plain values (None, bool, str, int or float rounded to 15 significant digits, list, dict) and returns a JSON value. It must be a pure function. An operator that is missing, raises, or returns NaN or an infinity makes the rule fail closed with a RULE-EVALUATION-ERROR finding.

The view in the request above narrows the evaluation to one place in the user interface, and finding["location"] reports the place the rule's target names (specification section 8).

Check at start-up that every operator the rules need is registered. A missing operator does not fail the load; it fails every rule that uses it, closed.

rules = RuleSet.from_bundle(bundle)                              # the onboarding catalog
print(rules.missing_operators({}))                               # ['x-luhn']
print(rules.missing_operators({"x-luhn": lambda text: True}))    # []

One manifest on its own

A front end receives the client manifest, not the bundle. A ruleset read from one manifest has that one channel:

from rule_cascade import ChannelError, RuleSet

client = RuleSet.from_manifest(bundle["manifests"]["client"])
print(client.channels)                                           # ['client']
print(client.evaluate({"entity": "Customer", "operation": "create", "data": {"fullName": "Maya"}})["decision"])  # deny
try:
    client.evaluate({"entity": "Customer", "operation": "create"}, "server")
except ChannelError as err:
    print(err)                                                   # ruleset acme.onboarding.customer has no server manifest

Expressions

from rule_cascade import EvalError, evaluate_expression

print(evaluate_expression({"op": "add", "args": [0.1, 0.2]}))
# 0.3
vat = {"params": ["amount"],
       "body": {"op": "round", "args": [{"op": "mul", "args": [{"var": "arg.amount"}, 0.21]}, 2]}}
print(evaluate_expression({"fn": "vat", "args": [{"var": "data.net"}]}, {"data": {"net": 19.99}}, {"vat": vat}))
# 4.2
print(evaluate_expression({"var": "data.rate"}, {"data": {"rate": 0.1234567890123456}}))
# 0.123456789012346
try:
    evaluate_expression({"op": "lt", "args": [None, 100]})
except EvalError as err:
    print(err)
# number expected, got None

evaluate_expression(expr, env=None, functions=None, operators=None) returns a plain JSON value and raises EvalError for an evaluation error. Arithmetic is decimal with 34 significant digits; every number that leaves is rounded half even to 15 significant digits (specification 4.2). rule_cascade.expressions.pattern_problem(pattern) returns why a pattern is outside the portable subset of specification 4.4, or None when it is inside.

Refreshing the rules

RuleSetHolder loads the rules again on an interval (seconds) or a cron schedule and swaps them in. When a load raises, it keeps the last good rules; a ruleset with the checksum already held is not swapped in. The schedule runs on a daemon threading.Timer.

from rule_cascade import RuleSet, RuleSetHolder


def load_bundle():
    return RuleSet.from_bundle(json.loads(Path("transfer.bundle.json").read_text(encoding="utf-8")))


rules = RuleSetHolder(load_bundle, interval=300, on_reload=lambda r: r.error and print("not refreshed:", r.error))
# or RuleSetHolder(load_bundle, cron="0 * * * *", tz=ZoneInfo("Europe/Paris"))
result = rules.get().evaluate(request)
rules.close()

CronSchedule("*/5 * * * *").next(after, tz) is the cron syntax of docs/caching.md. packages/python/bench/evaluate.py measures evaluations per second (docs/performance.md).

Engine protocol

The package runs as a program that any language drives over standard input and output: one JSON request per line in, one JSON response per line out (specification section 13).

printf '%s\n' '{"id":1,"command":"version"}' '{"id":2,"command":"expression","expr":{"op":"add","args":[0.1,0.2]}}' \
  | python -m rule_cascade engine
# {"id":1,"ok":true,"result":{"engine":"rule-cascade-python","engineVersion":"1.0.0a1","ruleCascade":"1.0.0","bundle":"1.0.0","levels":["evaluator","compiler"],"operators":[]}}
# {"id":2,"ok":true,"result":0.3}

It implements version, load, manifest, evaluate, expression and compile. --conformance-operators registers the three operators of the conformance suite (x-test-reverse, x-test-sum, x-luhn); without the option no custom operator is registered, and rules that use one fail closed. To serve the protocol with operators of your own, call rule_cascade.engine.serve(sys.stdin, sys.stdout, operators) from a program of yours, or use the Engine class directly: Engine(operators).handle(request) takes a request object and returns the response object.

Role as the reference implementation

The other runtimes (TypeScript, Java, Go) are ports of this package and must agree with it on every case of the conformance suite. Three things follow:

• A change to the specification is implemented here first, then in the other runtimes.
• The generated parts of the suite (checksums, bundles, the evaluation corpus) are produced by this package through python tools/rulecheck.py sync. They prove that the runtimes agree with the reference. The hand-written expression, load-error and protocol cases and the golden tests are what tie the reference to the specification.
• tools/rulecheck.py is the command-line front end of this package: check, compile, manifest, conformance, sync, jsonlogic and derive.

The code is written to be read next to the specification. It is not optimised: use it for tooling, tests and services where Python is the language of the host.

Tests

From the repository root, with the tool dependencies installed (pip install -r tools/requirements.txt):

PYTHONPATH=packages/python/src python -m unittest discover -s packages/python/tests -q   # or: make python
python tools/rulecheck.py conformance | tail -1                                          # or: make conformance
# conformance: 1899 cases, 0 failure(s)
PYTHONPATH=packages/python/src python tools/rulecheck.py conformance \
  --engine "python -m rule_cascade engine --conformance-operators" | tail -1
# conformance: 1899 cases, 0 failure(s)

The first command runs the whole conformance suite through the engine protocol in process and checks that the generated files are current. The last one runs the same cases against the package started as a separate program.