RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move sequential creation behavior into BulkCreateModelMixin to ensure consistent error reports

This commit is contained in:
Jeremy Stretch 2026-08-10 15:55:00 -04:00
parent f68c4f63a7
commit e6dfad94c5
7 changed files with 172 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
docs/integrations/rest-api.md
View File

@ -592,6 +592,9 @@ http://netbox/api/dcim/sites/ \
]
```
!!! note
The bulk creation of objects is an all-or-none operation, meaning that if NetBox fails to successfully create any of the specified objects (e.g. due to a validation error), the entire operation will be aborted and none of the objects will be created.
### Updating an Object
To modify an object which has already been created, make a `PATCH` request to the model's _detail_ endpoint specifying its unique numeric ID. Include any data which you wish to update on the object. As with object creation, the `Authorization` and `Content-Type` headers must also be specified.

3
netbox/dcim/api/views.py
View File

@ -17,7 +17,6 @@ from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
from netbox.api.metadata import ContentTypeMetadata
from netbox.api.pagination import StripCountAnnotationsPaginator
from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
from netbox.api.viewsets.mixins import SequentialBulkCreatesMixin
from utilities.api import get_serializer_for_model
from utilities.query import count_related
from utilities.query_functions import CollateAsChar
@ -428,7 +427,7 @@ class PlatformViewSet(NetBoxModelViewSet):
# Devices/modules
#
class DeviceViewSet(SequentialBulkCreatesMixin, ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
class DeviceViewSet(ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
queryset = Device.objects.prefetch_related(
'device_type__manufacturer', # Referenced by Device.__str__() for unnamed devices
'parent_bay', # Referenced by DeviceSerializer.get_parent_device()

34
netbox/dcim/tests/test_api.py
View File

@ -467,6 +467,33 @@ 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_create_objects_conflicting(self):
"""
POST a set of objects in which two conflict with one another. Objects are created one at a
time, so the second must fail validation against the first rather than passing validation
and then raising an IntegrityError on save.
"""
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()
data = [
{'name': 'Site 10', 'slug': 'site-10'},
{'name': 'Site 11', 'slug': 'site-11'},
{'name': 'Site 10', 'slug': 'site-10'}, # Duplicates the first item
]
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertEqual(self._get_queryset().count(), initial_count)
# Only the third item failed; the first two were provisionally created and rolled back
self.assertEqual(len(response.data['errors']), 1)
self.assertEqual(response.data['errors'][0]['index'], 2)
self.assertIn('slug', response.data['errors'][0]['errors'])
def test_bulk_delete_objects_protected(self):
"""
DELETE a set of objects where one has a protected FK dependency. Verify the structured
@ -2383,10 +2410,9 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
def test_bulk_create_objects_validation_error(self):
"""
POST a set of Device objects where the first passes and the second fails validation.
DeviceViewSet uses SequentialBulkCreatesMixin, so the response should report only the
failed object, and no objects should be created despite the first item passing
(atomic rollback).
POST a set of Device objects where the first passes and the second fails validation. The
response should report only the failed object, and no objects should be created despite
the first item passing (atomic rollback).
"""
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
obj_perm.save()

2
netbox/extras/tests/test_event_rules.py
View File

@ -306,7 +306,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
site = Site.objects.create(name='Site 1', slug='site-1')
# DeviceViewSet uses SequentialBulkCreatesMixin, so each valid object is provisionally
# Bulk creates are performed one object at a time, so each valid object is provisionally
# created (and its event queued) before a later object fails validation.
event_rule = EventRule.objects.get(name='Event Rule 1')
event_rule.object_types.set([ObjectType.objects.get_for_model(Device)])

27
netbox/netbox/api/viewsets/__init__.py
View File

@ -152,6 +152,7 @@ class NetBoxReadOnlyModelViewSet(
class NetBoxModelViewSet(
ETagMixin,
mixins.BackgroundOperationMixin,
mixins.BulkCreateModelMixin,
mixins.BulkUpdateModelMixin,
mixins.BulkDestroyModelMixin,
mixins.ObjectValidationMixin,
@ -250,31 +251,27 @@ class NetBoxModelViewSet(
if (response := self._handle_background_request(request, 'create')) is not None:
return response
# Creating multiple objects, which are validated and saved one at a time in order to
# collect per-object errors (see BulkCreateModelMixin)
if isinstance(request.data, list):
return self.bulk_create(request, *args, **kwargs)
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
bulk_create = getattr(serializer, 'many', False)
self.perform_create(serializer)
# After creating the instance(s), re-initialize the serializer with a queryset
# After creating the instance, re-initialize the serializer with a queryset
# to ensure related objects are prefetched.
if bulk_create:
instance_pks = [obj.pk for obj in serializer.instance]
# Order by PK to ensure that the ordering of objects in the response
# matches the ordering of those in the request.
qs = self.get_queryset().filter(pk__in=instance_pks).order_by('pk')
else:
qs = self.get_queryset().get(pk=serializer.instance.pk)
qs = self.get_queryset().get(pk=serializer.instance.pk)
# Re-serialize the instance(s) with prefetched data
serializer = self.get_serializer(qs, many=bulk_create)
# Re-serialize the instance with prefetched data
serializer = self.get_serializer(qs)
headers = self.get_success_headers(serializer.data)
response = Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
# Add ETag for single-object creation only (bulk returns a list, no single ETag)
if not bulk_create:
if etag := self._get_etag(qs):
response['ETag'] = etag
if etag := self._get_etag(qs):
response['ETag'] = etag
return response

135
netbox/netbox/api/viewsets/mixins.py
View File

@ -1,3 +1,4 @@
import warnings
from contextlib import contextmanager
from django.core.exceptions import ObjectDoesNotExist
@ -9,6 +10,7 @@ from rest_framework import status
from rest_framework.exceptions import ValidationError
from rest_framework.response import Response
from rest_framework.reverse import reverse
from rest_framework.settings import api_settings
from core.models import ObjectType
from core.signals import clear_events
@ -22,6 +24,7 @@ from utilities.rqworker import any_workers_for_queue
__all__ = (
'BackgroundOperationMixin',
'BulkCreateModelMixin',
'BulkDestroyModelMixin',
'BulkUpdateModelMixin',
'CustomFieldsMixin',
@ -232,64 +235,108 @@ class ExportTemplatesMixin:
return super().list(request, *args, **kwargs)
class SequentialBulkCreatesMixin:
class BulkCreateModelMixin:
"""
Perform bulk creation of new objects sequentially, rather than all at once. This ensures that any validation
which depends on the evaluation of existing objects (such as checking for free space within a rack) functions
appropriately.
Support the creation of multiple objects using the list endpoint for a model. Accepts a POST action with a list
of one or more JSON objects, each specifying the attributes of an object to be created. For example:
POST /api/dcim/sites/
[
{"name": "Site 1", "slug": "site-1"},
{"name": "Site 2", "slug": "site-2"}
]
"""
def create(self, request, *args, **kwargs):
# If background processing was requested for a bulk (list) create, enqueue a job and
# return immediately. _handle_background_request() comes from BackgroundOperationMixin;
# fall back to "no background" so this mixin remains usable on its own (e.g. in custom
# viewsets).
handle_background = getattr(self, '_handle_background_request', lambda *a, **kw: None)
if (response := handle_background(request, 'create')) is not None:
return response
# 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.
errors = []
return_data = []
using = router.db_for_write(self.queryset.model)
with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
if not isinstance(request.data, list):
# Creating a single object
return super().create(request, *args, **kwargs)
total = len(request.data)
for i, data in enumerate(request.data):
serializer = self.get_serializer(data=data)
if not serializer.is_valid():
errors.append({'index': i, 'errors': serializer.errors})
continue
try:
# Provisionally create even when a prior item failed, so subsequent
# cross-object validators (e.g. rack space checks) see a realistic state.
# All creates are rolled back together if any item in the batch fails.
self.perform_create(serializer)
except AbortRequest as e:
errors.append({'index': i, 'errors': {'__all__': [str(e.message)]}})
else:
return_data.append(serializer.data)
if errors:
transaction.set_rollback(True)
def bulk_create(self, request, *args, **kwargs):
created_pks, errors = self.perform_bulk_create(request.data)
if errors:
return Response(
{
'detail': _('{failed_count} of {total} objects failed validation.').format(
failed_count=len(errors),
total=total,
total=len(request.data),
),
'errors': errors,
},
status=status.HTTP_400_BAD_REQUEST,
)
headers = self.get_success_headers(return_data[-1]) if return_data else {}
return Response(return_data, status=status.HTTP_201_CREATED, headers=headers)
# Re-fetch the new objects to serialize them with their related objects prefetched. Order by PK
# to ensure that the ordering of objects in the response matches the ordering of those in the
# request (the objects were created in the order given, so PK order is request order).
qs = self.get_queryset().filter(pk__in=created_pks).order_by('pk')
serializer = self.get_serializer(qs, many=True)
return Response(serializer.data, status=status.HTTP_201_CREATED)
def perform_bulk_create(self, data):
created_pks = []
errors = []
using = router.db_for_write(self.queryset.model)
with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
# Validate and save each object in turn, rather than validating the entire batch up front,
# so that validation which depends on the state left by prior saves is evaluated correctly.
# This covers both validation against other existing objects (e.g. checking for free space
# within a rack) and uniqueness: two objects in one batch which conflict with one another
# would otherwise both validate against the pre-batch state and then fail on save, raising
# an unhandled IntegrityError.
for i, item in enumerate(data):
if not isinstance(item, dict):
# Checked explicitly because get_serializer() infers many=True from a list, so a
# nested list would otherwise be validated as a batch of its own
errors.append({
'index': i,
'errors': {
api_settings.NON_FIELD_ERRORS_KEY: [
_('Invalid data. Expected a dictionary, but got {datatype}.').format(
datatype=type(item).__name__
),
],
},
})
continue
serializer = self.get_serializer(data=item)
if not serializer.is_valid():
errors.append({'index': i, 'errors': serializer.errors})
continue
try:
# Provisionally create even when a prior item failed, so subsequent
# cross-object validators see a realistic state. All creates are rolled
# back together if any item in the batch fails.
self.perform_create(serializer)
except AbortRequest as e:
# Raised by a signal receiver rather than by validation (e.g. assigning a tag
# which is restricted to other object types). perform_create() wraps its write
# in its own atomic block, so the connection is rolled back to that savepoint
# and the remaining objects in the batch can still be evaluated. The message is
# coerced to a string because a few receivers pass an exception rather than text.
errors.append({'index': i, 'errors': {'__all__': [str(e.message)]}})
else:
created_pks.append(serializer.instance.pk)
if errors:
transaction.set_rollback(True)
return created_pks, errors
# TODO: Remove this in NetBox v5.0
class SequentialBulkCreatesMixin:
"""
Deprecated no-op mixin retained for backward compatibility.
Historically this was applied to individual ViewSets to make their bulk creates run one object
at a time. All ViewSets derived from NetBoxModelViewSet now do this unconditionally (see
BulkCreateModelMixin), so this mixin is a transparent pass-through and may be removed in a
future release. Plugins should stop inheriting from it.
"""
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
warnings.warn(
"SequentialBulkCreatesMixin is deprecated and no longer does anything; all bulk "
f"creates are now performed sequentially. Remove it from {cls.__name__}.",
DeprecationWarning,
stacklevel=2,
)
class BulkUpdateModelMixin:

34
netbox/utilities/testing/api.py
View File

@ -388,6 +388,40 @@ class APIViewTestCases:
self.assertObjectChange(oc, action=ObjectChangeActionChoices.ACTION_CREATE,
message=changelog_message)
def test_bulk_create_objects_invalid_item(self):
"""
POST a set of objects in which one item is invalid. The failure must be correlated to
that item's position in the request, and the entire batch must be rolled back.
"""
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()
# A non-dictionary is used as the invalid item because it is guaranteed to fail for every
# model, whereas which *fields* are required varies from one model to the next.
response = self.client.post(
self._get_list_url(),
[self.create_data[0], 'this is not an object'],
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 sibling fails validation'
)
self.assertIn('detail', response.data)
self.assertEqual(len(response.data['errors']), 1)
self.assertEqual(response.data['errors'][0]['index'], 1)
self.assertIn('errors', response.data['errors'][0])
class UpdateObjectViewTestCase(APITestCase):
update_data = {}
bulk_update_data = None