Django DRF：ListCreateAPIView实现过滤聚合+带验证批量创建交易

我有一个通过外键关联User模型的Transaction模型，需求为：过滤每个用户当日的所有交易，聚合求和后计算10%作为利息并创建新的交易记录。

此前在serializers.py中实现该逻辑时，使用Transaction.objects.create()会绕过所有验证——例如Transaction模型中trans_type、note等非空字段未赋值却未触发报错。现尝试在ListCreateAPIView的perform_create方法中实现，但未成功。相关代码及模型字段如下：

序列化器代码 #

class UserTransactionFilterSerializer(serializers.Serializer):
    
username = serializers.CharField(read_only = True)
id = serializers.IntegerField(read_only = True)
agg_amt_btc = serializers.SerializerMethodField(read_only=True)
trans = serializers.SerializerMethodField(read_only = True)

# Gets transaction data using UserTransactionSerializer
def get_trans(self, obj):
    user = obj
    print(user)
    today = timezone.now().date()
    my_trans_qs = user.transaction_set.filter(trans_date__date=today)
    return TransactionUserInlineSerializer(my_trans_qs, many=True, context = self.context).data

# Aggregates the trans amount (sum and avg)
def get_agg_amt_btc(self, obj):
    user = obj
    print(user)
    today = timezone.now().date()
    my_trans_qs = user.transaction_set.filter(trans_date__date=today)
    return my_trans_qs.aggregate(sum_btc = Sum('trans_amt_btc'), avg_btc = Avg('trans_amt_btc'))


def create(self,obj):
    today = timezone.now().date()
    for user in User.objects.filter(transaction__trans_date__date=today).distinct():

        total_transactions = user.transaction_set.aggregate(sum_btc = Sum('trans_amt_btc'))['sum_btc']

        print(user)
        transaction_amount = total_transactions * Decimal('0.1')
        print(transaction_amount)
        
        with transaction.atomic():
            Transaction.objects.create(owner=user, trans_amt_btc=transaction_amount)

    return user

Transaction模型字段 #

owner = models.ForeignKey(User, default = 1, null = True, on_delete = models.SET_NULL) #related_name defaults to transaction_set

trans_date = models.DateTimeField(auto_now_add=True, blank=False)
trans_amt_btc = models.DecimalField(decimal_places=12, max_digits=24,blank=False, null=False)

trans_type = models.CharField(choices= TRANSACTION_TYPE_CHOICES, max_length=24, blank=False, null=False)
trans_reason = models.CharField(choices= TRANSACTION_REASON_CHOICES, max_length=24, blank=False, null=False)
trans_mode = models.CharField(choices= TRANSACTION_MODE_CHOICES, max_length=24, blank=False, null=False)
trans_status = models.CharField(choices= TRANSACTION_STATUS_CHOICES, max_length=24, blank=False, default="APPROVED")

note = models.CharField(max_length=120, blank=False, null=False)

视图代码 #

class UserTransactionCommissionView(generics.ListCreateAPIView):
queryset = User.objects.all()
serializer_class = UserTransactionFilterSerializer

def perform_create(self,serializer):
    today = timezone.now().date()
    for user in list(User.objects.filter(transaction__trans_date__date=today).distinct()):

        total_transactions = user.transaction_set.aggregate(sum_btc = Sum('trans_amt_btc'))['sum_btc']

        print(total_transactions)
        
        total_transactions = serializer.validated_data.get('trans_amt_btc')

        
        transaction_amount = total_transactions * Decimal('0.1')
        print(transaction_amount)

        serializer.save(owner=user)

        
        # with transaction.atomic():
        #     Transaction.objects.create(owner=user, trans_amt_btc=transaction_amount)

    return user

核心问题分析 #

1. Transaction.objects.create()直接操作数据库，不会触发序列化器的验证逻辑，导致模型中必填字段缺失也能创建记录；
2. 当前视图的perform_create逻辑混乱：手动计算的交易总和被serializer.validated_data.get('trans_amt_btc')错误覆盖，且原序列化器是针对User模型的，不适合处理Transaction创建；
3. 原序列化器的create方法职责越界，既负责User数据序列化，又处理Transaction创建，违反单一职责原则。

