django中业务逻辑和数据访问的分离

Z时代
2024-01-10
分类：问答

我正在Django中编写一个项目，我发现文件中有80％的代码models.py。这段代码令人困惑，并且在一段时间之后，我不再了解实际发生的事情。

这是困扰我的事情：

我发现模型级别（应该只负责处理数据库中的数据）在发送电子邮件，使用API到其他服务等方面也很丑陋。

另外，我发现在视图中放置业务逻辑也是不可接受的，因为这样很难控制。例如，在我的应用程序中，至少有三种方法来创建的新实例User，但从技术上讲，它应统一创建它们。

我并不总是注意到模型的方法和属性何时变得不确定，以及何时出现副作用。
这是一个简单的例子。首先，User模型是这样的：

class User(db.Models):
    def get_present_name(self):
        return self.name or 'Anonymous'
    def activate(self):
        self.status = 'activated'
        self.save()

随着时间的流逝，它变成了：

class User(db.Models):
    def get_present_name(self): 
        # property became non-deterministic in terms of database
        # data is taken from another service by api
        return remote_api.request_user_name(self.uid) or 'Anonymous' 
    def activate(self):
        # method now has a side effect (send message to user)
        self.status = 'activated'
        self.save()
        send_mail('Your account is activated!', '…', [self.email])

我想要的是在代码中分离实体：

数据库级别的数据库实体：什么包含我的应用程序？

我的应用程序的实体，业务逻辑级别：可以使我的应用程序做什么？
有什么好的实践来实现可以在Django中应用的方法？

回答：

似乎你在询问数据模型和域模型之间的区别–后者是你可以找到最终用户感知的业务逻辑和实体的地方，前者是你实际存储数据的地方。

此外，我将问题的第三部分解释为：如何注意到未能将这些模型分开的问题。

这是两个截然不同的概念，很难将它们分开。但是，有一些常见的模式和工具可用于此目的。

你需要认识的第一件事是你的域模型并不是真正的数据。它涉及诸如“激活此用户”，“停用此用户”，“当前已激活哪些用户”和“此用户的名字是什么”之类的动作和问题。用经典术语来说：它是关于查询和命令的。

让我们从示例中的命令开始：“激活此用户”和“禁用此用户”。关于命令的好处是，它们可以很容易地用小给定的情况来表示：

given an inactive user

when the admin activates this user

then the user becomes active

and a confirmation e-mail is sent to the user

and an entry is added to the system log

(etc. etc.)

这种情况对于查看单个命令如何影响基础结构的不同部分很有用，在这种情况下，数据库（某种“活动”标志），邮件服务器，系统日志等会受到影响。

这样的场景也确实可以帮助你设置测试驱动开发环境。

最后，思考命令确实可以帮助你创建面向任务的应用程序。你的用户将不胜感激:-)

Django提供了两种简单的表达命令的方式：它们都是有效的选择，并且将两种方法混合使用并不罕见。

该服务模块已经通过@Hedde描述。在这里，你定义了一个单独的模块，每个命令都表示为一个函数。

def activate_user(user_id):
    user = User.objects.get(pk=user_id)
    # set active flag
    user.active = True
    user.save()
    # mail user
    send_mail(...)
    # etc etc

另一种方法是为每个命令使用Django表单。我更喜欢这种方法，因为它结合了多个紧密相关的方面：

命令的执行（它做什么？）

验证命令参数（可以执行此操作吗？）

命令演示（如何执行此操作？）

class ActivateUserForm(forms.Form):
    user_id = IntegerField(widget = UsernameSelectWidget, verbose_name="Select a user to activate")
    # the username select widget is not a standard Django widget, I just made it up
    def clean_user_id(self):
        user_id = self.cleaned_data['user_id']
        if User.objects.get(pk=user_id).active:
            raise ValidationError("This user cannot be activated")
        # you can also check authorizations etc. 
        return user_id
    def execute(self):
        """
        This is not a standard method in the forms API; it is intended to replace the 
        'extract-data-from-form-in-view-and-do-stuff' pattern by a more testable pattern. 
        """
        user_id = self.cleaned_data['user_id']
        user = User.objects.get(pk=user_id)
        # set active flag
        user.active = True
        user.save()
        # mail user
        send_mail(...)
        # etc etc

