How can I create a rules engine without using eval() or exec()?

Question

I have a simple rules/conditions table in my database which is used to generate alerts for one of our systems. I want to create a rules engine or a domain specific language.

A simple rule stored in this table would be..(omitting the relationships here)

if temp > 40 send email

Please note there would be many more such rules. A script runs once daily to evaluate these rules and perform the necessary actions. At the beginning, there was only one rule, so we had the script in place to only support that rule. However we now need to make it more scalable to support different conditions/rules. I have looked into rules engines , but I hope to achieve this in some simple pythonic way. At the moment, I have only come up with eval/exec and I know that is not the most recommended approach. So, what would be the best way to accomplish this??

( The rules are stored as data in database so each object like "temperature", condition like ">/=..etc" , value like "40,50..etc" and action like "email, sms, etc.." are stored in the database, i retrieve this to form the condition...if temp > 50 send email, that was my idea to then use exec or eval on them to make it live code..but not sure if this is the right approach )

Answer 1

You will want to take a look at NebriOS. Rules are written in pure Python rather than storing them in the DB. For example:

class hello(NebriOS):
    listens_to = ['temp']

    def check(self):
        return self.temp > 40

    def action(self):
        send_email ("angela@example.com","""
            Alert: Temp is now > 40! """)

I think there is a tun of value in using a rules engine for this application. To quote Martin Fowler in describing one:

A rules engine is all about providing an alternative computational model. Instead of the usual imperative model, commands in sequence with conditionals and loops, it provides a list of production rules. Each rule has a condition and an action – simplistically you can think of it as a bunch of if-then statements.

Taking a nonlinear approach to certain software projects helps make it more robust, accurate, and easy to understand. A small rule like "temp > 40 then do x" is much easier to write as a stand alone rule than it is to build a full application that uses the same rule. It doesn't require a linear chain to become evaluated. Once written, always enforced.

Another benefit is that if one rule breaks down, the rest of the rules operate as normal. This would generally cause much more commotion with traditional software (imperative models) techniques.

I released this app to the wild after building it for my current company. I think rules rule. Just my two cents.

Answer 2

I have used @jsbuenos code snippet and done a few changes to form this. Basicaly, I also need support to check the "unit of measure" for a rule to evaluate the condition. So to support diff. rules like if time > 24 hrs send email OR if temp > 40 celsius send email etc.(I would probably have other units later on as well..) I have included a new dict to map the unit of measure to a calculate function and accordingly changed the callable function for the class. Would this be the right approach to do it?

import operator

