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

Question

I have many endpoints which use ModelViewSet to manage CRUD operations for my models.

What I am trying to do, is to add bulk create, update, and delete at these same endpoints. In other words, I want to add POST, PUT, PATCH and DELETE to the collection endpoint (e.g.: /api/v1/my-model). There is a django-rest-framework-bulk package available, but it seems to be abandoned (hasn't been updated in 4 years) and I'm not comfortable using a package in production that is no longer active.

Additionally, There are several similar questions here that have solutions, as well as blog posts I've found. However, they all seem to use the base ViewSet, or APIView, which would require re-writing all of my existing ModelViewSet code.

Finally, there is the option of using the @action decorator, however this would require me to have a separate list endpoint (e.g.- /api/v1/my-model/bulk) which I'd like to avoid.

Are there any other ways to accomplish this while keeping my existing ModelViewSet views? I've been looking at GenericViewSet and mixins, and am wondering if creating my own mixin might be the way to go. However, looking at the mixin code, it doesn't appear that you can specify an HTTP Request method to be attached to a given mixin.

Finally, I have tried creating a separate ViewSet that accepts PUT and adding it to my URLs, but this doesn't work (I get a 405 Method not allowed when I try to PUT to /api/v1/my-model). The code I tried looks like this:

# 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..
]

Thoughts?

Answer 1

I know you said you wanted to avoid adding an extra action but in my opinion it's the simplest way to update your existing views for bulk create/update/delete.

You can create a mixin that you add to your views that will handle everything, you'd just be changing one line in your existing views and serializers.

Assuming your ListSerializer look similar to the DRF documentation the mixins would be as follows.

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)

Then you would just inherit from the mixins

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

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

And your PUT request would just have to point to '/api/v1/my-model/bulk_update'

Updated mixins that don't require extra viewset action:

For bulk operations submit a POST request to the list view with data as a list.

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

In get_serializer() there's a check to ensure that only POST requests can be accepted for bulk operations. If it's a POST and the request data is a list then add a flag so the ID field can be added back to the validated data and your ListSerializer can handle the bulk operations.

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)

I've tested that this works but I have no idea how it will affect API schema documentation.