• 创建 bulk_serializer.py 文件

"""批量操作"""
from rest_framework.serializers import ListSerializer


__all__ = [
    'BulkListSerializer',
    'BulkSerializerMixin',
]


class BulkSerializerMixin(object):
    def to_internal_value(self, data):
        ret = super(BulkSerializerMixin, self).to_internal_value(data)

        id_attr = getattr(self.Meta, 'update_lookup_field', 'id')
        request_method = getattr(getattr(self.context.get('view'), 'request'), 'method', '')

        # add update_lookup_field field back to validated data
        # since super by default strips out read-only fields
        # hence id will no longer be present in validated_data
        if all((isinstance(self.root, BulkListSerializer),
                id_attr,
                request_method in ('PUT', 'PATCH'))):
            id_field = self.fields[id_attr]
            id_value = id_field.get_value(data)

            ret[id_attr] = id_value

        return ret


class BulkListSerializer(ListSerializer):
    # 批量更新
    def update(self, queryset, all_validated_data):
        updated_objects = []
        all_validated_data_by_id = {
            str(i.pop('id')): i
            for i in all_validated_data
        }

        for obj in queryset:
            updated_objects.append(self.child.update(obj, all_validated_data_by_id.get(str(obj.id))))

        return updated_objects

• 序列器中使用

from rest_framework import serializers

from utils.bulk_serializer import BulkSerializerMixin, BulkListSerializer
from .models import MyModel


class MySerializer(BulkSerializerMixin, serializers.ModelSerializer):
    class Meta:
        model = MyModel
        fields = "__all__"
        list_serializer_class = BulkListSerializer

• 视图使用

class MyView(GenericViewSet):
    queryset = MyModel.objects.all()
    serializer_class = MySerializer

    @transaction.atomic
    @action(detail=False, methods=['patch'])
    def update_status(self, request):
        """批量编辑"""
        bulk = isinstance(request.data, list)
        if bulk:
            update_data = request.data
        else:
            # 单个也按批量处理
            update_data = [request.data]

        ids = []
        for item in update_data:
            pk = item.get('id')
            ids.append(pk)

        instances = self.get_queryset().filter(id__in=ids)

        serializer = self.get_serializer(
            instances,
            data=update_data,
            many=True,
            partial=True,
        )
        serializer.is_valid(raise_exception=True)
        serializer.save()

        return Response(serializer.data)