修改步骤 #

1. 编写Transaction专用创建序列化器 #

创建独立的序列化器处理利息交易的创建，确保所有必填字段的验证生效：

from rest_framework import serializers
from .models import Transaction

class TransactionInterestCreateSerializer(serializers.ModelSerializer):
    class Meta:
        model = Transaction
        fields = ['owner', 'trans_amt_btc', 'trans_type', 'trans_reason', 'trans_mode', 'note']
        # trans_status有默认值、trans_date为auto_now_add，无需纳入字段

    def validate(self, data):
        # 自定义验证：利息金额必须大于0
        if data['trans_amt_btc'] <= 0:
            raise serializers.ValidationError("利息金额不能为负")
        return data

2. 重构视图的perform_create逻辑 #

改用Transaction序列化器创建记录，同时保证事务原子性：

from rest_framework import generics
from django.utils import timezone
from django.db.models import Sum
from django.db import transaction
from decimal import Decimal
from .models import User, Transaction
from .serializers import UserTransactionFilterSerializer, TransactionInterestCreateSerializer

class UserTransactionCommissionView(generics.ListCreateAPIView):
    queryset = User.objects.all()
    serializer_class = UserTransactionFilterSerializer

    def perform_create(self, serializer):
        today = timezone.now().date()
        # 筛选当日有交易的用户
        target_users = User.objects.filter(transaction__trans_date__date=today).distinct()

        with transaction.atomic():
            for user in target_users:
                # 计算当日交易总和
                total_sum = user.transaction_set.filter(trans_date__date=today).aggregate(sum_btc=Sum('trans_amt_btc'))['sum_btc']
                if not total_sum or total_sum <= 0:
                    continue  # 无有效交易则跳过

                interest = total_sum * Decimal('0.1')
                # 构造利息交易数据（替换为你实际的选项值）
                trans_data = {
                    'owner': user.id,
                    'trans_amt_btc': interest,
                    'trans_type': 'INTEREST',
                    'trans_reason': 'DAILY_INTEREST_REWARD',
                    'trans_mode': 'SYSTEM_AUTO',
                    'note': f"当日交易10%利息，交易总额：{total_sum}"
                }
                # 初始化序列化器并触发验证
                trans_serializer = TransactionInterestCreateSerializer(data=trans_data)
                trans_serializer.is_valid(raise_exception=True)
                # 保存记录
                trans_serializer.save()

        # 执行原User序列化器的保存逻辑
        return super().perform_create(serializer)

3. 清理原序列化器的冗余代码 #

删除原UserTransactionFilterSerializer中的create方法，回归其仅负责User及交易数据序列化的职责：

class UserTransactionFilterSerializer(serializers.Serializer):
    username = serializers.CharField(read_only=True)
    id = serializers.IntegerField(read_only=True)
    agg_amt_btc = serializers.SerializerMethodField(read_only=True)
    trans = serializers.SerializerMethodField(read_only=True)

    def get_trans(self, obj):
        user = obj
        today = timezone.now().date()
        my_trans_qs = user.transaction_set.filter(trans_date__date=today)
        return TransactionUserInlineSerializer(my_trans_qs, many=True, context=self.context).data

    def get_agg_amt_btc(self, obj):
        user = obj
        today = timezone.now().date()
        my_trans_qs = user.transaction_set.filter(trans_date__date=today)
        return my_trans_qs.aggregate(sum_btc=Sum('trans_amt_btc'), avg_btc=Avg('trans_amt_btc'))

关键说明 #

• 通过专用序列化器创建交易，会自动验证所有必填字段，缺失时直接抛出验证错误，避免非法数据入库；
• transaction.atomic()确保所有利息交易要么全部创建成功，要么全部回滚，防止数据不一致；
• 跳过无有效交易的用户，避免生成无效的利息记录；
• 序列化器职责分离，UserTransactionFilterSerializer仅负责数据展示，TransactionInterestCreateSerializer负责交易创建，代码更易维护。

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