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 zen

cache = {}

def loader(key):
    if key in cache:
        return cache[key]

    with open(f"./rules/{key}") as f:
        content = zen.ZenDecisionContent(f.read())
        cache[key] = content
        return content

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

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

import boto3
import zen
import zipfile
import io

s3 = boto3.client("s3")
rules = {}

# Download and extract all decisions at startup
obj = s3.get_object(Bucket="my-rules-bucket", Key="decisions.zip")
zip_bytes = obj["Body"].read()

with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
    for name in zf.namelist():
        if not name.endswith("/"):
            rules[name] = zen.ZenDecisionContent(zf.read(name).decode("utf-8"))

engine = zen.ZenEngine({"loader": lambda key: rules.get(key)})

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

from azure.storage.blob import BlobServiceClient
import zen
import zipfile
import io
import os

blob_service = BlobServiceClient.from_connection_string(os.environ["AZURE_STORAGE_CONNECTION"])
container = blob_service.get_container_client("rules")
rules = {}

# Download and extract all decisions at startup
blob = container.get_blob_client("decisions.zip")
zip_bytes = blob.download_blob().readall()

with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
    for name in zf.namelist():
        if not name.endswith("/"):
            rules[name] = zen.ZenDecisionContent(zf.read(name).decode("utf-8"))

engine = zen.ZenEngine({"loader": lambda key: rules.get(key)})

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

from google.cloud import storage
import zen
import zipfile
import io

client = storage.Client()
bucket = client.bucket("my-rules-bucket")
rules = {}

# Download and extract all decisions at startup
blob = bucket.blob("decisions.zip")
zip_bytes = blob.download_as_bytes()

with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
    for name in zf.namelist():
        if not name.endswith("/"):
            rules[name] = zen.ZenDecisionContent(zf.read(name).decode("utf-8"))

engine = zen.ZenEngine({"loader": lambda key: rules.get(key)})

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

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