Thinking in Queries

你的示例不包含任何查询，因此我自由地编写了一些有用的查询。我更喜欢使用“问题”一词，但是查询是经典的术语。有趣的查询是：“此用户的名字是什么？”，“该用户可以登录吗？”，“向我显示停用用户的列表”和“停用用户的地理分布是什么？”。

在着手回答这些查询之前，你应该始终问自己两个问题：这是仅针对我的模板的表示性查询，和/或与执行命令相关的业务逻辑查询，和/或报告查询。

呈现查询只是为了改善用户界面。业务逻辑查询的答案直接影响命令的执行。报告查询仅用于分析目的，并且具有较宽松的时间限制。这些类别不是互相排斥的。

另一个问题是：“我是否完全控制答案？” 例如，在查询用户名时（在这种情况下），我们对结果没有任何控制权，因为我们依赖于外部API。

Django中最基本的查询是使用Manager对象：

User.objects.filter(active=True)

当然，这仅在数据实际在数据模型中表示时才有效。这并非总是如此。在这种情况下，你可以考虑以下选项。

自定义标签和过滤器

第一种替代方法仅对表示性查询有用：自定义标记和模板过滤器。

<h1>Welcome, {{ user|friendly_name }}</h1>

@register.filter
def friendly_name(user):
    return remote_api.get_cached_name(user.id)

如果你的查询不只是表示形式的查询，则可以将查询添加到你的services.py（如果正在使用的话），或者引入querys.py模块：

def inactive_users():
    return User.objects.filter(active=False)
def users_called_publysher():
    for user in User.objects.all():
        if remote_api.get_cached_name(user.id) == "publysher":
            yield user

代理模型在业务逻辑和报告的上下文中非常有用。你基本上可以定义模型的增强子集。你可以通过覆盖Manager.get_queryset()方法来覆盖Manager的基础QuerySet 。

class InactiveUserManager(models.Manager):
    def get_queryset(self):
        query_set = super(InactiveUserManager, self).get_queryset()
        return query_set.filter(active=False)
class InactiveUser(User):
    """
    >>> for user in InactiveUser.objects.all():
    …        assert user.active is False 
    """
    objects = InactiveUserManager()
    class Meta:
        proxy = True

对于本质上很复杂但经常执行的查询，存在查询模型的可能性。查询模型是非规范化的一种形式，其中单个查询的相关数据存储在单独的模型中。当然，技巧是使非规范化模型与主模型保持同步。仅当更改完全在你的控制之下时才能使用查询模型。

class InactiveUserDistribution(models.Model):
    country = CharField(max_length=200)
    inactive_user_count = IntegerField(default=0)

第一种选择是在命令中更新这些模型。如果仅通过一个或两个命令更改这些模型，这将非常有用。

class ActivateUserForm(forms.Form):
    # see above
    def execute(self):
        # see above
        query_model = InactiveUserDistribution.objects.get_or_create(country=user.country)
        query_model.inactive_user_count -= 1
        query_model.save()

更好的选择是使用自定义信号。这些信号当然是由你的命令发出的。信号的优势在于你可以使多个查询模型与原始模型保持同步。此外，可以使用Celery或类似框架将信号处理任务转移给后台任务。

user_activated = Signal(providing_args = ['user'])
user_deactivated = Signal(providing_args = ['user'])

class ActivateUserForm(forms.Form):
    # see above
    def execute(self):
        # see above
        user_activated.send_robust(sender=self, user=user)

models.py

class InactiveUserDistribution(models.Model):
    # see above
@receiver(user_activated)
def on_user_activated(sender, **kwargs):
        user = kwargs['user']
        query_model = InactiveUserDistribution.objects.get_or_create(country=user.country)
        query_model.inactive_user_count -= 1
        query_model.save()