Django同一QuerySet的filter()查询优化方案问询

问题背景 #

现有Transaction和Category模型，基于ListView实现仪表盘功能，需要统计当月/上月的收入、支出总和。当前实现中，每个统计项都会发起独立的数据库查询，导致多次数据库往返，希望优化为更高效的查询方式。

模型代码 #

class Transaction(models.Model):
    """ Invoice model.
     Represents a basic income/outcome transaction. """
    
    user = models.ForeignKey(CustomUser, related_name="transactions", on_delete=models.CASCADE)
    title = models.CharField(max_length=32, verbose_name="Title")
    category = models.ForeignKey(Category, related_name="transactions", on_delete=models.CASCADE, null=True, blank=True)
    operation = models.CharField(max_length=8, choices=OPERATION_TYPE, verbose_name="operation")
    value = models.DecimalField(max_digits=14, decimal_places=2, verbose_name="value")
    date_created = models.DateTimeField(auto_now_add=True, blank=True, null=True) 

class Category(MPTTModel):
    """ Category model. 
    Represents a category where money have been spent/earned."""

    name = models.CharField(max_length=54, unique=True)
    parent = TreeForeignKey("self", on_delete=models.CASCADE, null=True, blank=True, related_name='children')

当前视图实现 #

from django.views.generic import ListView
from django.db.models import Sum
from typing import Any, QuerySet

class DashboardView(ListView):
    """ View implements dashboard functionality. """
    model = Transaction
    template_name = "invoices/dashboard.html"
    ordering = "-date_created"


    def get_queryset(self) -> QuerySet[Any]:
        queryset = super().get_queryset()
        queryset = queryset.filter(user=self.request.user).select_related("category__parent")
        return queryset

    def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
        data = super().get_context_data(**kwargs)

        # 获取所有交易
        transactions_all = self.get_queryset()

        # 统计当月收入/支出
        incomes_this_month = transactions_all.filter(transaction_filter.transaction_date_filter(month="current"), operation="incomes")
        incomes_this_month_sum = incomes_this_month.aggregate(Sum("value")).get("value__sum")
        
        expenses_this_month = transactions_all.filter(transaction_filter.transaction_date_filter(month="current"), operation="expenses")                          
        expenses_this_month_sum = expenses_this_month.aggregate(Sum("value")).get("value__sum")
        
        # 统计上月收入/支出
        transactions_prev_month = transactions_all.filter(transaction_filter.transaction_date_filter(month="previous"))
   
        incomes_previous_month = transactions_prev_month.filter(operation="incomes")
        incomes_previous_month_sum = incomes_previous_month.aggregate(Sum("value")).get("value__sum")

        expenses_previous_month = transactions_prev_month.filter(operation="expenses")
        expenses_previous_month_sum = expenses_previous_month.aggregate(Sum("value")).get("value__sum")
        
        # 省略将统计值加入context的代码
        return data

注：transaction_date_filter是用于日期过滤的Q()对象。

当前问题：每次filter()+aggregate()都会发起独立的数据库查询，共4次额外查询，效率低下。

优化方案 #

方案1：使用ORM聚合+Case/When一次性计算所有统计（推荐） #

利用Django ORM的Case/When和Sum聚合，在一次数据库查询中计算所有需要的统计值，避免多次查询。

修改get_context_data中的统计逻辑：

from django.db.models import Sum, Case, When, DecimalField, Q

def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
    data = super().get_context_data(**kwargs)
    transactions_all = self.get_queryset()

    # 一次性聚合所有统计值
    summary = transactions_all.aggregate(
        incomes_this_month_sum=Sum(
            Case(
                When(transaction_filter.transaction_date_filter(month="current") & Q(operation="incomes"), then='value'),
                output_field=DecimalField(),
                default=0
            )
        ),
        expenses_this_month_sum=Sum(
            Case(
                When(transaction_filter.transaction_date_filter(month="current") & Q(operation="expenses"), then='value'),
                output_field=DecimalField(),
                default=0
            )
        ),
        incomes_previous_month_sum=Sum(
            Case(
                When(transaction_filter.transaction_date_filter(month="previous") & Q(operation="incomes"), then='value'),
                output_field=DecimalField(),
                default=0
            )
        ),
        expenses_previous_month_sum=Sum(
            Case(
                When(transaction_filter.transaction_date_filter(month="previous") & Q(operation="expenses"), then='value'),
                output_field=DecimalField(),
                default=0
            )
        )
    )

    # 将统计值加入context，处理无数据时的None情况
    for key, value in summary.items():
        data[key] = value or Decimal('0.00')
    
    return data

优势：

• 仅需1次聚合查询，替代原来的4次查询，大幅减少数据库往返次数
• 所有计算在数据库端完成，效率更高
• 保留了select_related对关联对象的优化

方案2：内存中过滤计算（适合小数据量场景） #

如果用户的交易数据量不大，可以先将所有交易加载到内存中，再通过Python代码过滤计算统计值，仅需1次数据库查询获取所有用户交易。

修改get_context_data：

from datetime import datetime, timedelta
from decimal import Decimal

def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
    data = super().get_context_data(**kwargs)
    # 将queryset转为列表，仅1次数据库查询
    transactions_list = list(self.get_queryset())

    # 辅助函数：判断交易是否属于指定月份
    def is_in_month(transaction, month_type):
        now = datetime.now()
        if month_type == "current":
            return transaction.date_created.year == now.year and transaction.date_created.month == now.month
        elif month_type == "previous":
            # 准确计算上月的年月
            first_day_current = now.replace(day=1)
            last_day_previous = first_day_current - timedelta(days=1)
            return transaction.date_created.year == last_day_previous.year and transaction.date_created.month == last_day_previous.month
        return False

    # 计算各统计值
    data['incomes_this_month_sum'] = sum(
        t.value for t in transactions_list if is_in_month(t, "current") and t.operation == "incomes"
    ) or Decimal('0.00')
    data['expenses_this_month_sum'] = sum(
        t.value for t in transactions_list if is_in_month(t, "current") and t.operation == "expenses"
    ) or Decimal('0.00')
    data['incomes_previous_month_sum'] = sum(
        t.value for t in transactions_list if is_in_month(t, "previous") and t.operation == "incomes"
    ) or Decimal('0.00')
    data['expenses_previous_month_sum'] = sum(
        t.value for t in transactions_list if is_in_month(t, "previous") and t.operation == "expenses"
    ) or Decimal('0.00')

    return data

优势：

• 仅1次数据库查询，避免多次往返
• 逻辑直观，无需复杂的ORM聚合语法

注意：如果用户交易数据量较大（比如上万条），会占用较多内存，此时方案1更合适。

内容的提问来源于stack exchange，提问作者Antony_K