django-rest-framework：在 ModelViewSet 中添加批量操作

Question

我有许多端点使用 ModelViewSet 来管理我的模型的 CRUD 操作。

我想做的是在这些相同的端点添加批量创建、更新和删除。换句话说，我想将 POST、PUT、PATCH 和 DELETE 添加到收集端点（例如：/api/v1/my-model）。有一个可用的 django-rest-framework-bulk 包，但它似乎已被废弃（4 年未更新），我不习惯在生产中使用不再活跃的包。

此外，这里有几个类似的问题有解决方案，以及我找到的博客文章。但是，它们似乎都使用基础 ViewSet 或 APIView，这将需要重写我现有的所有 ModelViewSet 代码。

最后，可以选择使用 @action 装饰器，但这需要我有一个单独的列表端点（例如- /api/v1/my-model/bulk），我想避免这一点。

有没有其他方法可以在保持现有 ModelViewSet 视图的同时完成此操作？我一直在研究 GenericViewSet 和 mixins，想知道创建我自己的 mixin 是否可行。但是，查看 mixin 代码，您似乎无法指定要附加到给定 mixin 的 HTTP 请求方法。

最后，我尝试创建一个单独的 ViewSet 接受 PUT 并将其添加到我的 URL，但这不起作用（当我尝试 PUT 到 /api/v1/my-model).我试过的代码如下所示：

# views.py
class MyModelViewSet(viewsets.ModelViewSet):
    serializer_class = MyModelSerializer
    permission_classes = (IsAuthenticated,)
    queryset = MyModel.objects.all()
    paginator = None

class ListMyModelView(viewsets.ViewSet):
    permission_classes = (IsAuthenticated,)

    def put(self, request):
        # Code for updating list of models will go here.

        return Response({'test': 'list put!'})


# urls.py
router = DefaultRouter(trailing_slash=False)
router.register(r'my-model', MyModelViewSet)
router.register(r'my-model', ListMyModelView, base_name='list-my-model')

urlpatterns = [
    path('api/v1/', include(router.urls)),
    # more paths for auth, admin, etc..
]

想法？

Answer 1

我知道你说过你想避免添加额外的操作，但我认为这是批量更新现有视图的最简单方法 create/update/delete。

您可以创建一个 mixin，添加到您的视图中来处理所有事情，您只需更改现有视图和序列化程序中的一行即可。

假设您的 ListSerializer 看起来与 DRF documentation 相似，mixin 将如下所示。

core/serializers.py

class BulkUpdateSerializerMixin:
    """
    Mixin to be used with BulkUpdateListSerializer & BulkUpdateRouteMixin
    that adds the ID back to the internal value from the raw input data so
    that it's included in the validated data.
    """
    def passes_test(self):
        # Must be an update method for the ID to be added to validated data
        test = self.context['request'].method in ('PUT', 'PATCH')
        test &= self.context.get('bulk_update', False)
        return test

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

        if self.passes_test():
            ret['id'] = self.fields['id'].get_value(data)

        return ret

core/views.py

class BulkUpdateRouteMixin:
    """
    Mixin that adds a `bulk_update` API route to a view set. To be used
    with BulkUpdateSerializerMixin & BulkUpdateListSerializer.
    """
    def get_object(self):
        # Override to return None if the lookup_url_kwargs is not present.
        lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field
        if lookup_url_kwarg in self.kwargs:
            return super().get_object()
        return

    def get_serializer(self, *args, **kwargs):
        # Initialize serializer with `many=True` if the data passed
        # to the serializer is a list.
        if self.request.method in ('PUT', 'PATCH'):
            data = kwargs.get('data', None)
            kwargs['many'] = isinstance(data, list)
        return super().get_serializer(*args, **kwargs)

    def get_serializer_context(self):
        # Add `bulk_update` flag to the serializer context so that
        # the id field can be added back to the validated data through
        # `to_internal_value()`
        context = super().get_serializer_context()
        if self.action == 'bulk_update':
            context['bulk_update'] = True
        return context

    @action(detail=False, methods=['put'], url_name='bulk_update')
    def bulk_update(self, request, **kwargs):
        queryset = self.filter_queryset(self.get_queryset())
        serializer = self.get_serializer(
            queryset,
            data=request.data,
            many=True,
        )
        serializer.is_valid(raise_exception=True)
        self.perform_update(serializer)
        return Response(serializer.data, status=status.HTTP_200_OK)

然后你就可以继承mixins

class MyModelSerializer(BulkUpdateSerializerMixin
                        serializers.ModelSerializer):
    class Meta:
        model = MyModel
        list_serializer_class = BulkUpdateListSerializer

class MyModelViewSet(BulkUpdateRouteMixin,
                     viewsets.ModelViewSet):
    ...

你的 PUT 请求只需要指向 '/api/v1/my-model/bulk_update'

更新的 mixin 不需要额外的视图集操作：

对于批量操作，向列表视图提交一个 POST 请求，并将数据作为列表。

class BulkUpdateSerializerMixin:
    def passes_test(self):
        test = self.context['request'].method in ('POST',)
        test &= self.context.get('bulk', False)
        return test

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

        if self.passes_test():
            ret['id'] = self.fields['id'].get_value(data)

        return ret

在 get_serializer() 中有一个检查以确保只有 POST 请求可以被接受以进行批量操作。如果它是 POST 并且请求数据是一个列表，则添加一个标志，以便可以将 ID 字段添加回经过验证的数据，并且您的 ListSerializer 可以处理批量操作。

class BulkUpdateViewSetMixin:
    def get_serializer(self, *args, **kwargs):
        serializer_class = self.get_serializer_class()
        kwargs['context'] = self.get_serializer_context()
        if self.request.method in ('POST',):
            data = kwargs.get('data', None)
            is_bulk = isinstance(data, list)
            kwargs['many'] = is_bulk
            kwargs['context']['bulk'] = is_bulk
        return serializer_class(*args, **kwargs)

    def create(self, request, *args, **kwargs):
        if isinstance(request.data, list):
            return self.bulk_update(request)
        return super().create(request, *args, **kwargs)

    def bulk_update(self, request):
        queryset = self.filter_queryset(self.get_queryset())
        serializer = self.get_serializer(
            queryset,
            data=request.data,
        )
        serializer.is_valid(raise_exception=True)
        self.perform_update(serializer)
        return Response(serializer.data, status=status.HTTP_200_OK)

我已经测试过它是否有效，但我不知道它将如何影响 API 架构文档。

django-rest-framework：在 ModelViewSet 中添加批量操作

django-rest-framework: Adding bulk operation in a ModelViewSet

python

django

django-rest-framework

django-rest-viewsets