Closes #20054: Return per-object error details for failed bulk operations

Bulk update (PATCH), sequential bulk create (POST), and bulk delete (DELETE) on
list endpoints now collect per-object errors instead of aborting on the first
failure. When any objects fail, the entire operation is rolled back atomically
and a 400/409 response is returned with a structured payload:

  {
    "detail": "1 of 3 objects failed validation.",
    "results": [
      {"id": 1, "status": "ok"},
      {"id": 2, "status": "error", "errors": {"name": ["..."]}},
      {"id": 3, "status": "ok"}
    ]
  }

For bulk creates via SequentialBulkCreatesMixin the correlator is "index"
(zero-based position in the request list) since no IDs exist yet. For bulk
delete the status code remains 409 and the correlator is "id".

Successful operations are unchanged (200/201/204).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

netbox/dcim/tests/test_api.py

@@ -151,6 +151,9 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase):
     bulk_update_data = {
         'status': 'planned',
     }
+    bulk_update_invalid_data = {
+        'status': 'not-a-valid-status',
+    }
     graphql_filter_tests = (
         GraphQLFilterTest(
             name='tenant__name__exact',
@@ -467,6 +470,40 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase):
         response = self.client.patch(url, data, format='json', **self.header)
         self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
 
+    def test_bulk_delete_objects_protected(self):
+        """
+        DELETE a set of objects where one has a protected FK dependency. Verify the structured
+        per-object error response and that no objects are deleted (atomic rollback).
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['delete'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        # Site 1 has no dependent Device; Site 2 gets one (Device FK is on_delete=PROTECT)
+        site1 = Site.objects.get(slug='site-1')
+        site2 = Site.objects.get(slug='site-2')
+        create_test_device('Protected Device', site=site2)
+
+        data = [{'id': site1.pk}, {'id': site2.pk}]
+        response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
+        self.assertIn('detail', response.data)
+        self.assertIn('results', response.data)
+        self.assertEqual(len(response.data['results']), 2)
+        # First site (no dependents) would have succeeded
+        self.assertEqual(response.data['results'][0]['id'], site1.pk)
+        self.assertEqual(response.data['results'][0]['status'], 'ok')
+        # Second site (has Device) should have failed
+        self.assertEqual(response.data['results'][1]['id'], site2.pk)
+        self.assertEqual(response.data['results'][1]['status'], 'error')
+        self.assertIn('errors', response.data['results'][1])
+
+        # Verify that no sites were actually deleted (transaction rolled back)
+        self.assertTrue(Site.objects.filter(pk=site1.pk).exists(), 'Site 1 should not have been deleted')
+        self.assertTrue(Site.objects.filter(pk=site2.pk).exists(), 'Site 2 should not have been deleted')
+
 
 class LocationTestCase(APIViewTestCases.APIViewTestCase):
     model = Location
@@ -2162,6 +2199,34 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
         response = self.client.post(url, {'config_template_id': override_template.pk}, format='json', **self.header)
         self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
 
+    def test_bulk_create_objects_validation_error(self):
+        """
+        POST a set of Device objects where all fail validation. DeviceViewSet uses
+        SequentialBulkCreatesMixin, so the response should be the structured per-object error
+        format rather than DRF's default list-of-errors response.
+        """
+        obj_perm = ObjectPermission(name='Test permission', actions=['add'])
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+        initial_count = self._get_queryset().count()
+        # Empty objects fail validation (required fields absent)
+        response = self.client.post(self._get_list_url(), [{}, {}], format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(
+            self._get_queryset().count(), initial_count,
+            'No objects should be created when any fail validation',
+        )
+        self.assertIn('detail', response.data)
+        self.assertIn('results', response.data)
+        self.assertEqual(len(response.data['results']), 2)
+        for i, result in enumerate(response.data['results']):
+            self.assertEqual(result['index'], i)
+            self.assertEqual(result['status'], 'error')
+            self.assertIn('errors', result)
+
 
 class ModuleTestCase(APIViewTestCases.APIViewTestCase):
     model = Module

netbox/netbox/api/viewsets/mixins.py

@@ -1,5 +1,6 @@
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import router, transaction
+from django.db.models import ProtectedError, RestrictedError
 from django.http import Http404
 from django.utils.translation import gettext_lazy as _
 from rest_framework import status
@@ -164,21 +165,39 @@ class SequentialBulkCreatesMixin:
         if (response := handle_background(request, 'create')) is not None:
             return response
 
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
-            if not isinstance(request.data, list):
-                # Creating a single object
-                return super().create(request, *args, **kwargs)
+        if not isinstance(request.data, list):
+            # Creating a single object
+            return super().create(request, *args, **kwargs)
 
-            return_data = []
-            for data in request.data:
+        # Create objects sequentially so each validation sees the state left by prior creates
+        # (e.g. rack space checks). Collect per-object errors instead of failing on the first.
+        results = []
+        return_data = []
+        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+            for i, data in enumerate(request.data):
                 serializer = self.get_serializer(data=data)
-                serializer.is_valid(raise_exception=True)
-                self.perform_create(serializer)
-                return_data.append(serializer.data)
-
-            headers = self.get_success_headers(serializer.data)
+                if serializer.is_valid():
+                    self.perform_create(serializer)
+                    return_data.append(serializer.data)
+                    results.append({'index': i, 'status': 'ok'})
+                else:
+                    results.append({'index': i, 'status': 'error', 'errors': serializer.errors})
+
+            if any(r['status'] == 'error' for r in results):
+                transaction.set_rollback(True)
+
+        if any(r['status'] == 'error' for r in results):
+            failed_count = sum(1 for r in results if r['status'] == 'error')
+            return Response(
+                {
+                    'detail': f'{failed_count} of {len(results)} objects failed validation.',
+                    'results': results,
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
 
-            return Response(return_data, status=status.HTTP_201_CREATED, headers=headers)
+        headers = self.get_success_headers(return_data[-1]) if return_data else {}
+        return Response(return_data, status=status.HTTP_201_CREATED, headers=headers)
 
 
 class BulkUpdateModelMixin:
@@ -226,7 +245,17 @@ class BulkUpdateModelMixin:
             obj.pop('id'): obj for obj in request.data
         }
 
-        object_pks = self.perform_bulk_update(qs, update_data, partial=partial)
+        object_pks, results = self.perform_bulk_update(qs, update_data, partial=partial)
+
+        if results:
+            failed_count = sum(1 for r in results if r['status'] == 'error')
+            return Response(
+                {
+                    'detail': f'{failed_count} of {len(results)} objects failed validation.',
+                    'results': results,
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
 
         # Prefetch related objects for all updated instances
         qs = self.get_queryset().filter(pk__in=object_pks)
@@ -236,17 +265,31 @@ class BulkUpdateModelMixin:
 
     def perform_bulk_update(self, objects, update_data, partial):
         updated_pks = []
+        results = []
         with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+            # Pass 1: validate all objects without writing to the database
+            prepared = []
             for obj in objects:
                 data = update_data.get(obj.id)
                 if hasattr(obj, 'snapshot'):
                     obj.snapshot()
                 serializer = self.get_serializer(obj, data=data, partial=partial)
-                serializer.is_valid(raise_exception=True)
-                self.perform_update(serializer)
-                updated_pks.append(obj.pk)
-
-        return updated_pks
+                prepared.append((obj, serializer, serializer.is_valid()))
+
+            if any(not valid for _, _, valid in prepared):
+                results = [
+                    {'id': obj.pk, 'status': 'error', 'errors': ser.errors} if not valid
+                    else {'id': obj.pk, 'status': 'ok'}
+                    for obj, ser, valid in prepared
+                ]
+                transaction.set_rollback(True)
+            else:
+                # Pass 2: all objects are valid — perform updates
+                for obj, serializer, _ in prepared:
+                    self.perform_update(serializer)
+                    updated_pks.append(obj.pk)
+
+        return updated_pks, results
 
     def get_bulk_update_serializer_class(self, *, partial=False):
         return get_bulk_update_serializer_class(
@@ -302,18 +345,48 @@ class BulkDestroyModelMixin:
             o['id']: o.get('changelog_message') for o in serializer.validated_data
         }
 
-        self.perform_bulk_destroy(qs, changelog_messages)
+        results = self.perform_bulk_destroy(qs, changelog_messages)
+
+        if results and any(r['status'] == 'error' for r in results):
+            failed_count = sum(1 for r in results if r['status'] == 'error')
+            return Response(
+                {
+                    'detail': f'{failed_count} of {len(results)} objects could not be deleted.',
+                    'results': results,
+                },
+                status=status.HTTP_409_CONFLICT,
+            )
 
         return Response(status=status.HTTP_204_NO_CONTENT)
 
     def perform_bulk_destroy(self, objects, changelog_messages=None):
         changelog_messages = changelog_messages or {}
+        results = []
         with transaction.atomic(using=router.db_for_write(self.queryset.model)):
             for obj in objects:
                 if hasattr(obj, 'snapshot'):
                     obj.snapshot()
                 obj._changelog_message = changelog_messages.get(obj.pk)
-                self.perform_destroy(obj)
+                pk = obj.pk  # Django sets obj.pk = None after deletion; capture it first
+                try:
+                    self.perform_destroy(obj)
+                    results.append({'id': pk, 'status': 'ok'})
+                except (ProtectedError, RestrictedError) as e:
+                    protected = list(
+                        e.protected_objects if isinstance(e, ProtectedError) else e.restricted_objects
+                    )
+                    n = len(protected)
+                    objects_str = ', '.join(f'{o} ({o.pk})' for o in protected[:10])
+                    if n > 10:
+                        objects_str += f', and {n - 10} more'
+                    results.append({
+                        'id': obj.pk,
+                        'status': 'error',
+                        'errors': {'detail': f'Unable to delete. {n} dependent object(s): {objects_str}'},
+                    })
+            if any(r['status'] == 'error' for r in results):
+                transaction.set_rollback(True)
+        return results
 
 
 class ObjectValidationMixin:

netbox/utilities/testing/api.py

@@ -395,6 +395,7 @@ class APIViewTestCases:
     class UpdateObjectViewTestCase(APITestCase):
         update_data = {}
         bulk_update_data = None
+        bulk_update_invalid_data = None
         validation_excluded_fields = []
 
         def test_update_object_without_permission(self):
@@ -546,6 +547,39 @@ class APIViewTestCases:
                     self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_UPDATE,
                         message=changelog_message)
 
+        def test_bulk_update_objects_validation_error(self):
+            """
+            PATCH a set of objects where one fails validation. Verify the structured per-object error
+            response and that no objects are modified (atomic rollback).
+            """
+            if self.bulk_update_data is None or self.bulk_update_invalid_data is None:
+                self.skipTest('Bulk update data not set')
+
+            obj_perm = ObjectPermission(name='Test permission', actions=['change'])
+            obj_perm.save()
+            obj_perm.users.add(self.user)
+            obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
+
+            id_list = list(self._get_queryset().values_list('id', flat=True)[:2])
+            self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk update validation error')
+
+            # First object: valid data; second: invalid data that must fail validation
+            data = [
+                {'id': id_list[0], **self.bulk_update_data},
+                {'id': id_list[1], **self.bulk_update_invalid_data},
+            ]
+            response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
+
+            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+            self.assertIn('detail', response.data)
+            self.assertIn('results', response.data)
+            self.assertEqual(len(response.data['results']), 2)
+            self.assertEqual(response.data['results'][0]['id'], id_list[0])
+            self.assertEqual(response.data['results'][0]['status'], 'ok')
+            self.assertEqual(response.data['results'][1]['id'], id_list[1])
+            self.assertEqual(response.data['results'][1]['status'], 'error')
+            self.assertIn('errors', response.data['results'][1])
+
     class DeleteObjectViewTestCase(APITestCase):
 
         def test_delete_object_without_permission(self):