Python Rules Engine

import zen

with open('./pricing-rules.json') as f:
    content = f.read()

engine = zen.ZenEngine()
decision = engine.create_decision(content)

response = decision.evaluate({
    "customer": {"tier": "gold", "yearsActive": 3},
    "order": {"subtotal": 150, "items": 5}
})

print(response["result"])
# => {'discount': 0.15, 'freeShipping': True}

import json
import zen

with open("./rules/pricing.json") as f:
    pricing = json.load(f)

engine = zen.ZenEngine({
    "loader": {"type": "static", "content": {"pricing.json": pricing}}
})

response = engine.evaluate("pricing.json", {"amount": 100})
print(response["result"])

import zen

engine = zen.ZenEngine({"loader": {"type": "fs", "path": "./rules"}})

response = engine.evaluate("pricing.json", {"amount": 100})
print(response["result"])

import boto3
import zen

s3 = boto3.client("s3")
obj = s3.get_object(Bucket="my-rules-bucket", Key="decisions.zip")

engine = zen.ZenEngine({"loader": {"type": "zip", "bytes": obj["Body"].read()}})

response = engine.evaluate("pricing.json", {"amount": 100})
print(response["result"])

import zen

def loader(key):
    with open(f"./rules/{key}") as f:
        return f.read()

engine = zen.ZenEngine({"loader": loader})

response = engine.evaluate("pricing.json", {"amount": 100})
print(response["result"])

results = engine.evaluate_batch([
    {"key": "pricing.json", "context": {"amount": 100}},
    {"key": "pricing.json", "context": {"amount": 250}},
    {"key": "eligibility.json", "context": {"age": 30}},
])

for result in results:
    if result["success"]:
        print(result["data"]["result"])
    else:
        print("Evaluation failed:", result["error"])

import asyncio
import zen

async def loader(key):
    with open(f"./rules/{key}") as f:
        return f.read()

engine = zen.ZenEngine({"loader": loader})

async def main():
    # Evaluate multiple decisions concurrently
    results = await asyncio.gather(
        engine.async_evaluate("pricing.json", {"amount": 100}),
        engine.async_evaluate("eligibility.json", {"score": 750}),
        engine.async_evaluate("shipping.json", {"weight": 5})
    )

    for response in results:
        print(response["result"])

asyncio.run(main())

try:
    response = decision.evaluate(input_data)
    print(response["result"])
except Exception as e:
    print(f"Evaluation failed: {e}")

response = decision.evaluate(input_data, {"trace": True})

print(response["trace"])
# Each node's input, output, and performance timing

print(response["performance"])
# Total evaluation time

import zen

# Standard expressions
result = zen.evaluate_expression("a + b", {"a": 5, "b": 3})
# => 8

total = zen.evaluate_expression("sum(items)", {"items": [1, 2, 3, 4]})
# => 10

# Unary expressions (comparison against $)
is_valid = zen.evaluate_unary_expression(">= 5", {"$": 10})
# => True

in_list = zen.evaluate_unary_expression("'US', 'CA', 'MX'", {"$": "US"})
# => True

import zen

expr = zen.compile_expression("a * b + c")

result1 = expr.evaluate({"a": 2, "b": 3, "c": 4})
# => 10

result2 = expr.evaluate({"a": 5, "b": 6, "c": 7})
# => 37