Coverage for core/api/schema.py: 88%
232 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import copy
2import re
3import typing
4from collections import OrderedDict
6from django.core.exceptions import ImproperlyConfigured
7from django.utils.translation import gettext_lazy as _
8from drf_spectacular.contrib.django_filters import DjangoFilterExtension
9from drf_spectacular.extensions import OpenApiSerializerExtension, OpenApiSerializerFieldExtension, _SchemaType
10from drf_spectacular.openapi import AutoSchema
11from drf_spectacular.plumbing import (
12 build_basic_type,
13 build_choice_field,
14 build_media_type_object,
15 build_object_type,
16 follow_field_source,
17 get_doc,
18)
19from drf_spectacular.types import OpenApiTypes
20from drf_spectacular.utils import Direction, OpenApiParameter, OpenApiResponse
21from rest_framework.fields import ReadOnlyField
22from rest_framework.utils import model_meta
24from netbox.api.fields import ChoiceField
25from netbox.api.serializers import BulkOperationErrorSerializer, WritableNestedSerializer
26from netbox.api.viewsets import NetBoxModelViewSet
28# see netbox.api.routers.NetBoxRouter
29BULK_ACTIONS = ("bulk_destroy", "bulk_partial_update", "bulk_update")
30WRITABLE_ACTIONS = ("PATCH", "POST", "PUT")
33class NetBoxDjangoFilterExtension(DjangoFilterExtension):
34 """
35 Overrides drf-spectacular's DjangoFilterExtension to fix a regression in v0.29.0 where
36 _get_model_field() incorrectly double-appends to_field_name when field_name already ends
37 with that value (e.g. field_name='tags__slug', to_field_name='slug' produces the invalid
38 path ['tags', 'slug', 'slug']). This caused hundreds of spurious warnings during schema
39 generation for filters such as TagFilter, TenancyFilterSet.tenant, and OwnerFilterMixin.owner.
41 See: https://github.com/netbox-community/netbox/issues/20787
42 https://github.com/tfranzel/drf-spectacular/issues/1475
43 """
44 priority = 1
46 def _get_model_field(self, filter_field, model):
47 if not filter_field.field_name: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true
48 return None
49 path = filter_field.field_name.split('__')
50 to_field_name = filter_field.extra.get('to_field_name')
51 if to_field_name is not None and path[-1] != to_field_name:
52 path.append(to_field_name)
53 return follow_field_source(model, path, emit_warnings=False)
56class FixTimeZoneSerializerField(OpenApiSerializerFieldExtension):
57 target_class = 'timezone_field.rest_framework.TimeZoneSerializerField'
59 def map_serializer_field(self, auto_schema, direction):
60 return build_basic_type(OpenApiTypes.STR)
63class ChoiceFieldFix(OpenApiSerializerFieldExtension):
64 target_class = 'netbox.api.fields.ChoiceField'
66 def map_serializer_field(self, auto_schema, direction):
67 build_cf = build_choice_field(self.target)
69 if direction == 'request':
70 return build_cf
72 if direction == "response": 72 ↛ 88line 72 didn't jump to line 88 because the condition on line 72 was always true
73 value = build_cf
74 label = {
75 **build_basic_type(OpenApiTypes.STR),
76 "enum": list(OrderedDict.fromkeys(self.target.choices.values()))
77 }
79 return build_object_type(
80 properties={
81 "value": value,
82 "label": label
83 }
84 )
86 # TODO: This function should never implicitly/explicitly return `None`
87 # The fallback should be well-defined (drf-spectacular expects request/response naming).
88 return None
91def viewset_handles_bulk_create(view):
92 """Check if view automatically provides list-based bulk create"""
93 return isinstance(view, NetBoxModelViewSet)
96class NetBoxAutoSchema(AutoSchema):
97 """
98 Overrides to drf_spectacular.openapi.AutoSchema to fix following issues:
99 1. bulk serializers cause operation_id conflicts with non-bulk ones
100 2. bulk operations should specify a list
101 3. bulk operations don't have filter params
102 4. bulk operations don't have pagination
103 5. bulk delete should specify input
104 """
106 writable_serializers = {}
108 @property
109 def is_bulk_action(self):
110 if hasattr(self.view, "action") and self.view.action in BULK_ACTIONS:
111 return True
112 return False
114 def get_operation_id(self):
115 """
116 bulk serializers cause operation_id conflicts with non-bulk ones
117 bulk operations cause id conflicts in spectacular resulting in numerous:
118 Warning: operationId "xxx" has collisions [xxx]. "resolving with numeral suffixes"
119 code is modified from drf_spectacular.openapi.AutoSchema.get_operation_id
120 """
121 if self.is_bulk_action:
122 tokenized_path = self._tokenize_path()
123 # replace dashes as they can be problematic later in code generation
124 tokenized_path = [t.replace('-', '_') for t in tokenized_path]
126 if self.method == 'GET' and self._is_list_view(): 126 ↛ 128line 126 didn't jump to line 128 because the condition on line 126 was never true
127 # this shouldn't happen, but keeping it here to follow base code
128 action = 'list'
129 else:
130 # action = self.method_mapping[self.method.lower()]
131 # use bulk name so partial_update -> bulk_partial_update
132 action = self.view.action.lower()
134 if not tokenized_path: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 tokenized_path.append('root')
137 if re.search(r'<drf_format_suffix\w*:\w+>', self.path_regex): 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 tokenized_path.append('formatted')
140 return '_'.join(tokenized_path + [action])
142 # if not bulk - just return normal id
143 return super().get_operation_id()
145 def get_request_serializer(self) -> typing.Any:
146 serializer = super().get_request_serializer()
148 # Bulk update/partial-update has a special request shape: a list of
149 # writable objects plus a required `id` field. The normal writable
150 # serializer omits `id` because it is read-only, so don't use the generic
151 # bulk handling for these actions.
152 action = getattr(self.view, 'action', None)
153 if action in ('bulk_update', 'bulk_partial_update'):
154 get_bulk_update_request_serializer = getattr(
155 self.view,
156 'get_bulk_update_request_serializer',
157 None,
158 )
159 if get_bulk_update_request_serializer is not None: 159 ↛ 165line 159 didn't jump to line 165 because the condition on line 159 was always true
160 return get_bulk_update_request_serializer(
161 partial=(action == 'bulk_partial_update' or self.method == 'PATCH')
162 )
164 # Bulk creates/deletes should specify a list.
165 if self.is_bulk_action:
166 return type(serializer)(many=True)
168 # handle mapping for Writable serializers - adapted from dansheps original
169 # code for drf-yasg.
170 if serializer is not None and self.method in WRITABLE_ACTIONS: 170 ↛ 179line 170 didn't jump to line 179 because the condition on line 170 was always true
171 writable_class = self.get_writable_class(serializer)
172 if writable_class is not None:
173 if hasattr(serializer, "child"): 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 child_serializer = self.get_writable_class(serializer.child)
175 serializer = writable_class(context=serializer.context, child=child_serializer)
176 else:
177 serializer = writable_class(context=serializer.context)
179 return serializer
181 def get_response_serializers(self) -> typing.Any:
182 # bulk operations should specify a list
183 response_serializers = super().get_response_serializers()
185 if self.is_bulk_action:
186 return type(response_serializers)(many=True)
188 return response_serializers
190 def _get_bulk_error_responses(self, direction) -> typing.Any:
191 """
192 Return the error responses of the current bulk write action, keyed by status code, or an
193 empty dict if this action is not a bulk write.
195 A failed bulk write returns a structured body correlating each failure with the object (or,
196 where no object could be identified, the request position) responsible for it. This is a
197 documented part of the API contract, but drf-spectacular cannot infer it: responses are
198 derived from the request/response serializer alone, which describes only the success case.
199 """
200 action = getattr(self.view, 'action', None)
202 if action in ('bulk_update', 'bulk_partial_update'):
203 return {
204 '400': OpenApiResponse(
205 response=BulkOperationErrorSerializer,
206 description=_(
207 "One or more of the objects specified could not be updated. No objects were "
208 "modified: a bulk update is an all-or-none operation."
209 ),
210 ),
211 '403': OpenApiResponse(
212 response=BulkOperationErrorSerializer,
213 description=_(
214 "The requesting user is not permitted to apply one or more of the "
215 "modifications specified. No objects were modified."
216 ),
217 ),
218 }
220 if action == 'bulk_destroy':
221 return {
222 '400': OpenApiResponse(
223 response=BulkOperationErrorSerializer,
224 description=_(
225 "The request was malformed, one or more of the objects specified could not "
226 "be found, or the deletion of one of them was prevented by a protection "
227 "rule. No objects were deleted."
228 ),
229 ),
230 '403': OpenApiResponse(
231 response=BulkOperationErrorSerializer,
232 description=_(
233 "The requesting user is not permitted to delete one or more of the objects "
234 "specified. No objects were deleted."
235 ),
236 ),
237 '409': OpenApiResponse(
238 response=BulkOperationErrorSerializer,
239 description=_(
240 "One or more of the objects specified could not be deleted, because a "
241 "dependent object prevents it. No objects were deleted: a bulk deletion is "
242 "an all-or-none operation."
243 ),
244 ),
245 }
247 if action == 'create' and viewset_handles_bulk_create(self.view):
248 # A POST to a list endpoint accepts either a single object or a list of them (see
249 # _get_request_for_media_type()), so its error body takes one of two shapes
250 # accordingly: field-keyed errors for a single object, or the bulk envelope for a list.
251 component = self.resolve_serializer(BulkOperationErrorSerializer, direction)
252 return {
253 '400': OpenApiResponse(
254 response={
255 'oneOf': [
256 build_basic_type(OpenApiTypes.OBJECT),
257 component.ref if component else build_basic_type(OpenApiTypes.OBJECT),
258 ],
259 },
260 description=_(
261 "The object could not be created. Where a list was submitted, no objects "
262 "were created: a bulk creation is an all-or-none operation."
263 ),
264 ),
265 # A 403 always carries a `detail`, and BulkOperationError's `errors` is optional, so
266 # the one component covers both the single-object and the bulk shape here.
267 '403': OpenApiResponse(
268 response=BulkOperationErrorSerializer,
269 description=_(
270 "The requesting user is not permitted to create one or more of the objects "
271 "specified. No objects were created."
272 ),
273 ),
274 }
276 return {}
278 def _get_response_bodies(self, direction='response') -> typing.Any:
279 responses = super()._get_response_bodies(direction=direction)
281 # Document the error responses of the bulk write actions, which cannot be inferred (see
282 # _get_bulk_error_responses). A status code already present -- for instance one declared
283 # via @extend_schema on a custom action -- is left as it is.
284 for code, response in self._get_bulk_error_responses(direction).items():
285 if code not in responses: 285 ↛ 284line 285 didn't jump to line 284 because the condition on line 285 was always true
286 responses[code] = self._get_response_for_code(response, code, direction=direction)
288 return responses
290 def _get_request_for_media_type(self, serializer, direction='request'):
291 """
292 Override to generate oneOf schema for serializers that support both
293 single object and array input (NetBoxModelViewSet POST operations).
295 Refs: #20638
296 """
297 # Get the standard schema first
298 schema, required = super()._get_request_for_media_type(serializer, direction)
300 # If this serializer supports arrays (marked in get_request_serializer),
301 # wrap the schema in oneOf to allow single object OR array
302 if (
303 direction == 'request' and
304 schema is not None and
305 getattr(self.view, 'action', None) == 'create' and
306 viewset_handles_bulk_create(self.view)
307 ):
308 return {
309 'oneOf': [
310 schema, # Single object
311 {
312 'type': 'array',
313 'items': schema, # Array of objects
314 }
315 ]
316 }, required
318 return schema, required
320 def _get_serializer_name(self, serializer, direction, bypass_extensions=False) -> str:
321 name = super()._get_serializer_name(serializer, direction, bypass_extensions)
323 # If this serializer is nested, prepend its name with "Brief". Serializers which declare an explicit
324 # Meta.ref_name are exempt: those are brief by design and have no complete form in the schema, so the
325 # prefix would only rename an existing component to no purpose. See #22989.
326 if getattr(serializer, 'nested', False) and not getattr(getattr(serializer, 'Meta', None), 'ref_name', None):
327 name = f'Brief{name}'
329 return name
331 def get_serializer_ref_name(self, serializer):
332 # from drf-yasg.utils
333 """Get serializer's ref_name
334 :param serializer: Serializer instance
335 :return: Serializer's ``ref_name`` or ``None`` for inline serializer
336 :rtype: str or None
337 """
338 serializer_meta = getattr(serializer, 'Meta', None)
339 serializer_name = type(serializer).__name__
340 if hasattr(serializer_meta, 'ref_name'): 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 ref_name = serializer_meta.ref_name
342 else:
343 ref_name = serializer_name
344 if ref_name.endswith('Serializer'): 344 ↛ 346line 344 didn't jump to line 346 because the condition on line 344 was always true
345 ref_name = ref_name[: -len('Serializer')]
346 return ref_name
348 @staticmethod
349 def _rebuilds_as_writable(serializer, field_name):
350 """
351 Return True if DRF would rebuild the named field in writable form if the field declared on
352 the serializer class were removed (see get_writable_class()).
354 This defers to ModelSerializer.build_field(), which is what get_fields() itself calls for
355 any field not explicitly declared on the class -- rather than testing the model for a field
356 of that name, which is a weaker condition. A name backed only by a model property, by a
357 non-editable model field, or by a generic foreign key (which lives in Meta.private_fields
358 and so is absent from DRF's field info) is rebuilt read-only, and is then dropped from the
359 request body altogether.
360 """
361 model = getattr(getattr(serializer, 'Meta', None), 'model', None)
362 if model is None or not hasattr(serializer, 'build_field'): 362 ↛ 363line 362 didn't jump to line 363 because the condition on line 362 was never true
363 return False
365 depth = getattr(serializer.Meta, 'depth', 0)
366 try:
367 field_class, field_kwargs = serializer.build_field(
368 field_name, model_meta.get_field_info(model), model, depth
369 )
370 except ImproperlyConfigured:
371 # build_unknown_field(): the model has nothing of this name at all
372 return False
374 if isinstance(field_class, type) and issubclass(field_class, ReadOnlyField):
375 return False
376 return not field_kwargs.get('read_only', False)
378 def get_writable_class(self, serializer):
379 properties = {}
380 fields = {} if hasattr(serializer, 'child') else serializer.fields
381 remove_fields = []
383 # If you get a failure here for "AttributeError: 'cached_property' object has no attribute 'items'"
384 # it is probably because you are using a viewsets.ViewSet for the API View and are defining a
385 # serializer_class. You will also need to define a get_serializer() method like for GenericAPIView.
386 for child_name, child in fields.items():
387 # read_only fields don't need to be in writable (write only) serializers
388 if 'read_only' in dir(child) and child.read_only:
389 remove_fields.append(child_name)
390 if isinstance(child, (ChoiceField, WritableNestedSerializer)):
391 if child.read_only or self._rebuilds_as_writable(serializer, child_name):
392 properties[child_name] = None
393 else:
394 # DRF cannot rebuild this one writably: it is backed by a read-only property
395 # (e.g. Service.protocol, derived from port_mappings). Nulling it would leave
396 # DRF to rebuild it as a ReadOnlyField, which is then omitted from the request
397 # body altogether -- silently dropping a field the serializer does accept on
398 # write. Keep the declared field instead; ChoiceFieldFix already renders it
399 # correctly for the request direction. The copy leaves the bound original
400 # untouched (Field.__deepcopy__ returns an unbound field built from the same
401 # arguments), and keeps `properties` non-empty so the writable variant is still
402 # generated rather than collapsing to None below.
403 properties[child_name] = copy.deepcopy(child)
405 if not properties:
406 return None
408 if type(serializer) not in self.writable_serializers:
409 writable_name = 'Writable' + type(serializer).__name__
410 meta_class = getattr(type(serializer), 'Meta', None)
411 if meta_class: 411 ↛ 421line 411 didn't jump to line 421 because the condition on line 411 was always true
412 ref_name = 'Writable' + self.get_serializer_ref_name(serializer)
413 # remove read_only fields from write-only serializers
414 fields = list(meta_class.fields)
415 for field in remove_fields:
416 fields.remove(field)
417 writable_meta = type('Meta', (meta_class,), {'ref_name': ref_name, 'fields': fields})
419 properties['Meta'] = writable_meta
421 self.writable_serializers[type(serializer)] = type(writable_name, (type(serializer),), properties)
423 writable_class = self.writable_serializers[type(serializer)]
424 return writable_class
426 def get_override_parameters(self):
427 params = super().get_override_parameters()
428 # Expose the ?fields, ?omit, and ?brief query parameters supported by NetBoxModelViewSet
429 # for all non-bulk GET operations (both list and detail).
430 if not self.is_bulk_action and self.method == 'GET':
431 params = list(params) + [
432 OpenApiParameter(
433 name='fields',
434 location=OpenApiParameter.QUERY,
435 required=False,
436 type=OpenApiTypes.STR,
437 description='Comma-separated list of fields to include in the response. Example: `fields=id,name`.',
438 ),
439 OpenApiParameter(
440 name='omit',
441 location=OpenApiParameter.QUERY,
442 required=False,
443 type=OpenApiTypes.STR,
444 description='Comma-separated list of fields to exclude from the response. '
445 'Example: `omit=description,tags`.',
446 ),
447 OpenApiParameter(
448 name='brief',
449 location=OpenApiParameter.QUERY,
450 required=False,
451 type=OpenApiTypes.BOOL,
452 description='Return only brief fields for each object.',
453 ),
454 ]
455 return params
457 def get_filter_backends(self):
458 # bulk operations don't have filter params
459 if self.is_bulk_action:
460 return []
461 return super().get_filter_backends()
463 def _get_paginator(self):
464 # bulk operations don't have pagination
465 if self.is_bulk_action:
466 return None
467 return super()._get_paginator()
469 def _get_request_body(self, direction='request'):
470 # bulk delete should specify input
471 if (not self.is_bulk_action) or (self.method != 'DELETE'):
472 return super()._get_request_body(direction)
474 # rest from drf_spectacular.openapi.AutoSchema._get_request_body
475 # but remove the unsafe method check
477 request_serializer = self.get_request_serializer()
479 if isinstance(request_serializer, dict): 479 ↛ 480line 479 didn't jump to line 480 because the condition on line 479 was never true
480 content = []
481 request_body_required = True
482 for media_type, serializer in request_serializer.items():
483 schema, partial_request_body_required = self._get_request_for_media_type(serializer, direction)
484 examples = self._get_examples(serializer, direction, media_type)
485 if schema is None:
486 continue
487 content.append((media_type, schema, examples))
488 request_body_required &= partial_request_body_required
489 else:
490 schema, request_body_required = self._get_request_for_media_type(request_serializer, direction)
491 if schema is None: 491 ↛ 492line 491 didn't jump to line 492 because the condition on line 491 was never true
492 return None
493 content = [
494 (media_type, schema, self._get_examples(request_serializer, direction, media_type))
495 for media_type in self.map_parsers()
496 ]
498 request_body = {
499 'content': {
500 media_type: build_media_type_object(schema, examples) for media_type, schema, examples in content
501 }
502 }
503 if request_body_required: 503 ↛ 505line 503 didn't jump to line 505 because the condition on line 503 was always true
504 request_body['required'] = request_body_required
505 return request_body
507 def get_description(self):
508 """
509 Return a string description for the ViewSet.
510 """
512 # If a docstring is provided, use it.
513 if self.view.__doc__:
514 return get_doc(self.view.__class__)
516 # When the action method is decorated with @action, use the docstring of the method.
517 action_or_method = getattr(self.view, getattr(self.view, 'action', self.method.lower()), None)
518 if action_or_method and action_or_method.__doc__:
519 return get_doc(action_or_method)
521 # Else, generate a description from the class name.
522 return self._generate_description()
524 def _generate_description(self):
525 """
526 Generate a docstring for the method. It also takes into account whether the method is for list or detail.
527 """
528 model_name = self.view.queryset.model._meta.verbose_name
530 # Determine if the method is for list or detail.
531 if '{id}' in self.path:
532 return f"{self.method.capitalize()} a {model_name} object."
533 return f"{self.method.capitalize()} a list of {model_name} objects."
536class FixSerializedPKRelatedField(OpenApiSerializerFieldExtension):
537 target_class = 'netbox.api.fields.SerializedPKRelatedField'
539 def map_serializer_field(self, auto_schema, direction):
540 if direction == "response":
541 # Resolve an instance of the serializer carrying the field's nested setting, so that the brief
542 # component is referenced wherever the field renders a brief representation. (The field's
543 # to_representation() passes nested in the same manner.) See #22989.
544 serializer = self.target.serializer(nested=self.target.nested)
545 component = auto_schema.resolve_serializer(serializer, direction)
546 return component.ref if component else None
547 return build_basic_type(OpenApiTypes.INT)
550class FixIntegerRangeSerializerSchema(OpenApiSerializerExtension):
551 target_class = 'netbox.api.fields.IntegerRangeSerializer'
552 match_subclasses = True
554 def map_serializer(self, auto_schema: 'AutoSchema', direction: Direction) -> _SchemaType:
555 # One range = two integers; many=True will wrap this in an outer array
556 return {
557 'type': 'array',
558 'items': {
559 'type': 'integer',
560 },
561 'minItems': 2,
562 'maxItems': 2,
563 'example': [10, 20],
564 }
567# Nested models can be passed by ID in requests
568# The logic for this is handled in `BaseModelSerializer.to_internal_value`
569class FixWritableNestedSerializerAllowPK(OpenApiSerializerFieldExtension):
570 target_class = 'netbox.api.serializers.BaseModelSerializer'
571 match_subclasses = True
573 def map_serializer_field(self, auto_schema, direction):
574 schema = auto_schema._map_serializer_field(self.target, direction, bypass_extensions=True)
575 if schema is None:
576 return schema
577 if direction == 'request' and self.target.nested:
578 return {
579 'oneOf': [
580 build_basic_type(OpenApiTypes.INT),
581 schema,
582 ]
583 }
584 return schema