Django：属性和查询集注释之间的重复逻辑答案

【问题标题】：Django: Duplicated logic between properties and queryset annotationsDjango：属性和查询集注释之间的重复逻辑
【发布时间】：2021-01-23 07:06:23
【问题描述】：

当我想定义我的业务逻辑时，我正在努力寻找正确的方法来执行此操作，因为我经常需要一个属性和一个自定义查询集来获取相同的信息。到头来，逻辑是重复的。

让我解释一下……

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

class PickupTimeSlot(models.Model):

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

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

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 并获得高效的答案，始终使用相同的业务逻辑。

我有一个想法，但我不确定......

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

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

对我来说似乎很老套和不自然。你怎么看？

【问题讨论】：

好问题。 Django项目相关票证：code.djangoproject.com/ticket/28822.
有时property 的实现更容易当你的数据库中有一个大数据集（比如 1M），因为此时，注释会变慢，因为你正在执行超过 1M 记录的查询.
恕我直言，只要它们返回相同的结果，就将这两个逻辑保留在您的代码中。让我们等待“官方”实现。
只是问：这不是为了获得特定的记录吗？ pickup_slot = PickupTimeSlot.objects.filter(pk=pickup_slot.pk).add_nb_bookings_data()[0]我想注释只会对一条记录进行，对吧？

标签： python django django-models django-queryset

【解决方案1】：

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

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)

这样我总是可以使用属性，如果它是带注释的查询集的一部分，它将使用带注释的值，否则它将计算它。这种方法保证我将完全控制何时通过使用所需值注释查询集来使查询集“更重”。如果我不需要这个，我只需使用普通的PickupTimeSlot.objects. ...

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

【讨论】：

谢谢，对我来说听起来像是一个可以接受的妥协。一个思路：在属性中，你觉得返回PickupTimeSlot.objects.with_nb_bookings().get(pk=self.pk)._nb_bookings避免逻辑重复怎么样？（而不是self.order_set.validated().count()）
你可以这样做，但你已经有了那个对象，通过从带有注释数据的 db 重新获取相同的对象来获得nb_bookings 对我个人来说是“臭”的。但这是主观的。
你说得对，它很丑。但是我认为我几乎总能设法在我的对象上的get() 之前调用with_nb_bookings()，所以这种情况不应该经常发生。但我想在那种情况下我根本不需要财产。这将解决双重名称的问题（_nb_bookings VS nb_booking）
是的，在这种情况下您不需要属性，但是如果出现属性错误，可能会让您的代码的未来读者感到困惑，可能需要一些时间才能找出 @987654332 的位置@ 来自（哪里。无论哪种方式，目前都没有很好的解决方案，所以做对您和您的项目更有效的方法。
我发布了一个答案，主要是基于你的，包含你谈到的装饰器。如果您发现任何改进想法，请随时发表评论。再次感谢您的宝贵时间。

【解决方案2】：

为避免任何重复，一种选择可能是：

移除模型中的属性
使用自定义管理器
覆盖它的 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()

优势：计算的属性透明地添加到任何查询集；无需进一步操作即可使用它

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

【讨论】：

不建议将注释与 default 管理器一起放置。 AFAIK，每当您使用.（点）运算符调用对象时，Django 都会执行一个注释每次
@ArakkalAbu 你的意思是同一个注释会被多次添加到查询集中吗？为什么？
[我还没有做测试] 我认为这是 N+1 的问题。只是为了澄清，“通过使用点运算符访问 相关对象”@Mario Orlandi
我怀疑相反：在单个模型中使用属性时，您可能更可能遇到 N+1 问题，然后在注释整个查询集时。这可能还取决于 ORM 在 SQL 级别表达注释的智能程度；拥有完整的模型代码，按如下方式检查它会很有趣： print(str(queryset.query))
你是对的。 property 实现会解决N+1 问题，但是注释不会。

【解决方案3】：

TL;DR

是否需要过滤“带注释的字段”结果？
- 如果是，“保留”经理并在需要时使用它。在任何其他情况，使用属性逻辑
- 如果否，请删除管理器/注释流程并坚持实施属性，除非您的表很小（约 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)

