How to create a custom function?

[eshan-pragma]

Please consider the following data

orders = [
    {
        'id': 1,
        'price': 5000,
        'products': ['PRD_A', 'PRD_B']
    },
    {
        'id': 2,
        'price': 2000,
        'products': ['PRD_C', 'PRD_B']
    },
    {
        'id': 3,
        'price': 1000,
        'products': ['PRD_C', 'PRD_B']
    },
    {
        'id': 4,
        'price': 4000,
        'products': ['PRD_A', 'PRD_B', 'PRD_C']
    },
]

and a custom function:

def past_orders_found(id):
    if id in [2,3]:
        return True
    else:
        return False

Now i want to evaluate the data via my custom function like:

past_orders_rule = rule_engine.Rule('$past_orders_found(id) == true')

I am unable to achieve this. I also tried creating a type_resolver and then pass the context to the rule like below, and still did not work.

def type_resolver(name):
    if name == 'id':
        return rule_engine.DataType.FLOAT
    if name == 'past_orders_found':
        return rule_engine.DataType.FUNCTION(
            # the name of the function is provided for error messages
            'past_orders_found',
            # the return data type, in this case an array of strings
            return_type=ast.DataType.BOOLEAN,
            # the data type of each of the three arguments
            argument_types=(
            ast.DataType.FLOAT,
            ),
            # the minimum number of arguments, in this case the second two arguments are optional
            minimum_arguments=1
        )
    # if the name is none of those, raise a SymbolResolutionError
    raise rule_engine.errors.SymbolResolutionError(name)

context = rule_engine.Context(type_resolver=type_resolver)

[zeroSteiner]

The type resolver isn't the issue. That's only necessary if you want to type hinting. Your function definition for past_orders_found looks correct though, but it needs to go somewhere else if you define past_orders_found as a builtin symbol. If you want the function available as a builtin, you need to create a custom context.

import rule_engine

orders = [
    {
        'id': 1,
        # ...
    },
    {
        'id': 2,
    },
    {
        'id': 3,
    },
    {
        'id': 4,
    },
]

def past_orders_found(id):
    return id in [2,3]

class CustomContext(rule_engine.Context):
	def __init__(self, *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.builtins = rule_engine.builtins.Builtins.from_defaults(
			{'past_orders_found': past_orders_found}
		)

rule = rule_engine.Rule('$past_orders_found(id) == true', CustomContext())

for order in orders:
	print("order#{} result: {}".format(order['id'], rule.matches(order)))

If you didn't want it as a builtin symbol, you could tweak it slightly:

def past_orders_found(id):
    return id in [2,3]

rule = rule_engine.Rule('past_orders_found(id) == true')

for order in orders:
	print("order#{} result: {}".format(order['id'], rule.matches({'past_orders_found': past_orders_found} | order)))

A function is just a symbol that's callable, so you pass it around the same as any other by either adding it to the namespace of the thing that's evaluated Rule.matches or the builtin symbols.

[eshan-pragma]

Thank you so much for the quick reply. I will test this out and confirm here. Thanks again.

[jjwsz]

Hello @zeroSteiner, the solution works for me too, but how do I use the rule_engine.DataType.FUNCTION(...) to verify the input of that custom function? In the example above, I would expect a error, if for "id" a sting is entered. How do I achieve that?

[zeroSteiner]

You just have to define the argument type to the function and then you should start to see FunctionCallError exceptions. In the following example, past_orders_found is defined as a function that takes a FLOAT argument and returns a BOOLEAN result. The rule on In [5] is okay, but the rule on In [6] will raise an exception because the string "test" will never be a valid argument.

In [1]: import rule_engine

In [2]: help(rule_engine.DataType.FUNCTION)

In [3]: f = rule_engine.DataType.FUNCTION('past_orders_found', argument_types=(rule_engine.DataType.FLOAT,), return_type=rule_engine.DataType.BOOLEAN)

In [4]: context = rule_engine.Context(type_resolver={'past_orders_found': f})

In [5]: r = rule_engine.Rule('past_orders_found(1)', context=context)

In [6]: r = rule_engine.Rule('past_orders_found("test")', context=context)
---------------------------------------------------------------------------
FunctionCallError                         Traceback (most recent call last)
Cell In[6], line 1
----> 1 r = rule_engine.Rule('past_orders_found("test")', context=context)

File ~/Repositories/rule-engine/lib/rule_engine/engine.py:599, in Rule.__init__(self, text, context)
    597 self.text = text
    598 self.context = context
--> 599 self.statement = self.parser.parse(text, context)

File ~/Repositories/rule-engine/lib/rule_engine/parser/base.py:82, in ParserBase.parse(self, text, context, **kwargs)
     80 	self.context = None
     81 # phase 2: initialize each AST node recursively, providing them with an opportunity to define assignments
---> 82 return result.build()

File ~/Repositories/rule-engine/lib/rule_engine/parser/__init__.py:56, in _DeferredAstNode.build(self)
     54 def build(self):
     55 	constructor = getattr(self.cls, self.method)
---> 56 	return constructor(*self.args, **self.kwargs)

File ~/Repositories/rule-engine/lib/rule_engine/ast.py:1152, in Statement.build(cls, context, expression, **kwargs)
   1150 @classmethod
   1151 def build(cls, context, expression, **kwargs):
-> 1152 	return cls(context, expression.build(), **kwargs).reduce()

File ~/Repositories/rule-engine/lib/rule_engine/parser/__init__.py:56, in _DeferredAstNode.build(self)
     54 def build(self):
     55 	constructor = getattr(self.cls, self.method)
---> 56 	return constructor(*self.args, **self.kwargs)

File ~/Repositories/rule-engine/lib/rule_engine/ast.py:1071, in FunctionCallExpression.build(cls, context, function, arguments)
   1069 @classmethod
   1070 def build(cls, context, function, arguments):
-> 1071 	return cls(context, function.build(), tuple(argument.build() for argument in arguments)).reduce()

File ~/Repositories/rule-engine/lib/rule_engine/ast.py:1065, in FunctionCallExpression.__init__(self, context, function, arguments)
   1063 if self.function.result_type != DataType.UNDEFINED:
   1064 	function_type = self.function.result_type
-> 1065 	self._validate_function(function_type, arguments)
   1066 	self.result_type = function_type.return_type
   1067 self.arguments = arguments

File ~/Repositories/rule-engine/lib/rule_engine/ast.py:1123, in FunctionCallExpression._validate_function(self, function_type, arguments)
   1121 	arg1_type = DataType.from_value(arg1)
   1122 if not DataType.is_compatible(arg1_type, arg2_type):
-> 1123 	raise errors.FunctionCallError(
   1124 		"data type mismatch (argument #{})".format(pos),
   1125 		function_name=function_type.value_name
   1126 	)

FunctionCallError: data type mismatch (argument #1)

In [7]: 

[jjwsz]

I have to retract and edit my earlier feedback. It does not really do what I expect. I will try working on it tough. But I was thinking about merging the tow answers, because I like to use the function as bulitin, for a more better reading and understanding of a rule, and to point out to the reader that its a function call form the rule.... but I just don't manage to do it.