Django嵌套聚合问题：如何在TransactionGroup层级实现金额转换汇总

我正在使用DRF开发，在视图类中定义queryset时遇到问题。现有三个模型：

class ExchangeRate(...):
    date = models.DateField(...)
    rate = models.DecimalField(...)
    from_currency = models.CharField(...)
    to_currency = models.CharField(...)

class Transaction(...):
    amount = models.DecimalField(...)
    currency = models.CharField(...)
    group = models.ForeignKey("TransactionGroup", ...)

class TransactionGroup(...):
    ...

需要在TransactionGroup层级构建queryset，实现两个需求：

• 为交易组内的每个Transaction添加注解字段converted_amount，值为amount乘以对应ExchangeRate的rate（currency匹配to_currency）；
• 将每个交易组内所有Transaction的converted_amount求和，作为TransactionGroup的注解字段converted_amount_sum。

期望的TransactionGroup JSON响应示例如下：

[
  {
    "id": 1,
    "converted_amount_sum": 5000,
    "transactions": [
      {
        "id": 1,
        "amount": 1000,
        "converted_amount": 500,
        "currency": "USD"
      },
      {
        "id": 2,
        "amount": 5000,
        "converted_amount": 4500,
        "currency": "EUR"
      }
    ]
  },
  ...
]

我尝试的代码如下：

from django.db.models import F

annotated_transactions = Transaction.objects.annotate(
    converted_amount = F("amount") * exchange_rate.rate     # <-- 这里做了简化
).values(
    "transaction_group"
).annotate(
    amount=Sum("converted_amount"),
)

这段代码在Transaction模型上的注解可以正常工作，但在TransactionGroup层级求和时抛出错误：

FieldError: Cannot compute Sum('converted_amount'), `converted_amount` is an aggregate

额外需求：希望能通过converted_amount_sum对TransactionGroups进行排序和过滤，且无需额外数据库查询/操作。

Django ORM不支持对annotate生成的字段直接做嵌套聚合，这就是报错的原因。下面是符合需求的实现方案：

1. 实现TransactionGroup的converted_amount_sum注解 #

直接在TransactionGroup的查询中关联Transaction和ExchangeRate，在数据库层面完成amount * rate的求和计算。这里假设你需要匹配Transaction.currency = ExchangeRate.to_currency，并取对应货币的最新汇率（可根据实际业务调整汇率匹配规则）：

from django.db.models import Sum, F, Subquery, OuterRef
from django.db.models.functions import Coalesce

# 定义子查询：获取每个货币对应的最新汇率
latest_exchange_rates = ExchangeRate.objects.filter(
    to_currency=OuterRef('currency')
).order_by('-date')

# 构建TransactionGroup的queryset，注解求和字段
transaction_groups = TransactionGroup.objects.annotate(
    converted_amount_sum=Coalesce(
        Sum(
            F('transaction__amount') * Subquery(
                latest_exchange_rates.values('rate')[:1]
            ),
            output_field=models.DecimalField()
        ),
        0  # 无交易时求和默认0
    )
)

这样生成的converted_amount_sum是数据库原生计算的，可以直接用于排序和过滤：

# 按转换金额总和降序排序
transaction_groups = transaction_groups.order_by('-converted_amount_sum')

# 过滤转换金额总和大于1000的组
transaction_groups = transaction_groups.filter(converted_amount_sum__gt=1000)

2. 为每个Transaction添加converted_amount字段 #

推荐使用第二种方式（数据库层面计算，无额外查询）：

方式一：Serializer中计算 #

在TransactionSerializer中通过SerializerMethodField获取汇率并计算转换金额，需注意优化避免N+1查询：

from rest_framework import serializers

class TransactionSerializer(serializers.ModelSerializer):
    converted_amount = serializers.SerializerMethodField()

    class Meta:
        model = Transaction
        fields = ['id', 'amount', 'currency', 'converted_amount']

    def get_converted_amount(self, obj):
        # 从上下文批量获取的汇率字典中取值，避免重复查询
        exchange_rates = self.context.get('exchange_rates', {})
        rate = exchange_rates.get(obj.currency)
        return obj.amount * rate if rate else obj.amount

class TransactionGroupSerializer(serializers.ModelSerializer):
    converted_amount_sum = serializers.DecimalField(max_digits=10, decimal_places=2)
    transactions = TransactionSerializer(many=True)

    class Meta:
        model = TransactionGroup
        fields = ['id', 'converted_amount_sum', 'transactions']

在视图中批量获取所有需要的汇率，传入Serializer上下文：

def get_queryset(self):
    transaction_groups = TransactionGroup.objects.annotate(
        converted_amount_sum=Coalesce(
            Sum(
                F('transaction__amount') * Subquery(
                    latest_exchange_rates.values('rate')[:1]
                ),
                output_field=models.DecimalField()
            ),
            0
        )
    ).prefetch_related('transaction_set')

    # 批量获取所有交易涉及的货币汇率
    currencies = Transaction.objects.filter(group__in=transaction_groups).values_list('currency', flat=True).distinct()
    exchange_rates = {er.to_currency: er.rate for er in ExchangeRate.objects.filter(to_currency__in=currencies).order_by('-date')}
    
    # 将汇率字典传入Serializer上下文
    self.serializer_context['exchange_rates'] = exchange_rates
    return transaction_groups

方式二：Prefetch注解Transaction（推荐） #

通过Prefetch对象给关联的Transaction批量添加converted_amount注解，所有计算都在数据库层面完成，无额外查询：

from django.db.models import Prefetch

# 定义带注解的Transaction queryset
annotated_transactions = Transaction.objects.annotate(
    converted_amount=F('amount') * Subquery(
        latest_exchange_rates.values('rate')[:1]
    )
)

# 在TransactionGroup查询中使用Prefetch加载带注解的交易
transaction_groups = TransactionGroup.objects.annotate(
    converted_amount_sum=Coalesce(
        Sum(
            F('transaction__amount') * Subquery(
                latest_exchange_rates.values('rate')[:1]
            ),
            output_field=models.DecimalField()
        ),
        0
    )
).prefetch_related(
    Prefetch('transaction_set', queryset=annotated_transactions)
)

对应的TransactionSerializer可以直接序列化converted_amount字段：

class TransactionSerializer(serializers.ModelSerializer):
    class Meta:
        model = Transaction
        fields = ['id', 'amount', 'currency', 'converted_amount']

这种方式完美满足需求：converted_amount_sum支持排序过滤，所有计算都在数据库完成，无额外查询。

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