在数据库中存储逻辑

Question

这与存储过程无关（或者至少我认为不是）。

假设我有一个数据库。数据库中包含一些城市以及在这些城市中发生的事件。有些人将要使用该网站，他们希望在进入该网站时收到有关某些事件的通知。
指定他们想要通知哪些事件的规则应该是通用的。

例如，我希望用户能够说 "I want to be notified about all events, that will take place on sunday, in city that was founded between year 1200 and 1400, in the country which name starts with the letter "F" or which is in South America"，这将翻译成 pseudo-logic-code of:

  (
  event.date.day == Sunday 
  and 
  event.city.founded_date.year.between(1200, 1400)
  ) 
AND 
  (
  event.city.country.starts_with("F")
  or
  event.city.country.continent == "South Africa"
  )

"continent is"、"day is"、"foundation date is between"等规则是预定义的，用户会选择它，但我希望将来能够添加新规则。

存储这种逻辑的最佳方式是什么？我能想出的唯一解决方案是 "NotificationGatherer" 模型。它将包含一个用户的 ID 和一个带有 json 的字符串。我会创建一个 json 二叉树，对于这种特殊情况，大写的 "AND" 将是一个具有两个 children 的根 - 内部和内部或。第一个children会有两个简单的条件，这将反映实际情况如上。

然后我会有一个根据用户请求调用的方法，它可以是：

评估为所有即将发生的事件设置的此条件的值 (true/false)

或
形成一个带有过滤器的查询集，它将获取所有满足给定条件的即将发生的事件（难度更大，效率更高）。

现在，这是一个好方法吗，还是我应该尝试其他方法？它看起来相当复杂（我已经看到测试它会有多痛苦），我可以想象过去很多人可能需要这样的东西，但我找不到任何建议，因为任何搜索"logic in database" 自动将我指向 articles/questions 关于存储过程。

如果这有什么不同，我正在使用 django 和 mysql。

Answer 1

这与数据库中的逻辑无关，最好称为存储过滤器模式或存储过滤器首选项。

一般来说，您希望让您的用户能够创建并存储在配置文件设置过滤器中，这些过滤器将从数据库中提取与他们匹配的所有事件并向用户发送有关它们的通知。

首先，您应该考虑过滤器的深度。例如，它可以是这样的：

模型FilterSet - 将有一些全局设置（例如通知类型）并将分配给特定用户
模型 Filter - 将具有一个过滤规则（或组合在一起的一组规则，例如日期范围）并将分配给 FilterSet

每个用户应该能够定义多个过滤器集。创建查询时，所有过滤器都将与 AND 连接在一起（过滤器内的某些规则除外。该特定过滤器的类型将对其进行设置）。

创建某些类型的过滤器（甚至开始的日期范围、星期几等）后，您将在一列中存储过滤器类型，在其他列中或在使用 json 序列化的一列中存储过滤器参数。

当应该发送通知时，处理器将检查每个 FilterSet 是否正在返回一些数据，如果是，它会将返回的数据发送给该 FilterSet 的所有者。

它不像在 json 中存储整个 WHERE 条件那么复杂，但它会提供类似的灵活性。您只需为用户创建多个 FilterSet 来涵盖一些复杂的案例。

Answer 2

如果是我的话，我会把规则存入数据库，然后不时使用 Celery 处理它们。

对于模型部分，我认为多table继承是可行的方法，因为不同的规则需要存储不同的数据。在我看来，django-polymorphic 是你的朋友：

我建议如下：

from django.db import models
from polymorphic import PolymorphicModel


class AbtractRuleObject(models.Model):
    class Meta:
        abstract = True

    def filter_queryset(self, queryset):
        """Will handle actual filtering of the event queryset"""
        raise NotImplementedError

    def match_instance(self, instance):
        raise NotImplementedError

