PyRulesEngine

PyRulesEngine is a cloud-agnostic, extensible Python library for abstracting hierarchical configuration logic without volatile eval vulnerabilities.

Features - YAML & JSON based rules definition compliant with Microsoft's JDM - Built using Google's Common Expression Language (CEL) for maximum security - Pluggable versioning structures via latest lookups - High-performance asynchronous execution tree - Action pipelines chaining Success/Failure outputs recursively

Table Of Content - Installation - Basic Usage - Create a workflow file with rules - Initialise RulesEngine with the workflow - Execute the workflow rules with input - Common Expression Language (CEL) support - Extending expression via custom actions

Installation

Assuming standard inclusion inside a larger Python project:

pip install -e .

Basic Usage

Create a workflow file with rules

Create Discount.yaml (Note: booleans are lowercase in CEL):

- WorkflowName: "Discount"
  Rules:
    - RuleName: "GiveDiscount10"
      Expression: "input1.country == 'india' && input1.loyaltyFactor <= 2 && input1.totalPurchasesToDate >= 5000 && input3.noOfVisitsPerMonth > 2"
    - RuleName: "GiveDiscount20"
      Expression: "input1.country == 'india' && input1.loyaltyFactor == 3 && input1.totalPurchasesToDate >= 10000 && input3.noOfVisitsPerMonth > 2"

Initialise RulesEngine with the workflow:

import asyncio
from rules_engine import RulesEngine, StorageManager
from rules_engine.storage.file_system import FileStorageProvider

manager = StorageManager()
manager.register_provider(FileStorageProvider("./rules"))
re = RulesEngine(manager)

Execute the workflow rules with input:

async def main():
    payload = {
        "input1": {"country": "india", "loyaltyFactor": 1, "totalPurchasesToDate": 6000},
        "input3": {"noOfVisitsPerMonth": 3}
    }

    resultList = await re.execute_all_rules_async("Discount", payload)

    for result in resultList:
        print(f"Rule - {result.rule.rule_name}, IsSuccess - {result.is_success}")

asyncio.run(main())

Common Expression Language (CEL) support

PyRulesEngine uses Google CEL for rule evaluation. CEL is intentionally limited and non-Turing complete, ensuring that rule authors cannot execute arbitrary code or trigger infinite loops. Typical usages include comparisons (>, <=, ==, !=), logical operators (&&, ||, !), and collection macros (.exists(), .all()).

Extending expression via custom actions

To chain behavior when nodes succeed or fail, PyRulesEngine uses robust Action decorators.

from rules_engine import register_action
from rules_engine.actions.base import ActionBase, ActionContext

@register_action("OutputExpression")
class OutputExpressionAction(ActionBase):
    async def run(self, context: ActionContext, inputs: dict) -> any:
        # Implementation to evaluate dynamic outputs natively
        pass

In YAML rule configurations, you hook your @register_action(NAMED_IDENTIFIER) using Actions.OnSuccess.Name:

Actions:
  OnSuccess:
    Name: OutputExpression
    Context:
      Expression: "customer.totalBilled * 0.9"