mirror of
https://github.com/netbox-community/netbox.git
synced 2025-12-29 16:15:55 +00:00
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
from django.core.exceptions import ObjectDoesNotExist
|
|
from django.db import transaction
|
|
from rest_framework import status
|
|
from rest_framework.response import Response
|
|
|
|
from netbox.api.serializers import BulkOperationSerializer
|
|
|
|
__all__ = (
|
|
'BulkUpdateModelMixin',
|
|
'BulkDestroyModelMixin',
|
|
'ObjectValidationMixin',
|
|
)
|
|
|
|
|
|
class BulkUpdateModelMixin:
|
|
"""
|
|
Support bulk modification of objects using the list endpoint for a model. Accepts a PATCH action with a list of one
|
|
or more JSON objects, each specifying the numeric ID of an object to be updated as well as the attributes to be set.
|
|
For example:
|
|
|
|
PATCH /api/dcim/sites/
|
|
[
|
|
{
|
|
"id": 123,
|
|
"name": "New name"
|
|
},
|
|
{
|
|
"id": 456,
|
|
"status": "planned"
|
|
}
|
|
]
|
|
"""
|
|
def bulk_update(self, request, *args, **kwargs):
|
|
partial = kwargs.pop('partial', False)
|
|
serializer = BulkOperationSerializer(data=request.data, many=True)
|
|
serializer.is_valid(raise_exception=True)
|
|
qs = self.get_queryset().filter(
|
|
pk__in=[o['id'] for o in serializer.data]
|
|
)
|
|
|
|
# Map update data by object ID
|
|
update_data = {
|
|
obj.pop('id'): obj for obj in request.data
|
|
}
|
|
|
|
data = self.perform_bulk_update(qs, update_data, partial=partial)
|
|
|
|
return Response(data, status=status.HTTP_200_OK)
|
|
|
|
def perform_bulk_update(self, objects, update_data, partial):
|
|
with transaction.atomic():
|
|
data_list = []
|
|
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)
|
|
data_list.append(serializer.data)
|
|
|
|
return data_list
|
|
|
|
def bulk_partial_update(self, request, *args, **kwargs):
|
|
kwargs['partial'] = True
|
|
return self.bulk_update(request, *args, **kwargs)
|
|
|
|
|
|
class BulkDestroyModelMixin:
|
|
"""
|
|
Support bulk deletion of objects using the list endpoint for a model. Accepts a DELETE action with a list of one
|
|
or more JSON objects, each specifying the numeric ID of an object to be deleted. For example:
|
|
|
|
DELETE /api/dcim/sites/
|
|
[
|
|
{"id": 123},
|
|
{"id": 456}
|
|
]
|
|
"""
|
|
def bulk_destroy(self, request, *args, **kwargs):
|
|
serializer = BulkOperationSerializer(data=request.data, many=True)
|
|
serializer.is_valid(raise_exception=True)
|
|
qs = self.get_queryset().filter(
|
|
pk__in=[o['id'] for o in serializer.data]
|
|
)
|
|
|
|
self.perform_bulk_destroy(qs)
|
|
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
def perform_bulk_destroy(self, objects):
|
|
with transaction.atomic():
|
|
for obj in objects:
|
|
if hasattr(obj, 'snapshot'):
|
|
obj.snapshot()
|
|
self.perform_destroy(obj)
|
|
|
|
|
|
class ObjectValidationMixin:
|
|
|
|
def _validate_objects(self, instance):
|
|
"""
|
|
Check that the provided instance or list of instances are matched by the current queryset. This confirms that
|
|
any newly created or modified objects abide by the attributes granted by any applicable ObjectPermissions.
|
|
"""
|
|
if type(instance) is list:
|
|
# Check that all instances are still included in the view's queryset
|
|
conforming_count = self.queryset.filter(pk__in=[obj.pk for obj in instance]).count()
|
|
if conforming_count != len(instance):
|
|
raise ObjectDoesNotExist
|
|
elif not self.queryset.filter(pk=instance.pk).exists():
|
|
raise ObjectDoesNotExist
|