测试用例

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

我正在使用Ipython shell 的%timeit-(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

毫无疑问，注解是这里的唯一途径

您可以在此处看到，与属性实现相比，注释过程需要大量时间。

【讨论】：

【解决方案4】：

让这成为归档您想要的内容的替代方式：

因为我通常在每次编写查询集时添加prefetch_related。所以当我遇到这个问题的时候，我会用Python来解决这个问题。

我将使用 Python 为我循环和统计数据，而不是使用 SQL 方式。

class PickupTimeSlot(models.Model):

    @property
    def nb_bookings(self) -> int:
        """ How many times this time slot is booked? """ 
        orders = self.order_set.all()  # this won't hit the database if you already did the prefetch_related
        validated_orders = filter(lambda x: x.status == Order.VALIDATED, orders)
        return len(validated_orders)

最重要的是，prefetch_related：

time_slots = PickupTimeSlot.objects.prefetch_related('order_set').all()

您可能有一个问题，为什么我没有 prefetch_related 使用过滤查询集，所以 Python 不需要再次过滤，例如：

time_slots = PickupTimeSlot.objects.prefetch_related(
    Prefetch('order_set', queryset=Order.objects.filter(status=Order.VALIDATED))
).all()

答案是有时我们还需要来自orders 的其他信息。如果我们无论如何都要预取它，那么执行第一种方式不会花费更多。

希望这或多或少对您有所帮助。祝你有美好的一天！

【讨论】：

谢谢，这确实是防止再次过滤订单的好主意。如果我错了，请纠正我，但我认为这个解决方案似乎是非常特定于上下文的，因为 Python filter() 方法可以被多次调用，例如如果有很多 PickupTimeSlot 对象。对于这些对象中的每一个，它将重新计算 valid_orders（始终相同）。
@DavidD。你是对的。也许我通常在序列化程序而不是模型中使用它。所以它只会被过滤一次。好话顺便说一句:)

【解决方案5】：

根据您不同的好答案，我决定坚持使用注释和属性。我创建了一个缓存机制以使其命名透明。主要优点是将业务逻辑仅保留在一个地方。我看到的唯一缺点是可以第二次从数据库调用对象以进行注释。 IMO 对性能的影响仍然很小。

这是一个完整的示例，其中包含我在模型中需要的 3 个不同属性。请随时发表评论以改进这一点。

models.py

class PickupTimeSlotQuerySet(query.QuerySet):

    def add_booking_data(self):
        return self \
            .prefetch_related('order_set') \
            .annotate(_nb_bookings=Count('order', filter=Q(order__status=Order.VALIDATED))) \
            .annotate(_nb_available_bookings=F('nb_max_bookings') - F('_nb_bookings')) \
            .annotate(_is_bookable=Case(When(_nb_bookings__lt=F('nb_max_bookings'),
                                             then=Value(True)),
                                        default=Value(False),
                                        output_field=BooleanField())
                      ) \
            .order_by('start')

class PickupTimeSlot(models.Model):
    objects = SafeDeleteManager.from_queryset(PickupTimeSlotQuerySet)()
   
    nb_max_bookings = models.PositiveSmallIntegerField()
    
    @annotate_to_property('add_booking_data', 'nb_bookings')
    def nb_bookings(self):
        pass
    
    @annotate_to_property('add_booking_data', 'nb_available_bookings')
    def nb_available_bookings(self):
        pass
    
    @annotate_to_property('add_booking_data', 'is_bookable')
    def is_bookable(self):
        pass

装饰器.py

def annotate_to_property(queryset_method_name, key_name):
    """
    allow an annotated attribute to be used as property.
    """
    from django.apps import apps

    def decorator(func):
        def inner(self):
            attr = "_" + key_name
            if not hasattr(self, attr):
                klass = apps.get_model(self._meta.app_label,
                                       self._meta.object_name)
                to_eval = f"klass.objects.{queryset_method_name}().get(pk={self.pk}).{attr}"
                value = eval(to_eval, {'klass': klass})
                setattr(self, attr, value)

            return getattr(self, attr)

        return property(inner)

    return decorator

【讨论】：