根据外键字段之一的聚合过滤 Django 中的查询集?
Filter a query set in Django based on aggregate of one of its fields for a foreign key?
考虑以下两个模型,Worker
和 Invoice
:
class Worker(models.Model):
name = models.CharField(max_length=255)
class Invoice(models.Model):
worker = models.ForeignKey(
'Worker', on_delete=models.CASCADE)
amount = models.DecimalField(max_digits=10, decimal_places=2)
我只想对给定 Worker
的总计(amount
的总和)大于零的那些 Invoice
执行查询。
基本上,我想定义一个 get_payable_invoices()
函数,其中 returns 一个 Queryset
使得测试通过:
from decimal import Decimal
from django.test import TestCase
from django.db.models import Sum
from myapp.models import Worker, Invoice
def get_payable_invoices():
return Invoice.objects.filter(
worker__in=Worker.objects.annotate(Sum('invoice__amount')))\
.filter(invoice__amount__sum__gt=0)
class PayableInvoicesTests(TestCase):
def test_get_payable_invoices(self):
worker1 = Worker.objects.create(name="John Doe")
invoice1 = Invoice.objects.create(
worker=worker1, amount=Decimal('100.00'))
invoice2 = Invoice.objects.create(
worker=worker1, amount=Decimal('-150.00'))
worker2 = Worker.objects.create(name="Mary Contrary")
invoice3 = Invoice.objects.create(
worker=worker2, amount=Decimal('200.00'))
self.assertEqual(get_payable_invoices().count(), 1)
self.assertEqual(get_payable_invoices().first(), invoice3)
不过,当前的实现不起作用,returns a
django.core.exceptions.FieldError: Cannot resolve keyword 'invoice' into field. Choices are: amount, id, worker, worker_id
看来虽然遍历查询集时返回的对象确实有invoice__amount__sum
属性,但是在filter()
中不能这样使用
在我看来,我应该按照 https://docs.djangoproject.com/en/2.2/ref/models/expressions/#using-aggregates-within-a-subquery-expression 中的行来制定查询,但我正在努力使该示例适合我的,因为 total_comments
returns一个数字,而我想要一个 Worker
的列表。我也不完全确定子查询是否是正确的方法,或者是否可以在没有子查询的情况下以更简单的方式完成。关于如何在 Django 中实现这样的查询有什么想法吗?
事实证明,从 https://docs.djangoproject.com/en/2.2/topics/db/aggregation/#filtering-on-annotations 开始,在对注释进行过滤时,您需要使用不同于 'disambiguate' 的默认名称的名称。以下函数使测试通过:
def get_payable_invoices():
return Invoice.objects.filter(
worker__in=Worker.objects
.annotate(invoice_total=Sum('invoice__amount'))
.filter(invoice_total__gt=0))
我还验证了一个查询已执行。例如,我可以在单元测试的底部添加以下内容:
with self.assertNumQueries(1):
for invoice in get_payable_invoices():
pass
考虑以下两个模型,Worker
和 Invoice
:
class Worker(models.Model):
name = models.CharField(max_length=255)
class Invoice(models.Model):
worker = models.ForeignKey(
'Worker', on_delete=models.CASCADE)
amount = models.DecimalField(max_digits=10, decimal_places=2)
我只想对给定 Worker
的总计(amount
的总和)大于零的那些 Invoice
执行查询。
基本上,我想定义一个 get_payable_invoices()
函数,其中 returns 一个 Queryset
使得测试通过:
from decimal import Decimal
from django.test import TestCase
from django.db.models import Sum
from myapp.models import Worker, Invoice
def get_payable_invoices():
return Invoice.objects.filter(
worker__in=Worker.objects.annotate(Sum('invoice__amount')))\
.filter(invoice__amount__sum__gt=0)
class PayableInvoicesTests(TestCase):
def test_get_payable_invoices(self):
worker1 = Worker.objects.create(name="John Doe")
invoice1 = Invoice.objects.create(
worker=worker1, amount=Decimal('100.00'))
invoice2 = Invoice.objects.create(
worker=worker1, amount=Decimal('-150.00'))
worker2 = Worker.objects.create(name="Mary Contrary")
invoice3 = Invoice.objects.create(
worker=worker2, amount=Decimal('200.00'))
self.assertEqual(get_payable_invoices().count(), 1)
self.assertEqual(get_payable_invoices().first(), invoice3)
不过,当前的实现不起作用,returns a
django.core.exceptions.FieldError: Cannot resolve keyword 'invoice' into field. Choices are: amount, id, worker, worker_id
看来虽然遍历查询集时返回的对象确实有invoice__amount__sum
属性,但是在filter()
中不能这样使用
在我看来,我应该按照 https://docs.djangoproject.com/en/2.2/ref/models/expressions/#using-aggregates-within-a-subquery-expression 中的行来制定查询,但我正在努力使该示例适合我的,因为 total_comments
returns一个数字,而我想要一个 Worker
的列表。我也不完全确定子查询是否是正确的方法,或者是否可以在没有子查询的情况下以更简单的方式完成。关于如何在 Django 中实现这样的查询有什么想法吗?
事实证明,从 https://docs.djangoproject.com/en/2.2/topics/db/aggregation/#filtering-on-annotations 开始,在对注释进行过滤时,您需要使用不同于 'disambiguate' 的默认名称的名称。以下函数使测试通过:
def get_payable_invoices():
return Invoice.objects.filter(
worker__in=Worker.objects
.annotate(invoice_total=Sum('invoice__amount'))
.filter(invoice_total__gt=0))
我还验证了一个查询已执行。例如,我可以在单元测试的底部添加以下内容:
with self.assertNumQueries(1):
for invoice in get_payable_invoices():
pass