class Rule(object):
    def __init__(self, variable_name, op, value, action):
      op_dict = {"=": operator.eq,
               ">": operator.gt,
               "<": operator.lt,
               #(...)
              }
      action_dict = {"email": email_function,
                   "log": log_function,
                   # ...
                  }

      eval_condition = {"hrs" :  self.raise_timeexceeded_alert,
                    "celsius" : self.raise_tempexceeded_alert,
                    #}  

      self.variable = variable_name
      self.op = op_dict[op]
      self.value = value
      self.action = action_dict[action]
      self.uom = measure      
      self.raise_alert = eval_condition[measure]

   def __call__(self, actual_value, *action_parameters):
     if self.raise_alert(actual_value,self.op,self.uom,self.threshold):
        return self.action(*action_parameters)
    return False

   def raise_timeexceeded_alert(self,timevalue, op, uom, threshold):
    #calculate time difference with respect to local timezone and return true
    # if diff is 'operator' threshold
    localtime=pytz.timezone(TIMEZONE)
    ....
    ...
    return False


   def raise_tempexceeded_alert(self,timevalue, op, uom, threshold):
     #return True if temp. is 'operator' threshold
     ....
     .....
     return False


rule = Rule("time", ">=", "24" , "hrs", "email")
args = [contact_email,message]
rule("2011-12-11 12:06:03",*args)

Answer 3

I think you mainly need two things:

• a Rule class to interface your front-end template
• pickle to store your rules in the db.

This is how your main could look like:

import pickle

# some data loaded from your DB
data = {'temp': 60, 'wind': 150}

# entry should be provided by your front-end template
entry = {'param_name': 'temp', 'test': Test(gt, 50), 'action': send_email}

rule = Rule(**entry)
to_store = pickle.dumps(rule)
# store 'to_store' into your DB

# Let's pretend to load the previously stored rule
stored = to_store
rule = pickle.loads(stored)
rule(data)

The idea is to get every information that you need to build a Rule from your template and then you just store that rule with pickle.

This may be a Rule implementation:

# =======
# Actions
# =======
#
# Any callable with no arguments is an Action
# (you may need to implement this)
#

def send_email():
    print('email sent')

# ==========
# Test class
# ==========
#
# Test class is a way to call your test function.
# The real test is in self.function
#

class Test:

    def __init__(self, function, *args):
        self.function = function
        self.args = args

    def __call__(self, parameter):
        return self.function(parameter, *self.args)

# ==============
# Test functions
# ==============
#
# These are the functions that are going to be executed
#

import operator

gt = operator.gt

def more_complex_test(*args):
    pass

# ==========
# Rule class
# ==========
#
# A Rule needs to know:
#  - the parameter to test
#  - the test to perform
#      - the action to execute
#

class Rule:

    def __init__(self, param_name, test, action):
        self.param_name = param_name
        self.test = test
        self.action = action

    def __call__(self, data):   # data is a dictionary {'temp': 60, ...}
        param_value = data[self.param_name]
        if self.test(param_value):
            return self.action()
        return False

Note: The two pieces of code above (if put togheter) can be executed. Give it a try!

Answer 4

Why do you need to store rules into a database?
Can't you just store the data in the database and put the rules into a python module?

For example in a file rules.py you could:

• Write a set of rules
• Write a parser for your data that will apply the right rule for each paramenter

Then in your main you just have to pass the data to your rules.parser() and everything will be taken care of.

Edit: Seen your comment I've made a new answer.

Answer 5

Since the "variable", "value" and comparison operator for eahc rule are on the database, you could write a Rule class that will take the apropriate parameters (operator, action, value, etc...), and yield a callable object that will receive all the relevant variables in the form of a dictionary, and take the proper registered action.

It would look like this, though you have to adapt it to proper get the parameters for your actions:

import operator

class Rule(object):
    def __init__(self, variable_name, op, value, action):
        op_dict = {"=": operator.eq,
                   ">": operator.gt,
                   "<": operator.lt,
                   #(...)
                  }
        action_dict = {"email": email_function,
                       "log": log_function,
                       # ...
                      }
        self.variable = variable_name
        self.op = op_dict[op]
        self.value = value
        self.action = action_dict[action]
    def __call__(self, value_dict, action_parameters, k_action_parameters):
        if self.op(value_dict[self.variable], self.value):
            return self.action(*action_parameters, **k_action_parameters)
        return False

rule = Rule("temp", ">", "email")
for result in query():
     rule(result, ())

Answer 6

There are several ways to achieve this. The other answers are valuable, and I'd like to add two techniques.

• Provided you can rewrite the table juste create each rules as a pickled function that you can deserialize when needed
• Write a big dictionary with rules as a key, and a function a a value. If you got 100 max rules, this is manageable. Just make sure you make very flexible functions using *args and **kwargs.

Example with pickle:

First, make a function that is flexible with its input.

def greater_than(value, *args, **kwargs):
    return all(value > i for i in args)

Then pickle it:

>>> import pickle
>>> rule = pickle.dumps(greater_than)
>>> rule # store this in DB
'ctest\ngreater_than\np0\n.'

Then when you need to get you business rule back:

>>> func = pickle.loads(rule) # rule is the sring from DB
>>> func(5, 4, 3, 1)
True
>>> func(5, 6) 
False

The purpose of having flexible input is that you can get an arbitrary number of parameters :

>>> args = [1, 2, 3]
>>> func(5, *args)
True 

Example with a dictionary

Store all functions in one big mapping:

def greater_than(value, *args, **kwargs):
    return all(value > i for i in args)

RULES = {
    'if x > y': greater_than
    'other rule': other_func,
    etc
}

Then when you need it:

   >>> func = RULES['if x > y']
   >>> func(5, 1)
   True

Answer 7

Well, if what you want to do is send emails then use the email module.

If I were you, I would write a simple Python script which processes a bunch of rules, probably just written as simple Python statements in a separate file, then send the emails / sms / ... for those rules that require an action to be performed.

You can make that run once a day (or whatever) using a service such as cron

For example, if your rules look like this:

# Rule file: rules.py

def rule1():
    if db.getAllUsers().contains("admin"): 
        return ('email', 'no admin user in db')
    else:
        return None, None

def rule2():
    if temp > 100.0: 
        return ('sms', 'too hot in greenhouse')
    else:
        return (None, None)

...

rules = [rule1, rule2, ....]

then your processing script might look like this:

# Script file: engine.py

import rules
import email
...

def send_email(message, receiver):
    # function that sends an email...

def send_sms(message, receiver):
    # function that sends an sms...

actions = {'email':send_email, 'sms':send_sms, ...}    

if __name__ == '__main__':

    # Declare receiver here...

    for rule in rules.rules:
        # Does the rule return a do-able action?
        # To be really paranoid we might wrap this in a try/finally
        # in case the rules themselves have any side effects,
        # or they don't all return 2-tuples.
        act, message = rule()
        if act in actions:
            # perform the action
            actions[rule()](message, receiver) 

Undoubtedly there are other ways to do this, such as creating a Pythonic DSL with which to write the rules.

Answer 8

Write a parser. See pyparsing.

Alternatively, make a table driven approach.