根据外键字段之一的聚合过滤 Django 中的查询集?

Filter a query set in Django based on aggregate of one of its fields for a foreign key?

考虑以下两个模型,WorkerInvoice

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