Django：属性和查询集注释之间的重复逻辑
Django: Duplicated logic between properties and queryset annotations
当我想定义我的业务逻辑时，我很难找到正确的方法来执行此操作，因为我经常需要 属性 和自定义查询集来获取相同的信息。最后，逻辑重复了。
让我解释一下...
首先，在定义了我的class之后，我很自然地开始为我需要的数据写一个简单的属性：
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)))
问题
然后，我遇到了 2 个问题：
- 我将相同的业务逻辑（“如何查找预订数量”）写了两次，这可能会导致功能错误。
- 我需要找到两个不同的属性名称以避免冲突，因为显然，为 属性 和注释设置
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)
对我来说似乎很老套和不自然。你怎么看？
为避免任何重复，一种选择可能是：
- 删除模型中的属性
- 使用自定义管理器
- 覆盖它的 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()
优点：计算的属性透明地添加到任何查询集中；无需进一步操作即可使用它
缺点：即使不使用计算的属性也会产生计算开销
让它成为您存档所需内容的替代方式：
因为我通常在每次编写查询集时都添加 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 的其他信息。如果我们无论如何都要预取它，那么执行第一种方法不会花费更多。
希望这或多或少对你有所帮助。祝你有美好的一天！
我认为这里没有灵丹妙药。但是我在我的项目中使用这种模式来处理这种情况。
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 装饰器，但如果它存在，它将使用注释值。
TL;DR
是否需要过滤 “注释字段” 结果？
- 如果是，“保留” 经理并在需要时使用它。在任何其他情况中，使用属性逻辑
- 如果否，删除 manager/annotation 流程并坚持 属性 实施，除非您的 table 很小（~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 个实例并检索 文章数 。我们通常在详细视图 中执行此操作
- 分页结果。我们切片 查询集并迭代 切片查询集。
- 过滤
我正在使用Ipythonshell的%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
毫无疑问，注释是这里唯一的方法
在这里您可以看到，与 属性 实施相比，注释 过程花费了大量时间。
根据您不同的好答案，我决定坚持使用注释 和 属性。我创建了一个缓存机制，使命名透明。主要优点是将业务逻辑仅保存在一个地方。
我看到的唯一缺点是可以从数据库中第二次调用对象以进行注释。 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
decorators.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
当我想定义我的业务逻辑时，我很难找到正确的方法来执行此操作，因为我经常需要 属性 和自定义查询集来获取相同的信息。最后，逻辑重复了。
让我解释一下...
首先，在定义了我的class之后，我很自然地开始为我需要的数据写一个简单的属性：
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)))
问题
然后，我遇到了 2 个问题：
- 我将相同的业务逻辑（“如何查找预订数量”）写了两次，这可能会导致功能错误。
- 我需要找到两个不同的属性名称以避免冲突，因为显然，为 属性 和注释设置
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)
对我来说似乎很老套和不自然。你怎么看？
为避免任何重复，一种选择可能是：
- 删除模型中的属性
- 使用自定义管理器
- 覆盖它的 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()
优点：计算的属性透明地添加到任何查询集中；无需进一步操作即可使用它
缺点：即使不使用计算的属性也会产生计算开销
让它成为您存档所需内容的替代方式：
因为我通常在每次编写查询集时都添加 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 的其他信息。如果我们无论如何都要预取它，那么执行第一种方法不会花费更多。
希望这或多或少对你有所帮助。祝你有美好的一天！
我认为这里没有灵丹妙药。但是我在我的项目中使用这种模式来处理这种情况。
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 装饰器，但如果它存在，它将使用注释值。
TL;DR
是否需要过滤 “注释字段” 结果？
- 如果是，“保留” 经理并在需要时使用它。在任何其他情况中，使用属性逻辑
- 如果否，删除 manager/annotation 流程并坚持 属性 实施，除非您的 table 很小（~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个实例并检索 文章数 。我们通常在详细视图 中执行此操作
- 分页结果。我们切片 查询集并迭代 切片查询集。
- 过滤
我正在使用Ipythonshell的%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
毫无疑问，注释是这里唯一的方法
在这里您可以看到，与 属性 实施相比，注释 过程花费了大量时间。
根据您不同的好答案，我决定坚持使用注释 和 属性。我创建了一个缓存机制，使命名透明。主要优点是将业务逻辑仅保存在一个地方。 我看到的唯一缺点是可以从数据库中第二次调用对象以进行注释。 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
decorators.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