如何在不使用eval()或exec()的情况下创建规则引擎？

Question

我在数据库里有一个简单的规则/条件表，用来生成我们系统的警报。我想创建一个规则引擎或者一个特定领域的语言。

这个表里存储的一个简单规则是……（这里省略了关系）

if temp > 40 send email

请注意，这里会有很多这样的规则。每天会有一个脚本运行一次，来评估这些规则并执行必要的操作。最开始的时候只有一个规则，所以我们只需要一个支持这个规则的脚本。但现在我们需要让它更灵活，以支持不同的条件和规则。我查过规则引擎，但我希望能用一些简单的 Python 方法来实现。目前，我只想到使用 eval/exec，但我知道这并不是最推荐的做法。那么，最好的实现方法是什么呢？

（这些规则作为数据存储在数据库中，每个对象比如“温度”，条件比如“>、=等”，值比如“40、50等”，以及动作比如“邮件、短信”等，都存储在数据库里。我提取这些数据来形成条件……如果温度 > 50，就发送邮件。这是我想用 exec 或 eval 让它变成可执行代码的想法……但我不确定这是否是正确的方法。）

Answer 1

因为每个规则的“变量”、“值”和比较运算符都存储在数据库里，所以你可以写一个规则类，这个类会接收合适的参数（比如运算符、动作、值等等），然后返回一个可调用的对象。这个对象会接收所有相关的变量，以字典的形式传入，并执行相应的注册动作。

它的样子大概是这样的，不过你需要根据你的动作来适当调整参数的获取方式：

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 2

有几种方法可以实现这个目标。其他的回答也很有价值，我想再补充两种技巧。

• 如果你可以重写表格，可以把每个规则做成一个可以保存的函数，这样在需要的时候就可以重新加载它。
• 写一个大的字典，把规则当作键，把函数当作值。如果你最多有100个规则，这样管理起来是可以的。只要确保你的函数非常灵活，使用 *args 和 **kwargs 来处理不同数量的参数。

使用 pickle 的例子：

首先，创建一个对输入很灵活的函数。

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

然后 使用 pickle 来保存它：

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

当你需要恢复你的业务规则时：

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

有灵活输入的目的在于你可以接收任意数量的参数：

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

使用字典的例子

把所有函数存储在一个大的映射中：

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
}

然后当你需要它的时候：

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

Answer 3

如果你想发送邮件，可以使用email模块。

如果我是你，我会写一个简单的Python脚本，处理一些规则，这些规则可以简单地写在一个单独的文件里，然后根据需要发送邮件或短信等。

你可以使用像cron这样的服务，让这个脚本每天运行一次（或者其他你想要的频率）。

比如说，如果你的规则是这样的：

# 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, ....]

那么你的处理脚本可能会是这样的：

# 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) 

当然，还有其他方法可以做到这一点，比如创建一个Pythonic DSL来编写这些规则。