Coverage for utilities/api.py: 80%
141 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import logging
3from django.contrib.contenttypes.fields import GenericForeignKey
4from django.core.exceptions import (
5 FieldDoesNotExist,
6 FieldError,
7 MultipleObjectsReturned,
8 ObjectDoesNotExist,
9 ValidationError,
10)
11from django.db.models.fields.related import ManyToManyRel, ManyToOneRel, RelatedField
12from django.urls import reverse
13from django.utils.module_loading import import_string
14from django.utils.translation import gettext_lazy as _
15from rest_framework.permissions import BasePermission
16from rest_framework.relations import ManyRelatedField
17from rest_framework.serializers import ListSerializer, Serializer
18from rest_framework.views import get_view_name as drf_get_view_name
20from extras.constants import HTTP_CONTENT_TYPE_JSON
21from netbox.api.exceptions import GraphQLTypeNotFound, SerializerNotFound
22from netbox.api.fields import RelatedObjectCountField, SerializedPKRelatedField
23from netbox.registry import registry
25from .query import count_related, dict_to_filter_params
26from .string import title
28logger = logging.getLogger('netbox.utilities.api')
30__all__ = (
31 'IsSuperuser',
32 'get_annotations_for_serializer',
33 'get_graphql_type_for_model',
34 'get_positional_errors',
35 'get_prefetches_for_serializer',
36 'get_related_object_by_attrs',
37 'get_serializer_for_model',
38 'get_view_name',
39 'is_api_request',
40 'is_graphql_request',
41)
44class IsSuperuser(BasePermission):
45 """
46 Allows access only to superusers.
47 """
48 def has_permission(self, request, view):
49 return bool(request.user and request.user.is_active and request.user.is_superuser)
52def get_serializer_for_model(model, prefix=''):
53 """
54 Return the appropriate REST API serializer for the given model.
56 A plugin (or internal app) may register a custom resolver for its own
57 app via netbox.plugins.register_serializer_resolver() to handle
58 dynamically generated models or to override serializer resolution. If
59 a resolver is registered for the model's app and returns a Serializer
60 subclass, that result is used. Otherwise, the default import-path
61 lookup runs.
62 """
63 app_label, model_name = model._meta.label.split('.')
65 if resolver := registry['serializer_resolvers'].get(app_label): 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true
66 try:
67 serializer = resolver(model, prefix=prefix)
68 except Exception:
69 # A buggy resolver must not break serializer lookup for the rest of NetBox.
70 logger.exception("Serializer resolver %r raised an exception; falling through to default lookup.", resolver)
71 serializer = None
72 if serializer is not None:
73 if isinstance(serializer, type) and issubclass(serializer, Serializer):
74 return serializer
75 logger.warning(
76 "Serializer resolver %r returned %r, which is not a Serializer subclass; "
77 "falling through to default lookup.",
78 resolver, serializer,
79 )
81 serializer_name = f'{app_label}.api.serializers.{prefix}{model_name}Serializer'
82 try:
83 return import_string(serializer_name)
84 except ImportError:
85 raise SerializerNotFound(
86 f"Could not determine serializer for {app_label}.{model_name} with prefix '{prefix}'"
87 )
90def get_graphql_type_for_model(model):
91 """
92 Return the GraphQL type class for the given model.
93 """
94 app_label, model_name = model._meta.label.split('.')
95 class_name = f'{app_label}.graphql.types.{model_name}Type'
96 try:
97 return import_string(class_name)
98 except ImportError:
99 raise GraphQLTypeNotFound(f"Could not find GraphQL type for {app_label}.{model_name}")
102def is_api_request(request):
103 """
104 Return True of the request is being made via the REST API.
105 """
106 return request.path_info.startswith(reverse('api-root'))
109def is_graphql_request(request):
110 """
111 Return True of the request is being made via the GraphQL API.
112 """
113 return request.path_info == reverse('graphql') and request.content_type == HTTP_CONTENT_TYPE_JSON
116def get_view_name(view):
117 """
118 Derive the view name from its associated model, if it has one. Fall back to DRF's built-in `get_view_name()`.
119 This function is provided to DRF as its VIEW_NAME_FUNCTION.
120 """
121 if hasattr(view, 'queryset') and view.queryset is not None:
122 # Derive the model name from the queryset.
123 name = title(view.queryset.model._meta.verbose_name)
124 if suffix := getattr(view, 'suffix', None):
125 name = f'{name} {suffix}'
126 return name
128 # Fall back to DRF's default behavior
129 return drf_get_view_name(view)
132def get_positional_errors(errors, count):
133 """
134 Return the errors reported by a serializer bound to a list of `count` entries as a list
135 correlated to the positions of those entries, with an empty dict standing in for each entry
136 which validated.
138 DRF 3.18 reports the errors of a ListSerializer as a mapping of the index of each failed entry
139 to that entry's errors, omitting the entries which passed; earlier releases reported a list
140 aligned with the request body. Restoring the positional form keeps the response shape stable for
141 API consumers which index into it.
143 Errors which pertain to the list as a whole rather than to any one entry (e.g. a body which is
144 not a list at all) carry no position, and are returned unchanged.
146 :param errors: The `errors` of a serializer instantiated with many=True.
147 :param count: The number of entries the serializer was bound to.
148 """
149 if not isinstance(errors, dict) or not any(isinstance(index, int) for index in errors): 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 return errors
152 return [errors.get(index, {}) for index in range(count)]
155def _get_nested_serializer(serializer_field):
156 """
157 Return the nested serializer instance for a declared serializer field.
158 """
159 if isinstance(serializer_field, ListSerializer):
160 serializer_field = serializer_field.child
162 # DRF wraps a many-valued related field, keeping the original field on child_relation
163 if isinstance(serializer_field, ManyRelatedField):
164 serializer_field = serializer_field.child_relation
166 if isinstance(serializer_field, SerializedPKRelatedField):
167 return serializer_field.serializer(nested=serializer_field.nested)
169 if isinstance(serializer_field, Serializer) and hasattr(serializer_field, 'nested'):
170 return serializer_field
172 return None
175def _get_serializer_fields(serializer: Serializer):
176 """
177 Return the effective field names for a serializer instance, honoring any
178 field-level fields=/omit= overrides.
179 """
180 fields = getattr(serializer, '_include_fields', None) or serializer.Meta.fields
181 omit = getattr(serializer, '_omit_fields', []) or []
183 return [field_name for field_name in fields if field_name not in omit]
186def get_prefetches_for_serializer(serializer_class, fields=None, omit=None, _serializer_states=None):
187 """
188 Compile and return a list of fields which should be prefetched on the queryset for a serializer.
189 """
190 if fields is not None and omit is not None: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true
191 raise TypeError("Cannot specify both 'fields' and 'omit' parameters.")
193 model = serializer_class.Meta.model
195 # If fields are not specified, default to all
196 fields_to_include = fields or serializer_class.Meta.fields
197 fields_to_omit = omit or []
198 effective_fields = tuple(name for name in fields_to_include if name not in fields_to_omit)
200 # Break reference cycles on the current path. The field set is in the key because re-entry at a
201 # narrower depth is finite, and the states are copied per frame to keep sibling fields independent.
202 serializer_states = set(_serializer_states or ())
203 serializer_state = (serializer_class, effective_fields)
204 if serializer_state in serializer_states: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true
205 return []
206 serializer_states.add(serializer_state)
208 prefetch_fields = []
209 for field_name in effective_fields:
210 serializer_field = serializer_class._declared_fields.get(field_name)
212 # Determine the name of the model field referenced by the serializer field
213 model_field_name = field_name
214 if serializer_field and getattr(serializer_field, 'source', None):
215 model_field_name = serializer_field.source
217 # If the serializer field does not map to a discrete model field, skip it.
218 try:
219 field = model._meta.get_field(model_field_name)
220 if isinstance(field, (RelatedField, ManyToOneRel, ManyToManyRel, GenericForeignKey)):
221 prefetch_fields.append(field.name)
222 except FieldDoesNotExist:
223 continue
225 # If this field is represented by a nested serializer, recurse to resolve
226 # prefetches for the related object, honoring any field-level fields=/omit=
227 # constraints set on that serializer field instance.
228 if nested_serializer := _get_nested_serializer(serializer_field):
229 subfields = _get_serializer_fields(nested_serializer)
230 for subfield in get_prefetches_for_serializer(
231 type(nested_serializer), fields=subfields, _serializer_states=serializer_states
232 ):
233 prefetch_fields.append(f'{field.name}__{subfield}')
235 return prefetch_fields
238def get_annotations_for_serializer(serializer_class, fields=None, omit=None):
239 """
240 Return a mapping of field names to annotations to be applied to the queryset for a serializer.
241 """
242 if fields is not None and omit is not None: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 raise TypeError("Cannot specify both 'fields' and 'omit' parameters.")
245 model = serializer_class.Meta.model
247 # If fields are not specified, default to all
248 fields_to_include = fields or serializer_class.Meta.fields
249 fields_to_omit = omit or []
251 annotations = {}
252 for field_name, field in serializer_class._declared_fields.items():
253 if field_name in fields_to_omit: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 continue
255 if field_name in fields_to_include and type(field) is RelatedObjectCountField:
256 related_field = getattr(model, field.relation).field
257 annotations[field_name] = count_related(related_field.model, related_field.name)
259 return annotations
262def get_related_object_by_attrs(queryset, attrs, user=None):
263 """
264 Return an object identified by either a dictionary of attributes or its numeric primary key (ID). This is used
265 for referencing related objects when creating/updating objects via the REST API.
267 When a dictionary of attributes is provided, the queryset is first restricted to only those objects on which the
268 given user has been granted view permission. This prevents an unprivileged user from enumerating objects by their
269 attributes. Referencing an object directly by its numeric ID is always permitted, regardless of the user's view
270 permissions.
272 :param queryset: The base queryset from which to retrieve the related object
273 :param attrs: A dictionary of attributes or a numeric primary key identifying the related object
274 :param user: The user making the request (used to enforce view permissions on attribute-based lookups)
275 """
276 if attrs is None: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 return None
279 # Dictionary of related object attributes
280 if isinstance(attrs, dict):
281 # Restrict the queryset to only those objects the user is permitted to view. This ensures that filtering by
282 # attributes cannot be used to enumerate objects which the user is not otherwise permitted to see. Referencing
283 # an object solely by its numeric ID (e.g. {"id": 123}) is equivalent to passing the ID directly, and is
284 # always permitted regardless of the user's view permissions.
285 if list(attrs) != ['id'] and user is not None and hasattr(queryset, 'restrict'): 285 ↛ 287line 285 didn't jump to line 287 because the condition on line 285 was always true
286 queryset = queryset.restrict(user, 'view')
287 params = dict_to_filter_params(attrs)
288 try:
289 return queryset.get(**params)
290 except ObjectDoesNotExist:
291 raise ValidationError(
292 _("Related object not found using the provided attributes: {params}").format(params=params))
293 except MultipleObjectsReturned:
294 raise ValidationError(
295 _("Multiple objects match the provided attributes: {params}").format(params=params)
296 )
297 except FieldError as e:
298 raise ValidationError(e)
300 # Integer PK of related object
301 try:
302 # Cast as integer in case a PK was mistakenly sent as a string
303 pk = int(attrs)
304 except (TypeError, ValueError):
305 raise ValidationError(
306 _(
307 "Related objects must be referenced by numeric ID or by dictionary of attributes. Received an "
308 "unrecognized value: {value}"
309 ).format(value=attrs)
310 )
312 # Look up object by PK
313 try:
314 return queryset.get(pk=pk)
315 except ObjectDoesNotExist:
316 raise ValidationError(_("Related object not found using the provided numeric ID: {id}").format(id=pk))