在数据库中存储逻辑

Question

这不是关于存储过程的问题（至少我认为不是）。

假设我有一个数据库，其中包含一些城市和在这些城市中举行的活动。有些人要使用这个网站，并希望在进入网站时得到有关某些活动的通知。
指定他们想要收到通知的事件的规则应该是通用的。

例如，我希望用户能够说“我想收到有关所有在周日举行、在1200年至1400年间建立的城市中发生的事件以及名称以字母“F”开头或位于南美洲国家的事件的通知”，这将被翻译为伪逻辑代码：

  (
  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"
  )

像“大陆是”，“日期是”，“成立日期介于”等规则应预先定义，用户将选择它们，但我希望能够在未来添加新的规则。如何存储这样的逻辑最佳？我能想到的唯一解决方案是“NotificationGatherer”模型。它将包含用户的ID和一个包含JSON的字符串。我会创建一个JSON二叉树，对于这种情况，大写的“AND”将是两个子节点 - 内部和内部或。第一个子节点将有两个简单条件，反映上述实际条件。然后我会有一个在用户请求时调用的方法，该方法将执行以下操作之一：1. 为所有即将发生的事件评估此条件集的值（true/false）；2. 创建一个查询集与过滤器一起使用，该过滤器将获取满足给定条件的所有即将发生的事件（更加复杂和高效）。现在，这是一个好方法吗，还是我应该尝试其他方法？它看起来相当复杂（我已经看到了测试它的痛苦），而且我可以想象许多人在过去都需要像这样的东西，但我找不到任何建议，因为搜索“数据库中的逻辑”会自动指向有关存储过程的文章/问题。如果有任何区别，我正在使用Django和MySQL。

- piezol

---

这就是异步队列和调度的任务。请查看Celery。 - petkostas

谢谢你的评论，我稍后会阅读有关Celery的文档，特别是因为我需要在我的应用程序中进行一些调度，但初步看来：这不是我问题的重点。通知只是我需要存储逻辑的例子之一。 - piezol

Answer 1

如果是我，我会把规则存储在数据库中，然后使用Celery定期处理它们。

对于模型部分，我认为多表继承是正确的选择，因为不同的规则需要存储不同的数据。在我看来，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

代码完全未经过测试，但我认为它最终可能会起作用，或者至少能够给您提供实现思路。希望这有所帮助！

- Agate

---

我曾经不得不编写一个ttw SQL查询构建器（希望是专业的 - 只有一小组相关表格，具有已知的模式），我的解决方案接近于这个。 - bruno desthuilliers

谢谢！我认为这里可能比较难的部分是前端 ;) - Agate

非常感谢，回复太棒了。乍一看，它看起来就像我需要的东西，如果没有更好的答案出现，我会将其标记为最佳答案。至于前端 - 对于我的示例，我试图简化我所需的内容，在实际应用程序中没有真正的通知，并且将是我实际上放置这些规则的人，因此我也可以编写一些脚本并忽略前端；）。我只是希望从一开始就尽可能灵活，而不需要每次需要另一种类型的规则时重写所有内容。 - piezol

哇，我刚意识到为了这个例子，我编造了一个全新的大陆。 - piezol

@piezol 我完全理解你的用例，事实上，我很快可能会遇到类似的情况，所以我可能会从这个示例中创建一个可重用的包。如果我这样做了，我会在这里分享它 :) - Agate

Answer 2

这不是关于数据库逻辑的问题，更应该称之为存储筛选模式或存储筛选偏好。

通常情况下，您希望让用户能够创建和存储在个人资料设置中的过滤器，以从数据库中提取所有匹配它们的事件并向用户发送有关它们的通知。

首先，您应该考虑这些过滤器需要多深。例如，可以像这样：

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

每个用户应该能够定义多个过滤器集。当查询创建时，所有过滤器将与AND连接在一起（除了筛选器内部的某些规则。该特定筛选器的类型将设置它）。

在创建一些类型的过滤器之后（例如活动开始的日期范围，每周的天数等），您将在一列中存储筛选器类型，并在其他列或使用json序列化在一列中存储筛选器参数。

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

这并不像在json中存储整个WHERE条件那样复杂，但它将提供类似的灵活性。您只需为用户创建多个FilterSet以覆盖一些复杂情况即可。

- GwynBleidD