Django：属性和queryset注释之间的重复逻辑

Question

当我想要定义我的业务逻辑时，我正在努力找到正确的方法来实现这一点，因为我通常都需要一个属性和一个自定义查询集来获得相同的信息。最后，逻辑被复制

让我解释一下……

首先，在定义类之后，我自然开始为我需要的数据编写一个简单的属性：

class PickupTimeSlot(models.Model):

    @property
    def nb_bookings(self) -> int:
        """ How many times this time slot is booked? """ 
        return self.order_set.validated().count()

然后，我很快意识到，在处理queryset中的许多对象时调用此属性会导致重复查询，并会降低性能（即使使用预取，因为过滤会再次调用）。因此，我解决了编写带有注释的自定义查询集的问题：

class PickupTimeSlotQuerySet(query.QuerySet):

    def add_nb_bookings_data(self):
        return self.annotate(db_nb_bookings=Count('order', filter=Q(order__status=Order.VALIDATED)))

问题

然后，我遇到了两个问题：

• 我有相同的业务逻辑（如何查找预订数量），写了两次，这可能会导致功能错误
• 我需要找到两个不同的属性名以避免冲突，因为显然，为属性和注释设置nb_bookings是不起作用的。这迫使我在使用对象时考虑如何生成数据，调用正确的属性名（比如pickup_slot.nb_bookings（属性）或pickup_slot.db_nb_bookings（注释））

这对我来说似乎设计得很糟糕，我很确定有办法做得更好。我需要一种总是使用相同的业务逻辑编写pickup_slot.nb_bookings并有一个性能良好的答案的方法

我有个主意，但我不确定…

我正在考虑完全删除该属性，只保留自定义查询集。然后，对于单个对象，将它们包装在QuerySet中，以便能够调用添加注释数据。比如：

pickup_slot = PickupTimeSlot.objects.add_nb_bookings_data().get(pk=pickup_slot.pk)

对我来说，这似乎很不自然。你觉得怎么样

Answer 1

TL；博士

• 是否需要过滤带注释的字段结果

  • 如果是，“保留”经理，并在需要时使用它。在任何其他情况下，使用属性逻辑
  • 如果没有，请删除管理器/注释过程并坚持使用属性实现，除非您的表很小（~1000个条目），并且在此期间没有增长

• 我在这里看到的注释过程的唯一优点是数据的数据库级过滤功能

---

我已经进行了一些测试以得出结论，如下所示

环境

• Django 3.0.7
• Python 3.8
• PostgreSQL 10.14

模型结构

为了简单和模拟，我遵循下面的模型表示

class ReporterManager(models.Manager):
    def article_count_qs(self):
        return self.get_queryset().annotate(
            annotate_article_count=models.Count('articles__id', distinct=True))


class Reporter(models.Model):
    objects = models.Manager()
    counter_manager = ReporterManager()
    name = models.CharField(max_length=30)

    @property
    def article_count(self):
        return self.articles.distinct().count()

    def __str__(self):
        return self.name


class Article(models.Model):
    headline = models.CharField(max_length=100)
    reporter = models.ForeignKey(Reporter, on_delete=models.CASCADE,
                                 related_name="articles")

    def __str__(self):
        return self.headline

我已经用随机字符串填充了我的数据库Reporter和Article模型

• 记者行~220K（220514）
• 文章行数~1M（997311）

测试用例

1. 随机选取Reporter实例并检索文章计数。我们通常在细节视图中执行此操作
2. 分页的结果。我们切片查询集，并在切片的查询集上迭代
3. 过滤

我正在使用Ipythonshell的^{}-(ipython doc)命令来计算执行时间

测试用例1

为此，我创建了这些函数，它们从数据库中随机选取实例

import random

MAX_REPORTER = 220514


def test_manager_random_picking():
    pos = random.randint(1, MAX_REPORTER)
    return Reporter.counter_manager.article_count_qs()[pos].annotate_article_count


def test_property_random_picking():
    pos = random.randint(1, MAX_REPORTER)
    return Reporter.objects.all()[pos].article_count

结果

In [2]: %timeit test_manager_random_picking()
8.78 s ± 6.1 s per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [3]: %timeit test_property_random_picking()
6.36 ms ± 221 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

测试用例2

我创建了另外两个函数

import random

PAGINATE_SIZE = 50


def test_manager_paginate_iteration():
    start = random.randint(1, MAX_REPORTER - PAGINATE_SIZE)
    end = start + PAGINATE_SIZE
    qs = Reporter.counter_manager.article_count_qs()[start:end]
    for reporter in qs:
        reporter.annotate_article_count


def test_property_paginate_iteration():
    start = random.randint(1, MAX_REPORTER - PAGINATE_SIZE)
    end = start + PAGINATE_SIZE
    qs = Reporter.objects.all()[start:end]
    for reporter in qs:
        reporter.article_count

结果

In [8]: %timeit test_manager_paginate_iteration()
4.99 s ± 312 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [9]: %timeit test_property_paginate_iteration()
47 ms ± 1.16 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)

测试用例3

毫无疑问，注释是唯一的方法

在这里您可以看到，与属性实现相比，注释过程花费了大量的时间

Answer 2

我不认为这里有什么灵丹妙药。但是我在我的项目中使用这种模式来处理这种情况

class PickupTimeSlotAnnotatedManager(models.Manager):
    def with_nb_bookings(self):
        return self.annotate(
            _nb_bookings=Count(
                'order', filter=Q(order__status=Order.VALIDATED)
            )
        )

class PickupTimeSlot(models.Model):
    ...
    annotated = PickupTimeSlotAnnotatedManager()

    @property
    def nb_bookings(self) -> int:
        """ How many times this time slot is booked? """ 
        if hasattr(self, '_nb_bookings'):
            return self._nb_bookings
        return self.order_set.validated().count()

编码

qs = PickupTimeSlot.annotated.with_nb_bookings()
for item in qs:
    print(item.nb_bookings)

通过这种方式，我可以始终使用属性，如果它是带注释查询集的一部分，它将使用带注释的值，如果不是，它将计算它。这种方法通过使用所需的值对queryset进行注释，保证我可以完全控制何时使其“更重”。如果我不需要这个，我就使用常规的PickupTimeSlot.objects. ...

此外，若有许多这样的属性，那个么您可以编写一个decorator来包装属性并简化代码。它将作为cached_property装饰器工作，但如果存在，它将使用带注释的值

Answer 3

为避免任何重复，一个选项可以是：

• 删除模型中的属性
• 使用自定义管理器
• 重写它的get_queryset（）方法：

class PickupTimeSlotManager(models.Manager):

    def get_queryset(self):
        return super().get_queryset().annotate(
            db_nb_bookings=Count(
                'order', filter=Q(order__status=Order.VALIDATED)
            )
        )

from django.db import models
from .managers import PickupTimeSlotManager

class PickupTimeSlot(models.Model):
    ...
    # Add custom manager
    objects = PickupTimeSlotManager()

优点：计算出的属性透明地添加到任何查询集中；使用它无需采取进一步行动

缺点：即使未使用计算属性，也会产生计算开销