Django 查询集在多次使用和过滤后停止工作

Question

我目前正在执行一项任务，要求我根据条件将模型数据分为多个部分，然后对它们进行不同的处理。我能够正确从模型中获取数据并检查它们存在的所有可用详细信息。但是在应用一定数量的过滤器后，查询集突然停止工作并且没有给出正确的结果。我不知道为什么会这样，有人能解释一下吗？这是导致问题的我的代码的一部分：

def test_selection():
    po_numbers = (
        Model1.objects
        .filter(schedule__isnull=True)
        .filter(order_hold_type__isnull=True)
        .filter(so_line_status='Awaiting Supply')
        .values_list("customer_po", flat=True)
        .distinct()
    )
    rows = Model1.objects.filter(customer_po__in=po_numbers).annotate(
            total_qty=Sum(F('line_item_quantity')*F('cup_count'))
        ).all()
    
    old_sizes = Sizes.get_old_sizes()
    old_po = rows.filter(
        item_size__in=old_sizes).values_list('customer_po', flat=True).distinct()
    rows = rows.filter(~Q(customer_po__in=old_po)).all()

    partially_scheduled = rows.filter(
        customer_po__in=list(rows.filter(schedule__isnull=False).values_list(
            'customer_po', flat=True).distinct())
    ).all()

    unscheduled = rows.filter(
        ~Q(customer_po__in=list(partially_scheduled.values_list(
            'customer_po', flat=True).distinct().all()))
    ).all()

    model2_po = Model2.objects.values_list('customer_po', flat=True).distinct()
    g1 = unscheduled.exclude(customer_po__in=model2_po)
    po_list = g1.values_list('customer_po', flat=True).distinct()
    print(rows.count(), po_list)

我认为问题出在我的流程中，所以我创建了一个单独的函数来测试它，这就是问题开始发生的点。这不会引发任何错误，但会出现空的情况，我尝试搜索任何有类似问题的人，但找不到任何内容。到目前为止我已经尝试过以下操作：

• 使用filter(~Q())而不是exclusion来过滤数据
• 使用 all() 因此它会创建查询集的副本，而不是处理旧的查询集
• 将所有项目作为列表进行过滤，而不是直接使用查询集

但我仍然无法找到这里可能发生的问题。在代码的最后 3 行中，我得到 g1.count() 作为 546 行数据，但 PO 为空。但如果我直接使用

unscheduled

并删除 g1 行，那么它会在列表中显示 43 个值。

    g1 = unscheduled.exclude(customer_po__in=model2_po)
    po_list = unscheduled.exclude(customer_po__in=model2_po).values_list('customer_po', flat=True).distinct()

这给出了 g1.count() = 546，po_list = 空。

    # g1 = unscheduled.exclude(customer_po__in=model2_po) #commented code
    po_list = unscheduled.exclude(customer_po__in=model2_po).values_list('customer_po', flat=True).distinct()

这给出了 po_list = 43 个项目列表的查询集。

Answer 1

避免不必要地调用 .all()。

优化并验证中间结果。

谨慎对待查询集的列表转换“如果行有大量记录，list(rows.filter(...).values_list(...)) 可能会出现问题，因为将查询集转换为列表会立即对其进行评估。它是最好尽可能直接使用查询集链接。”

打印或记录中间结果以验证正确性。例如，检查每个过滤步骤的结果，以确保它们返回您期望的结果。

这是代码的修改版本，其中包含改进和调试步骤：

def test_selection(): po_numbers = ( Model1.对象 .filter(schedule__isnull=True, order_hold_type__isnull=True, so_line_status='等待供应') .values_list("customer_po", flat=True) 。清楚的（） )

rows = Model1.objects.filter(customer_po__in=po_numbers).annotate(
    total_qty=Sum(F('line_item_quantity') * F('cup_count'))
)

old_sizes = Sizes.get_old_sizes()
old_po = rows.filter(item_size__in=old_sizes).values_list('customer_po', flat=True).distinct()
rows = rows.exclude(customer_po__in=old_po)

partially_scheduled = rows.filter(schedule__isnull=False).values_list('customer_po', flat=True).distinct()
unscheduled = rows.exclude(customer_po__in=partially_scheduled)

model2_po = Model2.objects.values_list('customer_po', flat=True).distinct()
g1 = unscheduled.exclude(customer_po__in=model2_po)

po_list = g1.values_list('customer_po', flat=True).distinct()

print(rows.count(), len(po_list))