class RuleSet(AbtractRuleObject):
    """Will manage the painful part o handling the OR / AND logic inside the database"""
    NATURE_CHOICES = (
        ('or', 'OR'),
        ('and', 'AND'),
    )
    nature = models.CharField(max_length=5, choices=NATURE_CHOICES, default='and')

    # since a set can belong to another set, etc.
    parent_set = models.ForeignKey('self', null=True, blank=True, related_name='children')

    def filter_queryset(self, queryset):
        """This is rather naive and could be optimized"""
        if not self.parent_set:
            # this is a root rule set so we just filter according to registered rules
            for rule in self.rules:
                if self.nature == 'and':
                    queryset = rule.filter_queryset(queryset)
                elif self.nature == 'or':
                    queryset = queryset | rule.filter_queryset(queryset)
        else:
            # it has children rules set
            for rule_set in self.children:
                if self.nature == 'and':
                    queryset = rule_set.filter_queryset(queryset)
                elif self.nature == 'or':
                    queryset = queryset | rule_set.filter_queryset(queryset)
        return queryset

    def match_instance(self, instance):
        if not self.parent_set:
            if self.nature == 'and':
                return all([rule_set.match_instance(instance) for rule_set in self.children])
            if self.nature == 'any':
                return any([rule_set.match_instance(instance) for rule_set in self.children])
        else:
            if self.nature == 'and':
                return all([rule_set.match_instance(instance) for rule_set in self.children])
            if self.nature == 'any':
                return any([rule_set.match_instance(instance) for rule_set in self.children])

class Rule(AbtractRuleObject, PolymorphicModel):
    """Base class for all rules"""
    attribute = models.CharField(help_text="Attribute of the model on which the rule will apply")
    rule_set = models.ForeignKey(RuleSet, related_name='rules')

class DateRangeRule(Rule):
    start = models.DateField(null=True, blank=True)
    end = models.DateField(null=True, blank=True)

    def filter_queryset(self, queryset):
        filters = {}
        if self.start:
            filters['{0}__gte'.format(self.attribute)] = self.start
        if self.end:
            filters['{0}__lte'.format(self.attribute)] = self.end
        return queryset.filter(**filters)

    def match_instance(self, instance):
        start_ok = True
        end_ok = True
        if self.start:
            start_ok = getattr(instance, self.attribute) >= self.start
        if self.end:
            end_ok = getattr(instance, self.attribute) <= self.end

        return start_ok and end_ok

class MatchStringRule(Rule):
    match = models.CharField()
    def filter_queryset(self, queryset):
        filters = {'{0}'.format(self.attribute): self.match}
        return queryset.filter(**filters)

    def match_instance(self, instance):
        return getattr(instance, self.attribute) == self.match

class StartsWithRule(Rule):
    start = models.CharField()

    def filter_queryset(self, queryset):
        filters = {'{0}__startswith'.format(self.attribute): self.start}
        return queryset.filter(**filters)

    def match_instance(self, instance):
        return getattr(instance, self.attribute).startswith(self.start)

现在，假设您的 Event 和 City 模型如下所示：

class Country(models.Model):
    continent = models.CharField()
    name = models.CharField(unique=True)

class City(models.Model):
    name = models.CharField(unique=True)
    country = models.ForeignKey(Country)
    founded_date = models.DateField()

class Event(models.Model):
    name = models.CharField(unique=True)
    city = models.ForeignKey(City)
    start = models.DateField()
    end = models.DateField()

那么你可以使用我的例子如下：

global_set = RuleSet(nature='and')
global_set.save()

set1 = RuleSet(nature='and', parent_set=global_set)
set1.save()

year_range = DateRangeRule(start=datetime.date(1200, 1, 1),
                           end=datetime.date(1400, 1, 1),
                           attribute='city__founded_date',
                           rule_set=set1)
year_range.save()

set2 = RuleSet(nature='or', parent_set=global_set)
set2.save()

startswith_f = StartsWithRule(start='F',
                              attribute='city__country__name')
                              rule_set=set2)
startswith_f.save()

exact_match = MatchStringRule(match='South Africa',
                              attribute='city__country__continent')
                              rule_set=set2)
exact_match.save()

queryset = Event.objects.all()

# Magic happens here

# Get all instances corresponding to the rules
filtered_queryset = global_set.filter_queryset(queryset)

# Check if a specific instance match the rules
assert global_set.match_instance(filtered_queryset[0]) == True

代码绝对未经测试，但我认为它最终可以工作，或者至少给你实现思路。

希望对您有所帮助！

在数据库中存储逻辑

Storing logic inside database

python

database

django

logic

database-design