Coverage for netbox/api/viewsets/__init__.py: 79%
202 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import logging
2import warnings
3from functools import cached_property
5from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
6from django.db import router, transaction
7from django.db.models import ProtectedError, RestrictedError
8from rest_framework import mixins as drf_mixins
9from rest_framework import status
10from rest_framework.exceptions import MethodNotAllowed
11from rest_framework.response import Response
12from rest_framework.viewsets import GenericViewSet
14from netbox.api.serializers.features import ChangeLogMessageSerializer
15from utilities.api import get_annotations_for_serializer, get_prefetches_for_serializer
16from utilities.exceptions import AbortRequest, PreconditionFailed
17from utilities.query import reapply_model_ordering
19from . import mixins
21__all__ = (
22 'MPTTLockedMixin',
23 'NetBoxModelViewSet',
24 'NetBoxReadOnlyModelViewSet',
25)
27HTTP_ACTIONS = {
28 'GET': 'view',
29 'OPTIONS': None,
30 'HEAD': 'view',
31 'POST': 'add',
32 'PUT': 'change',
33 'PATCH': 'change',
34 'DELETE': 'delete',
35}
38class ETagMixin:
39 """
40 Adds ETag header support to ViewSets. Generates weak ETags (W/ prefix per
41 RFC 7232 §2.1) from `last_updated` (or `created` if unavailable). Weak ETags
42 are appropriate here because the tag is derived from a modification timestamp
43 rather than a hash of the serialized payload.
44 """
46 @staticmethod
47 def _get_etag(obj):
48 """Return a weak ETag string for the given object, or None."""
49 if ts := getattr(obj, 'last_updated', None) or getattr(obj, 'created', None):
50 return f'W/"{ts.isoformat()}"'
51 return None
53 @staticmethod
54 def _get_if_match(request):
55 """Return the list of If-Match header values (if specified)."""
56 if (if_match := request.META.get('HTTP_IF_MATCH')) and if_match != '*': 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 return [e.strip() for e in if_match.split(',')]
58 return []
60 def _validate_etag(self, request, instance):
61 """Validate the request's ETag"""
62 if provided := self._get_if_match(request): 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true
63 current_etag = self._get_etag(instance)
64 if current_etag and current_etag not in provided:
65 raise PreconditionFailed(etag=current_etag)
67 def handle_exception(self, exc):
68 response = super().handle_exception(exc)
69 if isinstance(exc, PreconditionFailed) and exc.etag: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true
70 response['ETag'] = exc.etag
71 return response
73 def retrieve(self, request, *args, **kwargs):
74 instance = self.get_object()
75 serializer = self.get_serializer(instance)
76 response = Response(serializer.data)
77 if etag := self._get_etag(instance):
78 response['ETag'] = etag
79 return response
82class BaseViewSet(GenericViewSet):
83 """
84 Base class for all API ViewSets. This is responsible for the enforcement of object-based permissions.
85 """
86 brief = False
88 def initial(self, request, *args, **kwargs):
89 super().initial(request, *args, **kwargs)
91 # Reject any method for which no action has been declared, rather than proceeding against an
92 # unrestricted QuerySet. (A method mapped to None, e.g. OPTIONS, is permitted: it needs no
93 # restriction.) This is the same 405 DRF would return when resolving the handler for an unmapped
94 # method, but it also covers a handler bound to such a method (e.g. @action(methods=['trace'])).
95 if request.method not in HTTP_ACTIONS: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true
96 raise MethodNotAllowed(request.method)
98 # Restrict the view's QuerySet to allow only the permitted objects
99 if request.user.is_authenticated: 99 ↛ exitline 99 didn't return from function 'initial' because the condition on line 99 was always true
100 if action := HTTP_ACTIONS[request.method]:
101 self.queryset = self.queryset.restrict(request.user, action)
103 def initialize_request(self, request, *args, **kwargs):
105 # Annotate whether brief mode is active
106 self.brief = request.method == 'GET' and request.GET.get('brief')
108 return super().initialize_request(request, *args, **kwargs)
110 def get_queryset(self):
111 qs = super().get_queryset()
112 serializer_class = self.get_serializer_class()
114 # Dynamically resolve prefetches for included serializer fields and attach them to the queryset
115 if prefetch := get_prefetches_for_serializer(serializer_class, **self.field_kwargs):
116 qs = qs.prefetch_related(*prefetch)
118 # Dynamically resolve annotations for RelatedObjectCountFields on the serializer and attach them to the queryset
119 if annotations := get_annotations_for_serializer(serializer_class, **self.field_kwargs):
120 qs = qs.annotate(**annotations)
122 return qs
124 def get_serializer(self, *args, **kwargs):
125 # Pass the fields/omit kwargs (if specified by the request) to the serializer
126 kwargs.update(**self.field_kwargs)
127 return super().get_serializer(*args, **kwargs)
129 @cached_property
130 def field_kwargs(self):
131 """Return a dictionary of keyword arguments to be passed when instantiating the serializer."""
132 # An explicit list of fields was requested
133 if requested_fields := self.request.query_params.get('fields'):
134 return {'fields': requested_fields.split(',')}
136 # An explicit list of fields to omit was requested
137 if omit_fields := self.request.query_params.get('omit'):
138 return {'omit': omit_fields.split(',')}
140 # Brief mode has been enabled for this request
141 if self.brief:
142 serializer_class = self.get_serializer_class()
143 if brief_fields := getattr(serializer_class.Meta, 'brief_fields', None):
144 return {'fields': brief_fields}
146 return {}
149class NetBoxReadOnlyModelViewSet(
150 ETagMixin,
151 mixins.CustomFieldsMixin,
152 mixins.ExportTemplatesMixin,
153 drf_mixins.RetrieveModelMixin,
154 drf_mixins.ListModelMixin,
155 BaseViewSet
156):
157 pass
160class NetBoxModelViewSet(
161 ETagMixin,
162 mixins.BackgroundOperationMixin,
163 mixins.BulkCreateModelMixin,
164 mixins.BulkUpdateModelMixin,
165 mixins.BulkDestroyModelMixin,
166 mixins.ObjectValidationMixin,
167 mixins.CustomFieldsMixin,
168 mixins.ExportTemplatesMixin,
169 drf_mixins.CreateModelMixin,
170 drf_mixins.RetrieveModelMixin,
171 drf_mixins.UpdateModelMixin,
172 drf_mixins.DestroyModelMixin,
173 drf_mixins.ListModelMixin,
174 BaseViewSet
175):
176 """
177 Extend DRF's ModelViewSet to support bulk update and delete functions.
178 """
179 def get_object_with_snapshot(self):
180 """
181 Save a pre-change snapshot of the object immediately after retrieving it. This snapshot will be used to
182 record the "before" data in the changelog.
183 """
184 obj = super().get_object()
185 if hasattr(obj, 'snapshot'):
186 obj.snapshot()
187 return obj
189 def get_queryset(self):
190 qs = super().get_queryset()
191 return reapply_model_ordering(qs)
193 def get_serializer(self, *args, **kwargs):
194 # If a list of objects has been provided, initialize the serializer with many=True
195 if isinstance(kwargs.get('data', {}), list): 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true
196 kwargs['many'] = True
198 return super().get_serializer(*args, **kwargs)
200 def dispatch(self, request, *args, **kwargs):
201 logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
203 try:
204 return super().dispatch(request, *args, **kwargs)
205 except (ProtectedError, RestrictedError) as e:
206 if type(e) is ProtectedError: 206 ↛ 209line 206 didn't jump to line 209 because the condition on line 206 was always true
207 protected_objects = list(e.protected_objects)
208 else:
209 protected_objects = list(e.restricted_objects)
210 msg = f'Unable to delete object. {len(protected_objects)} dependent objects were found: '
211 msg += ', '.join([f'{obj} ({obj.pk})' for obj in protected_objects])
212 logger.warning(msg)
213 return self.finalize_response(
214 request,
215 Response({'detail': msg}, status=409),
216 *args,
217 **kwargs
218 )
219 except AbortRequest as e:
220 logger.debug(e.message)
221 return self.finalize_response(
222 request,
223 Response({'detail': e.message}, status=400),
224 *args,
225 **kwargs
226 )
228 def exception_to_response(self, exc):
229 """
230 Translate a NetBox/Django exception that is not a DRF APIException into the same
231 Response that dispatch() would return for it. Returns None if the exception is not
232 one this method handles (the caller should then re-raise or defer to DRF).
234 This mirrors the except clauses in dispatch(); it is also called by the background
235 job runner (netbox.jobs.AsyncAPIJob), which executes action methods directly and so
236 bypasses dispatch(). NOTE: dispatch() does not yet call this helper itself; the two
237 should be consolidated into a single source of truth in a future change.
238 """
239 logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
240 if isinstance(exc, (ProtectedError, RestrictedError)):
241 if type(exc) is ProtectedError:
242 protected_objects = list(exc.protected_objects)
243 else:
244 protected_objects = list(exc.restricted_objects)
245 msg = f'Unable to delete object. {len(protected_objects)} dependent objects were found: '
246 msg += ', '.join([f'{obj} ({obj.pk})' for obj in protected_objects])
247 logger.warning(msg)
248 return Response({'detail': msg}, status=409)
249 if isinstance(exc, AbortRequest):
250 logger.debug(exc.message)
251 return Response({'detail': exc.message}, status=400)
252 return None
254 # Creates
256 def create(self, request, *args, **kwargs):
257 # If background processing was requested for a bulk (list) create, enqueue a job and
258 # return immediately. Single-object creates always run synchronously.
259 if (response := self._handle_background_request(request, 'create')) is not None: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 return response
262 # Creating multiple objects, which are validated and saved one at a time in order to
263 # collect per-object errors (see BulkCreateModelMixin)
264 if isinstance(request.data, list):
265 return self.bulk_create(request, *args, **kwargs)
267 serializer = self.get_serializer(data=request.data)
268 serializer.is_valid(raise_exception=True)
269 self.perform_create(serializer)
271 # After creating the instance, re-initialize the serializer with a queryset
272 # to ensure related objects are prefetched.
273 qs = self.get_queryset().get(pk=serializer.instance.pk)
275 # Re-serialize the instance with prefetched data
276 serializer = self.get_serializer(qs)
278 headers = self.get_success_headers(serializer.data)
279 response = Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
281 if etag := self._get_etag(qs):
282 response['ETag'] = etag
284 return response
286 def perform_create(self, serializer):
287 model = self.queryset.model
288 logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
289 logger.info(f"Creating new {model._meta.verbose_name}")
291 # Enforce object-level permissions on save()
292 using = router.db_for_write(model)
293 try:
294 with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
295 instance = serializer.save()
296 self._validate_objects(instance)
297 except ObjectDoesNotExist:
298 raise PermissionDenied()
300 # Updates
302 def update(self, request, *args, **kwargs):
303 partial = kwargs.pop('partial', False)
304 instance = self.get_object_with_snapshot()
306 # Enforce If-Match precondition (RFC 9110 §13.1.1)
307 self._validate_etag(self.request, instance)
309 serializer = self.get_serializer(instance, data=request.data, partial=partial)
310 serializer.is_valid(raise_exception=True)
311 self.perform_update(serializer)
313 # After updating the instance, re-initialize the serializer with a queryset
314 # to ensure related objects are prefetched.
315 qs = self.get_queryset().get(pk=serializer.instance.pk)
317 # Re-serialize the instance(s) with prefetched data
318 serializer = self.get_serializer(qs)
319 response = Response(serializer.data)
321 if etag := self._get_etag(qs):
322 response['ETag'] = etag
324 return response
326 def perform_update(self, serializer):
327 model = self.queryset.model
328 logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
329 logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})")
331 # Enforce object-level permissions on save()
332 using = router.db_for_write(model)
333 try:
334 with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
335 # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window
336 # between the initial check in update() and the actual write.
337 if self._get_if_match(self.request): 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 locked = model.objects.select_for_update().get(pk=serializer.instance.pk)
339 self._validate_etag(self.request, locked)
340 instance = serializer.save()
341 self._validate_objects(instance)
342 except ObjectDoesNotExist:
343 raise PermissionDenied()
345 # Deletes
347 def destroy(self, request, *args, **kwargs):
348 instance = self.get_object_with_snapshot()
350 # Enforce If-Match precondition (RFC 9110 §13.1.1)
351 self._validate_etag(request, instance)
353 # Attach changelog message (if any)
354 serializer = ChangeLogMessageSerializer(data=request.data)
355 serializer.is_valid(raise_exception=True)
356 instance._changelog_message = serializer.validated_data.get('changelog_message')
358 self.perform_destroy(instance)
360 return Response(status=status.HTTP_204_NO_CONTENT)
362 def perform_destroy(self, instance):
363 model = self.queryset.model
364 logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
365 logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})")
367 using = router.db_for_write(model)
368 try:
369 with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
370 # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window
371 # between the initial check in destroy() and the actual delete.
372 if self._get_if_match(self.request): 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 locked = model.objects.select_for_update().get(pk=instance.pk)
374 self._validate_etag(self.request, locked)
375 super().perform_destroy(instance)
376 except ObjectDoesNotExist:
377 raise PermissionDenied()
380# TODO: Remove this in NetBox v5.0
381class MPTTLockedMixin:
382 """
383 Deprecated no-op mixin retained for backward compatibility.
385 Historically this acquired a pglock around create/update/destroy to serialize
386 concurrent writes to MPTT-based tree models. NetBox no longer uses MPTT: tree
387 integrity is now maintained by the PostgreSQL ltree triggers (see
388 `utilities.ltree`), which take per-tree advisory locks at the database level.
389 This mixin is therefore now a transparent pass-through and may be removed in a
390 future release. Plugins should stop inheriting from it.
391 """
393 def __init_subclass__(cls, **kwargs):
394 super().__init_subclass__(**kwargs)
395 warnings.warn(
396 "MPTTLockedMixin is deprecated and no longer does anything; tree write "
397 "concurrency is now handled by ltree database triggers. Remove it from "
398 f"{cls.__name__}.",
399 DeprecationWarning,
400 stacklevel=2,
401 )