在Django REST Framework中更新多个嵌套对象

我有三个存在多对多关联的模型，希望通过三个独立的序列化器实现所有实例的一次性更新操作，目前已经完成了创建逻辑，但不清楚如何实现嵌套对象的批量更新，现有代码如下：

class PlannedChargePriceSerializer(serializers.ModelSerializer):
    class Meta:
        model = ChargePrice
        fields = ['id', 'start', 'end', 'price']

class PlannedChargeSerializer(serializers.ModelSerializer):
    prices = PlannedChargePriceSerializer(many=True)

    class Meta:
        model = Charge
        fields = ['id', 'type', 'qtd', 'status', 'prices']

class PlannedSerializer(serializers.ModelSerializer):
    charges = PlannedChargeSerializer(many=True)

    class Meta:
        model = Planned
        fields = '__all__'
        read_only_fields = ['id', 'date', 'charges']

    def create(self, validated_data):
        charges_data = validated_data.pop('charges')
        planned = Planned.objects.create(**validated_data)

        for charge_data in charges_data:
            charge_prices_data = charge_data.pop('prices')
            charge = Charge.objects.create(planned=planned, **charge_data)

            for charge_price_data in charge_prices_data:
                ChargePrice.objects.create(charge=charge, **charge_price_data)

        return planned

    def update(self, instance, validated_data):
        # No idea how update multiple nested objects
        return instance

嵌套对象批量更新的实现思路 #

要实现多层嵌套对象的更新，需要按照从外到内的层级，分别处理更新现有实例、创建新实例、删除多余实例三个操作，核心是通过id字段匹配新旧数据。

完整的update方法实现 #

替换PlannedSerializer中的update方法为以下代码：

def update(self, instance, validated_data):
    # 1. 更新Planned自身字段（排除charges）
    planned_data = validated_data.copy()
    charges_data = planned_data.pop('charges', [])
    for attr, value in planned_data.items():
        setattr(instance, attr, value)
    instance.save()

    # 2. 处理Charge层级的更新、创建、删除
    existing_charges = {charge.id: charge for charge in instance.charges.all()}
    incoming_charge_ids = []

    for charge_data in charges_data:
        charge_id = charge_data.get('id')
        incoming_charge_ids.append(charge_id)
        prices_data = charge_data.pop('prices', [])

        if charge_id and charge_id in existing_charges:
            # 更新现有Charge实例
            charge = existing_charges[charge_id]
            for attr, value in charge_data.items():
                setattr(charge, attr, value)
            charge.save()

            # 3. 处理ChargePrice层级的更新、创建、删除
            existing_prices = {price.id: price for price in charge.prices.all()}
            incoming_price_ids = []

            for price_data in prices_data:
                price_id = price_data.get('id')
                incoming_price_ids.append(price_id)

                if price_id and price_id in existing_prices:
                    # 更新现有ChargePrice实例
                    price = existing_prices[price_id]
                    for attr, value in price_data.items():
                        setattr(price, attr, value)
                    price.save()
                elif price_data:
                    # 创建新的ChargePrice实例
                    ChargePrice.objects.create(charge=charge, **price_data)

            # 删除不在传入数据中的ChargePrice实例
            for price_id in existing_prices.keys():
                if price_id not in incoming_price_ids:
                    existing_prices[price_id].delete()

        elif charge_data:
            # 创建新的Charge实例及其关联的ChargePrice
            charge = Charge.objects.create(planned=instance, **charge_data)
            for price_data in prices_data:
                ChargePrice.objects.create(charge=charge, **price_data)

    # 删除不在传入数据中的Charge实例
    for charge_id in existing_charges.keys():
        if charge_id not in incoming_charge_ids:
            existing_charges[charge_id].delete()

    return instance

关键说明 #

• 层级处理：从最外层Planned到中间层Charge，再到最内层ChargePrice，逐层处理更新逻辑
• 实例匹配：通过id字段将传入数据与数据库已有实例关联，确保更新目标准确
• 全量同步：传入数据中不存在的实例会被删除，实现数据全量同步（不需要删除逻辑可移除对应代码块）
• 字段更新：通过setattr批量更新模型字段，避免逐个字段赋值的冗余代码

内容的提问来源于stack exchange，提问作者Tiago Lourenço