Version 1.0.0 - Pydantic

README.md

@@ -5,9 +5,11 @@
 [![license](https://img.shields.io/github/license/pydantic/pydantic.svg)](https://github.com/pydantic/pydantic/blob/main/LICENSE)
 
 
-
 A rule engine where rules are defined in JSON format. The syntax of the rules belongs to the [json-rules-engine](https://github.com/CacheControl/json-rules-engine) javascript library though it contains some changes to make it more powerfull.
 
+- [Rule Syntax](docs/rules.md)
+- [Operators](docs/operators.md)
+
 ## Installation
 ```
 pip install python-rule-engine
@@ -50,6 +52,35 @@ results = engine.evaluate(obj)
 
 ```
 
-## Rule Format
+## Generating rules with ChatGPT
+You can leverage the Rule's model definition to generate rules using ChatGPT. The following example shows how to generate a rule using OpenAI's API. More info on how to use the API can be found [here](https://platform.openai.com/docs/guides/structured-outputs).
+
+```python
+from openai import OpenAI
+from python_rule_engine import Rule
+
+
+obj = {
+    "player": {
+        "name": "Lionel",
+        "age": 34,
+    }
+}
+
+client = OpenAI(api_key="your_api_key")
+
+completion = client.beta.chat.completions.parse(
+    model="gpt-4o-2024-08-06",
+    messages=[
+        {"role": "system", "content": "Generate a json rule that matches the following conditions:"},
+        {"role": "user", "content": "Create a rule that matches if the player's name is Lionel and the player's age bigger than 30"},
+    ],
+    response_format=Rule,
+)
+
+rule = completion.choices[0].message.parsed
+
+engine = RuleEngine([rule])
 
-Find more info about the rules [here](docs/rules.md).
+assert engine.evaluate(obj).match
+```

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python-rule-engine"
-version = "0.5.1"
+version = "1.0.0"
 description = "A rule engine where rules are written in JSON format"
 authors = ["Santiago Alvarez <[email protected]>"]
 homepage = "https://github.com/santalvarez/python-rule-engine"
@@ -17,8 +17,10 @@ classifiers = [
 keywords = ["rule-engine", "rules", "json", "python"]
 
 [tool.poetry.dependencies]
-python = "^3.7.2"
+python = "^3.9"
 jsonpath-ng = "^1.5.3"
+pydantic = "^2.10.6"
+pydantic-core = "^2.29.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.2.0"

src/python_rule_engine/__init__.py

@@ -1,6 +1,8 @@
 import importlib.metadata
 from .engine import RuleEngine
 from .operators import Operator
+from .decoder import RuleDecoder
+
 
 __version__ = importlib.metadata.version("python-rule-engine")
 

src/python_rule_engine/decoder.py

@@ -0,0 +1,64 @@
+import json
+from json import JSONDecodeError
+from typing import Any, Dict, List, Optional, Type
+
+from pydantic import ValidationError
+
+from .errors import (DuplicateOperatorError, InvalidRuleJSONError,
+                     InvalidRuleSchemaError, InvalidRuleTypeError)
+from .models.rule import Rule
+from .operators import DEFAULT_OPERATORS, Operator
+
+
+class RuleDecoder:
+    """ Class responsible for decoding rules """
+    def __init__(self, custom_operators: Optional[List[Type[Operator]]]=None):
+        self.operators = {}
+
+        if custom_operators is None: custom_operators = []
+
+        for p in DEFAULT_OPERATORS + custom_operators:
+            if p.id in self.operators:
+                raise DuplicateOperatorError
+            self.operators[p.id] = p
+
+
+    def decode_rules(self, rules: List[Dict]) -> List[Rule]:
+        """ Decodes a list of rule dictionaries into a list of Rule objects """
+        return [self.decode_rule(rule) for rule in rules]
+
+    def decode_rule(self, rule: Dict) -> Rule:
+        """ Decodes a rule dictionary into a Rule object """
+        if not isinstance(rule, dict):
+            raise InvalidRuleTypeError
+        try:
+            return Rule(**rule, operators_dict=self.operators)
+        except Exception as e:
+            raise InvalidRuleSchemaError from e
+
+    def decode_str_rules(self, rules: List[str]) -> List[Rule]:
+        """ Decodes a list of rule strings into a list of Rule objects """
+        return [self.decode_str_rule(rule) for rule in rules]
+
+    def decode_str_rule(self, rule: str) -> Rule:
+        """ Decodes a rule string into a Rule object """
+        if not isinstance(rule, str):
+            raise InvalidRuleTypeError
+        try:
+            rule_dict = json.loads(rule)
+            return Rule(**rule_dict, operators_dict=self.operators)
+        except ValidationError as e:
+            raise InvalidRuleSchemaError from e
+        except JSONDecodeError as e:
+            raise InvalidRuleJSONError
+
+    def rule_schema(self) -> Dict[str, Any]:
+        schema = Rule.model_json_schema()
+
+        operator_names = list(self.operators.keys())
+
+        schema["$defs"]["SimpleCondition"]["properties"]["operator"]["examples"] = operator_names
+
+        return schema
+
+

src/python_rule_engine/engine.py

@@ -1,38 +1,35 @@
-from copy import deepcopy
-from typing import Any, Dict, List, Optional, Type
+from typing import Any, Dict, List, Optional
 
-from .exceptions import DuplicateOperatorError
 from .models.rule import Rule
-from .operators import (Contains, Equal, GreaterThan, GreaterThanInclusive, In,
-                        LessThan, LessThanInclusive, NotContains, NotEqual,
-                        NotIn, Operator)
+from .operators import Operator
+from .decoder import RuleDecoder
 
 
 class RuleEngine:
-    default_operators: List[Type[Operator]] = [Equal, NotEqual, LessThan,
-                                               LessThanInclusive, GreaterThan, GreaterThanInclusive,
-                                               In, NotIn, Contains, NotContains]
-
     def __init__(self, rules: List[Dict], operators: Optional[List[Operator]] = None):
-        self.operators: Dict[str, Type[Operator]] = self._merge_operators(operators)
-        self.rules = self._deserialize_rules(rules)
-
-    def _merge_operators(self, operators: Optional[List[Type[Operator]]] = None) -> Dict[str, Type[Operator]]:
-        merged_operators = {}
-
-        if operators is None:
-            operators = []
-        for p in self.default_operators + operators:
-            if p.id in merged_operators:
-                raise DuplicateOperatorError
-            merged_operators[p.id] = p
-        return merged_operators
-
-    def _deserialize_rules(self, rules: List[Dict]) -> List[Rule]:
-        aux_rules = []
-        for rule in rules:
-            aux_rules.append(Rule(rule, self.operators))
-        return aux_rules
+        self.decoder = RuleDecoder(operators)
+        self.rules = self.decoder.decode_rules(rules)
+
+    def add_rule(self, rule: Dict):
+        """ Add a rule to the engine
+
+        :param Dict rule: The rule to add
+        """
+        self.rules.append(self.decoder.decode_rule(rule))
+
+    def add_str_rule(self, rule: str):
+        """ Add a rule to the engine
+
+        :param str rule: The rule to add
+        """
+        self.rules.append(self.decoder.decode_str_rule(rule))
+
+    def remove_rule(self, rule_name: str):
+        """ Remove a rule from the engine
+
+        :param str rule_name: The name of the rule to remove
+        """
+        self.rules = [rule for rule in self.rules if rule.name != rule_name]
 
     def evaluate(self, obj: Any) -> List[Rule]:
         """ Evaluate an object on the loaded rules
@@ -42,7 +39,7 @@ def evaluate(self, obj: Any) -> List[Rule]:
         """
         results = []
         for rule in self.rules:
-            rule_copy = deepcopy(rule)
+            rule_copy = rule.model_copy(deep=True)
             rule_copy.conditions.evaluate(obj)
 
             if rule_copy.conditions.match:

src/python_rule_engine/errors.py

@@ -0,0 +1,26 @@
+from pydantic import ValidationError
+
+class RuleEngineBaseError(Exception):
+    """ Base Rule Engine Exception """
+
+class DuplicateOperatorError(RuleEngineBaseError):
+    """ Raised when there is already a operator with the same ID
+    loaded in the engine """
+
+class JSONPathValueNotFoundError(RuleEngineBaseError):
+    """ Raised when the value indicated by 'path' could not be found
+    on the object."""
+
+class RuleDecodeError(RuleEngineBaseError):
+    """ Base error for all rule decoding errors """
+
+class InvalidRuleSchemaError(RuleDecodeError, ValidationError):
+    """ Raised when the field of the provided rule don't match with the rule's schema """
+
+class InvalidRuleTypeError(RuleDecodeError):
+    """ Raised when the provided rule is of an incorrect type """
+
+class InvalidRuleJSONError(RuleDecodeError):
+    """ Raised when the rule can't be converted to JSON.
+    This will be raised when a provided string rule has invalid JSON format """
+

src/python_rule_engine/json_path.py

@@ -1,19 +1,25 @@
 from typing import Any
 
 from jsonpath_ng import parse
+from pydantic_core import core_schema
 
-from .exceptions import JSONPathValueNotFound
+from .errors import JSONPathValueNotFoundError
 
 
-class JSONPath:
-    def __init__(self, path: str) -> None:
-        self.original_path: str = path
-        self.jsonpath_parsed = parse(path)
+class JSONPath(str):
+    def __new__(cls, value):
+        self = super().__new__(cls, value)
+        self.parsed =  parse(self)
+        return self
+
+    @classmethod
+    def __get_pydantic_core_schema__(cls, source_type, handler):
+        return core_schema.no_info_after_validator_function(cls, handler(str))
 
     def get_value_from(self, obj: Any) -> Any:
-        result = self.jsonpath_parsed.find(obj)
+        result = self.parsed.find(obj)
         if len(result) == 0:
-            raise JSONPathValueNotFound(f"Value not found at path {self.original_path}")
+            raise JSONPathValueNotFoundError(f"Value not found at path {self}")
 
         if len(result) == 1:
             return result[0].value

src/python_rule_engine/models/condition.py

@@ -1,7 +1,9 @@
+from pydantic import BaseModel
+from pydantic.json_schema import SkipJsonSchema
 
-class Condition:
-    def __init__(self) -> None:
-        self.match = False
+
+class Condition(BaseModel):
+    match: SkipJsonSchema[bool] = False
 
     def evaluate(self, obj):
         raise NotImplementedError

src/python_rule_engine/models/multi_condition.py

@@ -1,42 +1,41 @@
 from __future__ import annotations
 
-from typing import List, Optional
+from typing import Dict, List, Union
+
+from pydantic import Field, model_validator
+from pydantic.json_schema import SkipJsonSchema
 
 from .condition import Condition
 from .simple_condition import SimpleCondition
 
 
 class MultiCondition(Condition):
-    def __init__(self, **data) -> None:
-        super().__init__()
+    any: List[Union[SimpleCondition, MultiCondition]] = None
+    all: List[Union[SimpleCondition, MultiCondition]] = None
+    not_: Union[SimpleCondition, MultiCondition] = Field(None, alias="not")
+
+    operators_dict: SkipJsonSchema[Dict] = Field(..., exclude=True)
 
-        if sum([bool(data.get("any", [])), bool(data.get("all", [])), bool(data.get("not", {}))]) != 1:
+    @model_validator(mode="after")
+    def validate_conditions(self):
+        if sum([bool(self.any), bool(self.all), bool(self.not_)]) != 1:
             raise ValueError("Only one of any, all or not can be defined")
+        return self
+
+    @model_validator(mode="before")
+    @classmethod
+    def set_operators_dict(cls, values):
+        operators_dict = values["operators_dict"]
+
+        for key in ["any", "all"]:
+            if values.get(key):
+                for item in values[key]:
+                    item["operators_dict"] = operators_dict
+
+        if values.get("not"):
+            values["not"]["operators_dict"] = operators_dict
 
-        self.all = self.__validate_conditions(data.get("all", []), data["operators_dict"])
-        self.any = self.__validate_conditions(data.get("any", []), data["operators_dict"])
-        self.not_ = self.__validate_not_condition(data.get("not", {}), data["operators_dict"])
-
-    def __validate_conditions(self, data: List[dict], operators_dict) -> Optional[List[Condition]]:
-        if not data:
-            return None
-        cds = []
-        for cd in data:
-            cd["operators_dict"] = operators_dict
-            try:
-                cds.append(SimpleCondition(**cd))
-            except ValueError:
-                cds.append(MultiCondition(**cd))
-        return cds
-
-    def __validate_not_condition(self, data: dict, operators_dict) -> Optional[Condition]:
-        if not data:
-            return None
-        data["operators_dict"] = operators_dict
-        try:
-            return SimpleCondition(**data)
-        except ValueError:
-            return MultiCondition(**data)
+        return values
 
     def evaluate(self, obj):
         """ Run a multi condition on an object or dict