Skip to content

nautobot.apps.api

Helpers for an app to implement a REST API.

nautobot.apps.api.APIRootView

Bases: NautobotAPIVersionMixin, APIView

This is the root of the REST API. API endpoints are arranged by app and model name; e.g. /api/dcim/locations/.

Source code in nautobot/core/api/views.py
class APIRootView(NautobotAPIVersionMixin, APIView):
    """
    This is the root of the REST API. API endpoints are arranged by app and model name; e.g. `/api/dcim/locations/`.
    """

    _ignore_model_permissions = True

    def get_view_name(self):
        return "API Root"

    @extend_schema(exclude=True)
    def get(self, request, format=None):  # pylint: disable=redefined-builtin
        return Response(
            OrderedDict(
                (
                    (
                        "circuits",
                        reverse("circuits-api:api-root", request=request, format=format),
                    ),
                    (
                        "dcim",
                        reverse("dcim-api:api-root", request=request, format=format),
                    ),
                    (
                        "extras",
                        reverse("extras-api:api-root", request=request, format=format),
                    ),
                    ("graphql", reverse("graphql-api", request=request, format=format)),
                    (
                        "ipam",
                        reverse("ipam-api:api-root", request=request, format=format),
                    ),
                    (
                        "plugins",
                        reverse("plugins-api:api-root", request=request, format=format),
                    ),
                    ("status", reverse("api-status", request=request, format=format)),
                    (
                        "tenancy",
                        reverse("tenancy-api:api-root", request=request, format=format),
                    ),
                    (
                        "users",
                        reverse("users-api:api-root", request=request, format=format),
                    ),
                    (
                        "virtualization",
                        reverse(
                            "virtualization-api:api-root",
                            request=request,
                            format=format,
                        ),
                    ),
                )
            )
        )

nautobot.apps.api.BaseModelSerializer

Bases: OptInFieldsMixin, serializers.HyperlinkedModelSerializer

This base serializer implements common fields and logic for all ModelSerializers.

Namely, it:

  • defines the display field which exposes a human friendly value for the given object.
  • ensures that id field is always present on the serializer as well.
  • ensures that created and last_updated fields are always present if applicable to this model and serializer.
  • ensures that object_type field is always present on the serializer which represents the content-type of this serializer's associated model (e.g. "dcim.device"). This is required as the OpenAPI schema, using the PolymorphicProxySerializer class defined below, relies upon this field as a way to identify to the client which of several possible serializers are in use for a given attribute.
  • supports ?depth query parameter. It is passed in as nested_depth to the build_nested_field() function to enable the dynamic generation of nested serializers.
Source code in nautobot/core/api/serializers.py
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
class BaseModelSerializer(OptInFieldsMixin, serializers.HyperlinkedModelSerializer):
    """
    This base serializer implements common fields and logic for all ModelSerializers.

    Namely, it:

    - defines the `display` field which exposes a human friendly value for the given object.
    - ensures that `id` field is always present on the serializer as well.
    - ensures that `created` and `last_updated` fields are always present if applicable to this model and serializer.
    - ensures that `object_type` field is always present on the serializer which represents the content-type of this
      serializer's associated model (e.g. "dcim.device"). This is required as the OpenAPI schema, using the
      PolymorphicProxySerializer class defined below, relies upon this field as a way to identify to the client
      which of several possible serializers are in use for a given attribute.
    - supports `?depth` query parameter. It is passed in as `nested_depth` to the `build_nested_field()` function
      to enable the dynamic generation of nested serializers.
    """

    serializer_related_field = NautobotHyperlinkedRelatedField

    display = serializers.SerializerMethodField(read_only=True, help_text="Human friendly display value")
    object_type = ObjectTypeField()
    # composite_key = serializers.SerializerMethodField()  # TODO: Revisit if we reintroduce composite keys
    natural_keys_values = None
    natural_slug = serializers.SerializerMethodField()

    def __init__(self, *args, force_csv=False, **kwargs):
        """
        Instantiate a BaseModelSerializer.

        The force_csv kwarg allows you to force _is_csv_request() to evaluate True without passing a Request object,
        which is necessary to be able to export appropriately structured CSV from a Job that doesn't have a Request.
        """
        self._force_csv = force_csv

        super().__init__(*args, **kwargs)
        # If it is not a Nested Serializer, we should set the depth argument to whatever is in the request's context
        if not self.is_nested:
            self.Meta.depth = self.context.get("depth", 0)

        # Check if the request is related to CSV export;
        if self._is_csv_request() and self.instance:
            # Retrieve the natural key values of related fields in an optimized way.
            all_related_fields_natural_key_lookups = self._get_related_fields_natural_key_field_lookups()
            case_query = self._build_query_case_for_natural_key_field_lookup(all_related_fields_natural_key_lookups)
            if isinstance(self.instance, models.QuerySet):
                queryset = self.instance
            else:
                # We would only need to run one additional query, making this a more efficient method of
                # obtaining all the natural key values for this instance;
                queryset = self.Meta.model.objects.filter(pk=self.instance.pk)
            self.natural_keys_values = queryset.annotate(**case_query).values(
                *all_related_fields_natural_key_lookups, "pk"
            )

    def _get_lookup_field_name_and_output_field(self, lookup_field):
        """Get lookup field name and its corresponding output_field.

        Used in building this lookup Case in `_build_query_case_for_natural_key_field_lookup`.

        Example:
            >>> self._get_lookup_field_name_and_output_field("device__location__name")
            ("device__location", CharField)
            >>> self._get_lookup_field_name_and_output_field("ipaddress__parent__network")
            ("ipaddress__parent", VarbinaryIPField)
        """
        *field_names, lookup = lookup_field.split("__")
        model = self.Meta.model
        for field_component in field_names:
            model = model._meta.get_field(field_component).remote_field.model

        lookup = "id" if lookup == "pk" else lookup
        field = model._meta.get_field(lookup)
        # VarbinaryIPField needs to be handled specially in `_build_query_case_for_natural_key_field_lookup`
        output_field = field.__class__ if field.__class__ is VarbinaryIPField else models.CharField

        field_name = "__".join(field_names)
        return field_name, output_field

    def _build_query_case_for_natural_key_field_lookup(self, lookups):
        """
        Build a query using Case expressions to handle natural key field instances that do not exist.

        This function constructs a database query with Case expressions to handle natural key lookup fields
        that may have missing instances. In cases where the natural key field instance does not exist
        (i.e., is None), this function replaces it with the value 'NoObject'. This is particularly
        useful for CSV Export processes, as it allows fields with missing instances to be safely ignored.
        Such handling is essential for CSV Import processes, as attempting to import missing instances can
        lead to 'Object Not Found' errors, potentially causing the import to fail.

        Example:
            Consider a Device model with a related field 'tenant' and a 'name' attribute. In this case, the
            function can be used as follows:

            Device.objects.annotate(
                tenant__name=Case(
                    When(tenant__isnull=False, then=F("tenant__name")),
                    default="NoObject"
                )
            ).values("tenant__name")

        Explanation:
            - If `device.tenant` is None, the 'tenant__name' field is set to "NoObject".
            - If `device.tenant` is not None, the 'tenant__name' field is set to the actual tenant name.

        Args:
            lookups: List of natural key lookups
        """
        case_query = {}
        for lookup_field in lookups:
            field_name, output_field = self._get_lookup_field_name_and_output_field(lookup_field)

            # Since VarbinaryIPField cant be cast into CharField we would have to set the output_field as
            # `VarbinaryIPField`
            when_case = {
                f"{field_name}__isnull": False,
                "then": models.F(lookup_field)
                if output_field == VarbinaryIPField
                else Cast(models.F(lookup_field), models.CharField()),
            }
            case_query[lookup_field] = models.Case(
                models.When(**when_case),
                default=models.Value(constants.CSV_NO_OBJECT),
                output_field=output_field(),
            )
        return case_query

    def _get_related_fields_natural_key_field_lookups(self):
        """Retrieve a list of field lookups for natural key fields of related models.

        This method iterates through the related fields of the Serializer model,
        retrieves the natural_key_field_lookups for each related model, and prepends the field name
        to create a list of field lookups.

        Examples:
            >>> # Example usage on Device
            >>> self._get_related_fields_natural_key_field_lookups()
            [
                "tenant__name",
                "status__name",
                "role__name",
                "location__name",
                "location__parent__name",
                "location__parent__parent__name"
                ...
            ]

        """
        model = self.Meta.model
        field_lookups = []
        # NOTE: M2M and One2M fields field are ignored in csv export
        fields = [
            field
            for field in model._meta.get_fields()
            if field.is_relation
            and not field.many_to_many
            and not field.one_to_many
            # Ignore GenericRel since its `fk` and `content_type` would be used.
            and not isinstance(field, GenericRel)
        ]
        # Get each related field model's natural_key_fields and prepend field name
        for field in fields:
            # ContentType and Group are not Nautobot Model hence do not have the `natural_key_field_lookups` attr.
            # fallback to using default behavior for these fields
            with contextlib.suppress(AttributeError):
                field_lookups.extend(
                    f"{field.name}__{lookup}" for lookup in field.related_model.csv_natural_key_field_lookups()
                )
        return field_lookups

    def _is_csv_request(self):
        """Return True if this a CSV export request"""
        if self._force_csv:
            return True
        request = self.context.get("request")
        return hasattr(request, "accepted_media_type") and "text/csv" in request.accepted_media_type

    @property
    def is_nested(self):
        """Return whether this is a nested serializer."""
        return getattr(self.Meta, "is_nested", False)

    @property
    def list_display_fields(self):
        return list(getattr(self.Meta, "list_display_fields", []))

    @property
    def advanced_tab_fields(self):
        advanced_fields = list(
            getattr(
                self.Meta,
                "advanced_tab_fields",
                ["id", "url", "object_type", "created", "last_updated", "natural_slug"],
            )
        )
        return [field for field in advanced_fields if field in self.fields]

    @extend_schema_field(serializers.CharField)
    def get_display(self, instance):
        """
        Return either the `display` property of the instance or `str(instance)`
        """
        return getattr(instance, "display", str(instance))

    # TODO(jathan): Rip out composite key after natural key fields for import/export work has been
    # completed (See: https://github.com/nautobot/nautobot/issues/4367)
    @extend_schema_field(
        {
            "type": "string",
            "example": constants.COMPOSITE_KEY_SEPARATOR.join(["attribute1", "attribute2"]),
        }
    )
    def get_composite_key(self, instance):
        try:
            return getattr(instance, "composite_key", construct_composite_key(instance.natural_key()))
        except (AttributeError, NotImplementedError):
            return "unknown"

    @extend_schema_field(
        {
            "type": "string",
            "example": constants.NATURAL_SLUG_SEPARATOR.join(["attribute1", "attribute2"]),
        }
    )
    def get_natural_slug(self, instance):
        try:
            return getattr(instance, "natural_slug", construct_natural_slug(instance.natural_key(), pk=instance.pk))
        except (AttributeError, NotImplementedError):
            return "unknown"

    def extend_field_names(self, fields, field_name, at_start=False, opt_in_only=False):
        """Prepend or append the given field_name to `fields` and optionally self.Meta.opt_in_fields as well."""
        if field_name in fields:
            fields.remove(field_name)
        if at_start:
            fields.insert(0, field_name)
        else:
            fields.append(field_name)
        if opt_in_only:
            if not getattr(self.Meta, "opt_in_fields", None):
                self.Meta.opt_in_fields = [field_name]
            elif field_name not in self.Meta.opt_in_fields:
                self.Meta.opt_in_fields.append(field_name)
        return fields

    def get_field_names(self, declared_fields, info):
        """
        Override get_field_names() to add some custom logic.

        Assuming that we follow the pattern where `fields = "__all__" for the vast majority of serializers in Nautobot,
        we do not strictly need to use this method to protect against inadvertently omitting standard fields
        like `display`, `created`, and `last_updated`. However, we continue to do as a bit of redundant safety.

        The other purpose of this method now is to manipulate the set of fields that "__all__" actually means as a
        way of *excluding* fields that we *don't* want to include by default for performance or data exposure reasons.
        """
        fields = list(super().get_field_names(declared_fields, info))  # Meta.fields could be defined as a tuple

        # Add initial fields in "reverse" order since they're each inserted at the start of the list.
        self.extend_field_names(fields, "display", at_start=True)
        self.extend_field_names(fields, "object_type", at_start=True)
        # Since we use HyperlinkedModelSerializer as our base class, "url" is auto-included by "__all__" but "id" isn't.
        self.extend_field_names(fields, "id", at_start=True)

        # Move these fields to the end
        fields_to_include = ["created", "last_updated"]
        for field in fields_to_include:
            if hasattr(self.Meta.model, field):
                self.extend_field_names(fields, field)

        def filter_field(field):
            # Eliminate all field names that start with "_" as those fields are not user-facing
            if field.startswith("_"):
                return False
            # These are expensive to look up, so we have decided not to include them on nested serializers
            if self.is_nested and isinstance(getattr(self.Meta.model, field, None), ManyToManyDescriptor):
                return False

            # Ignore M2M fields
            with contextlib.suppress(FieldDoesNotExist):
                if self._is_csv_request():
                    field = self.Meta.model._meta.get_field(field)
                    # ContentType is ManyToMany Field that is specially handled, Hence it can be exported/imported
                    if field.many_to_many and field.related_model is not ContentType:
                        return False
            return True

        fields = [field for field in fields if filter_field(field)]
        return fields

    def determine_view_options(self, request=None):
        """
        Determine view options to use for rendering the list and detail views associated with this serializer.
        """
        list_display = []
        fields = []

        from nautobot.core.api.metadata import NautobotColumnProcessor  # avoid circular import

        processor = NautobotColumnProcessor(self, request.parser_context if request else {})
        field_map = dict(self.fields)
        all_fields = list(field_map)

        # Explicitly order the "big ugly" fields to the bottom
        processor.order_fields(all_fields)
        list_display_fields = self.list_display_fields

        # Process the list_display_fields first.
        for field_name in list_display_fields:
            try:
                field = field_map[field_name]
            except KeyError:
                continue  # Ignore unknown fields.
            column_data = processor._get_column_properties(field, field_name)
            list_display.append(column_data)
            fields.append(column_data)

        # Process the rest of the fields second.
        for field_name in all_fields:
            # Don't process list display fields twice.
            if field_name in list_display_fields:
                continue
            try:
                field = field_map[field_name]
            except KeyError:
                continue  # Ignore unknown fields.
            column_data = processor._get_column_properties(field, field_name)
            fields.append(column_data)

        return {
            "retrieve": {
                "tabs": self._determine_detail_view_tabs(),
            },
            "list": {
                "default_fields": list_display,
                "all_fields": fields,
            },
        }

    def _determine_detail_view_tabs(self):
        """Determine the layout for the detail view tabs that are intrinsic to this serializer."""
        tabs = self.get_additional_detail_view_tabs()

        if hasattr(self.Meta, "detail_view_config"):
            detail_view_config = self._validate_view_config(self.Meta.detail_view_config)
        else:
            detail_view_config = self._get_default_detail_view_config()
        detail_view_config = self._refine_detail_view_config(detail_view_config, tabs)

        return {
            bettertitle(self.Meta.model._meta.verbose_name): detail_view_config,
            **tabs,
        }

    def get_additional_detail_view_tabs(self):
        """
        Retrieve definitions of non-default detail view tabs.

        By default provides an "Advanced" tab containing `self.advanced_tab_fields`, but subclasses
        can override this to move additional serializer fields to this or other tabs.

        Returns:
            (dict): `{<tab label>: [{<panel label>: {"fields": [<list of fields>]}, ...}, ...], ...}`
        """
        return {
            "Advanced": [{"Object Details": {"fields": self.advanced_tab_fields}}],
        }

    def _get_default_detail_view_config(self):
        """
        Generate detail view config for the view based on the serializer's fields.

        Examples:
            >>> DeviceSerializer._get_default_detail_view_config().
            {
                "layout":[
                    {
                        Device: {
                            "fields": ["name", "subdevice_role", "height", "comments"...]
                        }
                    },
                    {
                        Tags: {
                            "fields": ["tags"]
                        }
                    }
                ]
            }

        Returns:
            (list): A list representing the view config.
        """
        m2m_fields, other_fields = self._get_m2m_and_non_m2m_fields()
        # TODO(timizuo): How do we get verbose_name of not model serializers?
        model_verbose_name = self.Meta.model._meta.verbose_name
        return {
            "layout": [
                {
                    bettertitle(model_verbose_name): {
                        "fields": [field["name"] for field in other_fields],
                    }
                },
                {field["label"]: {"fields": [field["name"]]} for field in m2m_fields},
            ]
        }

    def _get_m2m_and_non_m2m_fields(self):
        """
        Retrieve the many-to-many (m2m) fields and other non-m2m fields from the serializer.

        Returns:
            A tuple containing two lists: m2m_fields and non m2m fields.
                - m2m_fields: A list of dictionaries, each containing the name and label of an m2m field.
                - non_m2m_fields: A list of dictionaries, each containing the name and label of a non m2m field.
        """
        m2m_fields = []
        non_m2m_fields = []

        for field_name, field in self.fields.items():
            if isinstance(field, drf_relations.ManyRelatedField):
                m2m_fields.append({"name": field_name, "label": field.label or field_name})
            else:
                non_m2m_fields.append({"name": field_name, "label": field.label or field_name})

        return m2m_fields, non_m2m_fields

    def _refine_detail_view_config(self, detail_view_config, other_tabs):
        """
        Refine the detail view config for the default tab (auto-generated, or as defined by Meta.detail_view_config).

        - Remove fields that should never be present in the detail view config (e.g. `notes_url`).
        - Ensure that fields that are already present in `other_tabs` aren't included in the detail view config.
        - Ensure that certain fields such as `tags` are always included in the detail view config if applicable.

        Args:
            detail_view_config (dict): `{"layout": [{<left-column>}, {<right-column}], "include_others": False}`

        Returns:
            (list): `[{"Panel 1 Name": {"fields": ["field1", "field2", ...]}, "Panel 2 Name": ...}, {...}]`
        """
        fields_to_always_move_to_right_column = [
            "comments",
            "tags",
        ]
        fields_to_always_remove = [
            # always handled explicitly by the UI
            "display",
            "status",
            # not yet supported in the UI
            "custom_fields",
            "relationships",
            "computed_fields",
            # irrelevant to the UI
            "notes_url",
        ]
        fields_to_remove = fields_to_always_remove + fields_to_always_move_to_right_column
        # Any field that's already present in another tab
        for tab_layout in other_tabs.values():
            for column in tab_layout:
                for grouping in column.values():
                    fields_to_remove += grouping["fields"]

        # Make a deepcopy to avoid altering view_config
        view_config_layout = deepcopy(detail_view_config.get("layout"))

        # Remove fields_to_remove from the view_config_layout
        for column in view_config_layout:
            for section in column.values():
                for field in fields_to_remove:
                    if field in section["fields"]:
                        section["fields"].remove(field)

        serializer_fields = list(self.fields)

        # Add special-cased fields to right column
        for field in fields_to_always_move_to_right_column:
            if field in serializer_fields:
                if len(view_config_layout) < 2:
                    view_config_layout.append({})
                view_config_layout[1].setdefault(bettertitle(field), {}).setdefault("fields", []).append(field)

        # Add fields not otherwise included in another tab, only if include_others is set to True.
        if detail_view_config.get("include_others", False):
            view_config_fields = [
                field for column in view_config_layout for section in column.values() for field in section["fields"]
            ]
            missing_fields = sorted(set(serializer_fields) - set(view_config_fields) - set(fields_to_remove))
            view_config_layout[0]["Other Fields"] = {"fields": missing_fields}

        return view_config_layout

    def _validate_view_config(self, view_config):
        """Validate view config."""
        # 1. Validate key `layout` is in view_config; as this is a required key is creating a view config
        if not view_config["layout"]:
            raise ViewConfigException("`layout` is a required key in creating a custom view_config")

        # 2. Validate `Other Fields` is not part of a layout group name, as this is a reserved keyword for group names
        for col in view_config["layout"]:
            for group_name in col.keys():
                if group_name in constants.RESERVED_NAMES_FOR_OBJECT_DETAIL_VIEW_SCHEMA:
                    raise ViewConfigException(f"`{group_name}` is a reserved group name keyword.")

        return view_config

    def build_field(self, field_name, info, model_class, nested_depth):
        """
        Return a two tuple of (cls, kwargs) to build a serializer field with.
        """
        request = self.context.get("request")
        # Make sure that PATCH/POST/PUT method response serializers are consistent
        # with depth of 0
        if request is not None and request.method != "GET":
            nested_depth = 0
        # For tags field, DRF does not recognize the relationship between tags and the model itself (?)
        # so instead of calling build_nested_field() it will call build_property_field() which
        # makes the field impervious to the `?depth` parameter.
        # So we intercept it here to call build_nested_field()
        # which will make the tags field be rendered with TagSerializer() and respect the `depth` parameter.
        if isinstance(getattr(model_class, field_name, None), TagsManager) and nested_depth > 0:
            tags_field = getattr(model_class, field_name)
            relation_info = RelationInfo(
                model_field=tags_field,
                related_model=Tag,
                to_many=True,
                has_through_model=True,
                to_field=_get_to_field(tags_field),
                reverse=False,
            )
            return self.build_nested_field(field_name, relation_info, nested_depth)

        return super().build_field(field_name, info, model_class, nested_depth)

    def _get_natural_key_lookups_value_for_field(self, field_name, natural_key_field_instance):
        """Extract natural key field lookups for a specific field name.

        Args:
            field_name (str): The field name to extract lookups for.
            natural_key_field_instance (dict): The dict containing natural key field values.

        Example:
            >>> natural_key_field_instance = Device.objects.values("tenant__name", "location__name", "location__parent__name", ...)
            >>> _get_natural_key_lookups_value_for_field("location", natural_key_field_instance)
            {
                "location__name": "Sample Location",
                "location__parent__name": "Sample Location Parent Name",
                "location__parent__parent__name": "NoObject"
                ...
            }

        """
        data = {}
        for key, value in natural_key_field_instance.items():
            if key.startswith(f"{field_name}__"):
                if isinstance(value, uuid.UUID):
                    data[key] = str(value)
                elif value == constants.VARBINARY_IP_FIELD_REPR_OF_CSV_NO_OBJECT:
                    data[key] = constants.CSV_NO_OBJECT
                elif not value:
                    data[key] = constants.CSV_NULL_TYPE
                else:
                    data[key] = value
        return data

    def to_representation(self, instance):
        data = super().to_representation(instance)
        altered_data = {}

        if self._is_csv_request() and self.natural_keys_values is not None:
            if natural_key_field_instance := [item for item in self.natural_keys_values if item["pk"] == instance.pk]:
                cleaned_natural_key_field_instance = natural_key_field_instance[0]
                for key, value in data.items():
                    # FK field with natural_field_lookups
                    if natural_key_field_lookups_for_field := self._get_natural_key_lookups_value_for_field(
                        key, cleaned_natural_key_field_instance
                    ):
                        altered_data.update(natural_key_field_lookups_for_field)
                    else:
                        # Not FK field
                        altered_data[key] = constants.CSV_NULL_TYPE if value is None else value
        else:
            altered_data = data

        return altered_data

    def build_relational_field(self, field_name, relation_info):
        """Override DRF's default relational-field construction to be app-aware."""
        field_class, field_kwargs = super().build_relational_field(field_name, relation_info)
        if "view_name" in field_kwargs:
            field_kwargs["view_name"] = get_route_for_model(relation_info.related_model, "detail", api=True)
        return field_class, field_kwargs

    def build_property_field(self, field_name, model_class):
        """
        Create a property field for model methods and properties.
        """
        if isinstance(getattr(model_class, field_name, None), TagsManager):
            field_class = NautobotHyperlinkedRelatedField
            field_kwargs = {
                "queryset": Tag.objects.get_for_model(model_class),
                "many": True,
                "required": False,
            }

            return field_class, field_kwargs
        return super().build_property_field(field_name, model_class)

    def build_nested_field(self, field_name, relation_info, nested_depth):
        return nested_serializer_factory(relation_info, nested_depth)

    def build_url_field(self, field_name, model_class):
        """Override DRF's default 'url' field construction to be app-aware."""
        field_class, field_kwargs = super().build_url_field(field_name, model_class)
        if "view_name" in field_kwargs:
            field_kwargs["view_name"] = get_route_for_model(model_class, "detail", api=True)
        return field_class, field_kwargs

is_nested property

Return whether this is a nested serializer.

__init__(*args, force_csv=False, **kwargs)

Instantiate a BaseModelSerializer.

The force_csv kwarg allows you to force _is_csv_request() to evaluate True without passing a Request object, which is necessary to be able to export appropriately structured CSV from a Job that doesn't have a Request.

Source code in nautobot/core/api/serializers.py
def __init__(self, *args, force_csv=False, **kwargs):
    """
    Instantiate a BaseModelSerializer.

    The force_csv kwarg allows you to force _is_csv_request() to evaluate True without passing a Request object,
    which is necessary to be able to export appropriately structured CSV from a Job that doesn't have a Request.
    """
    self._force_csv = force_csv

    super().__init__(*args, **kwargs)
    # If it is not a Nested Serializer, we should set the depth argument to whatever is in the request's context
    if not self.is_nested:
        self.Meta.depth = self.context.get("depth", 0)

    # Check if the request is related to CSV export;
    if self._is_csv_request() and self.instance:
        # Retrieve the natural key values of related fields in an optimized way.
        all_related_fields_natural_key_lookups = self._get_related_fields_natural_key_field_lookups()
        case_query = self._build_query_case_for_natural_key_field_lookup(all_related_fields_natural_key_lookups)
        if isinstance(self.instance, models.QuerySet):
            queryset = self.instance
        else:
            # We would only need to run one additional query, making this a more efficient method of
            # obtaining all the natural key values for this instance;
            queryset = self.Meta.model.objects.filter(pk=self.instance.pk)
        self.natural_keys_values = queryset.annotate(**case_query).values(
            *all_related_fields_natural_key_lookups, "pk"
        )

build_field(field_name, info, model_class, nested_depth)

Return a two tuple of (cls, kwargs) to build a serializer field with.

Source code in nautobot/core/api/serializers.py
def build_field(self, field_name, info, model_class, nested_depth):
    """
    Return a two tuple of (cls, kwargs) to build a serializer field with.
    """
    request = self.context.get("request")
    # Make sure that PATCH/POST/PUT method response serializers are consistent
    # with depth of 0
    if request is not None and request.method != "GET":
        nested_depth = 0
    # For tags field, DRF does not recognize the relationship between tags and the model itself (?)
    # so instead of calling build_nested_field() it will call build_property_field() which
    # makes the field impervious to the `?depth` parameter.
    # So we intercept it here to call build_nested_field()
    # which will make the tags field be rendered with TagSerializer() and respect the `depth` parameter.
    if isinstance(getattr(model_class, field_name, None), TagsManager) and nested_depth > 0:
        tags_field = getattr(model_class, field_name)
        relation_info = RelationInfo(
            model_field=tags_field,
            related_model=Tag,
            to_many=True,
            has_through_model=True,
            to_field=_get_to_field(tags_field),
            reverse=False,
        )
        return self.build_nested_field(field_name, relation_info, nested_depth)

    return super().build_field(field_name, info, model_class, nested_depth)

build_property_field(field_name, model_class)

Create a property field for model methods and properties.

Source code in nautobot/core/api/serializers.py
def build_property_field(self, field_name, model_class):
    """
    Create a property field for model methods and properties.
    """
    if isinstance(getattr(model_class, field_name, None), TagsManager):
        field_class = NautobotHyperlinkedRelatedField
        field_kwargs = {
            "queryset": Tag.objects.get_for_model(model_class),
            "many": True,
            "required": False,
        }

        return field_class, field_kwargs
    return super().build_property_field(field_name, model_class)

build_relational_field(field_name, relation_info)

Override DRF's default relational-field construction to be app-aware.

Source code in nautobot/core/api/serializers.py
def build_relational_field(self, field_name, relation_info):
    """Override DRF's default relational-field construction to be app-aware."""
    field_class, field_kwargs = super().build_relational_field(field_name, relation_info)
    if "view_name" in field_kwargs:
        field_kwargs["view_name"] = get_route_for_model(relation_info.related_model, "detail", api=True)
    return field_class, field_kwargs

build_url_field(field_name, model_class)

Override DRF's default 'url' field construction to be app-aware.

Source code in nautobot/core/api/serializers.py
def build_url_field(self, field_name, model_class):
    """Override DRF's default 'url' field construction to be app-aware."""
    field_class, field_kwargs = super().build_url_field(field_name, model_class)
    if "view_name" in field_kwargs:
        field_kwargs["view_name"] = get_route_for_model(model_class, "detail", api=True)
    return field_class, field_kwargs

determine_view_options(request=None)

Determine view options to use for rendering the list and detail views associated with this serializer.

Source code in nautobot/core/api/serializers.py
def determine_view_options(self, request=None):
    """
    Determine view options to use for rendering the list and detail views associated with this serializer.
    """
    list_display = []
    fields = []

    from nautobot.core.api.metadata import NautobotColumnProcessor  # avoid circular import

    processor = NautobotColumnProcessor(self, request.parser_context if request else {})
    field_map = dict(self.fields)
    all_fields = list(field_map)

    # Explicitly order the "big ugly" fields to the bottom
    processor.order_fields(all_fields)
    list_display_fields = self.list_display_fields

    # Process the list_display_fields first.
    for field_name in list_display_fields:
        try:
            field = field_map[field_name]
        except KeyError:
            continue  # Ignore unknown fields.
        column_data = processor._get_column_properties(field, field_name)
        list_display.append(column_data)
        fields.append(column_data)

    # Process the rest of the fields second.
    for field_name in all_fields:
        # Don't process list display fields twice.
        if field_name in list_display_fields:
            continue
        try:
            field = field_map[field_name]
        except KeyError:
            continue  # Ignore unknown fields.
        column_data = processor._get_column_properties(field, field_name)
        fields.append(column_data)

    return {
        "retrieve": {
            "tabs": self._determine_detail_view_tabs(),
        },
        "list": {
            "default_fields": list_display,
            "all_fields": fields,
        },
    }

extend_field_names(fields, field_name, at_start=False, opt_in_only=False)

Prepend or append the given field_name to fields and optionally self.Meta.opt_in_fields as well.

Source code in nautobot/core/api/serializers.py
def extend_field_names(self, fields, field_name, at_start=False, opt_in_only=False):
    """Prepend or append the given field_name to `fields` and optionally self.Meta.opt_in_fields as well."""
    if field_name in fields:
        fields.remove(field_name)
    if at_start:
        fields.insert(0, field_name)
    else:
        fields.append(field_name)
    if opt_in_only:
        if not getattr(self.Meta, "opt_in_fields", None):
            self.Meta.opt_in_fields = [field_name]
        elif field_name not in self.Meta.opt_in_fields:
            self.Meta.opt_in_fields.append(field_name)
    return fields

get_additional_detail_view_tabs()

Retrieve definitions of non-default detail view tabs.

By default provides an "Advanced" tab containing self.advanced_tab_fields, but subclasses can override this to move additional serializer fields to this or other tabs.

Returns:

Type Description
dict

{<tab label>: [{<panel label>: {"fields": [<list of fields>]}, ...}, ...], ...}

Source code in nautobot/core/api/serializers.py
def get_additional_detail_view_tabs(self):
    """
    Retrieve definitions of non-default detail view tabs.

    By default provides an "Advanced" tab containing `self.advanced_tab_fields`, but subclasses
    can override this to move additional serializer fields to this or other tabs.

    Returns:
        (dict): `{<tab label>: [{<panel label>: {"fields": [<list of fields>]}, ...}, ...], ...}`
    """
    return {
        "Advanced": [{"Object Details": {"fields": self.advanced_tab_fields}}],
    }

get_display(instance)

Return either the display property of the instance or str(instance)

Source code in nautobot/core/api/serializers.py
@extend_schema_field(serializers.CharField)
def get_display(self, instance):
    """
    Return either the `display` property of the instance or `str(instance)`
    """
    return getattr(instance, "display", str(instance))

get_field_names(declared_fields, info)

Override get_field_names() to add some custom logic.

Assuming that we follow the pattern where fields = "__all__" for the vast majority of serializers in Nautobot, we do not strictly need to use this method to protect against inadvertently omitting standard fields likedisplay,created, andlast_updated`. However, we continue to do as a bit of redundant safety.

The other purpose of this method now is to manipulate the set of fields that "all" actually means as a way of excluding fields that we don't want to include by default for performance or data exposure reasons.

Source code in nautobot/core/api/serializers.py
def get_field_names(self, declared_fields, info):
    """
    Override get_field_names() to add some custom logic.

    Assuming that we follow the pattern where `fields = "__all__" for the vast majority of serializers in Nautobot,
    we do not strictly need to use this method to protect against inadvertently omitting standard fields
    like `display`, `created`, and `last_updated`. However, we continue to do as a bit of redundant safety.

    The other purpose of this method now is to manipulate the set of fields that "__all__" actually means as a
    way of *excluding* fields that we *don't* want to include by default for performance or data exposure reasons.
    """
    fields = list(super().get_field_names(declared_fields, info))  # Meta.fields could be defined as a tuple

    # Add initial fields in "reverse" order since they're each inserted at the start of the list.
    self.extend_field_names(fields, "display", at_start=True)
    self.extend_field_names(fields, "object_type", at_start=True)
    # Since we use HyperlinkedModelSerializer as our base class, "url" is auto-included by "__all__" but "id" isn't.
    self.extend_field_names(fields, "id", at_start=True)

    # Move these fields to the end
    fields_to_include = ["created", "last_updated"]
    for field in fields_to_include:
        if hasattr(self.Meta.model, field):
            self.extend_field_names(fields, field)

    def filter_field(field):
        # Eliminate all field names that start with "_" as those fields are not user-facing
        if field.startswith("_"):
            return False
        # These are expensive to look up, so we have decided not to include them on nested serializers
        if self.is_nested and isinstance(getattr(self.Meta.model, field, None), ManyToManyDescriptor):
            return False

        # Ignore M2M fields
        with contextlib.suppress(FieldDoesNotExist):
            if self._is_csv_request():
                field = self.Meta.model._meta.get_field(field)
                # ContentType is ManyToMany Field that is specially handled, Hence it can be exported/imported
                if field.many_to_many and field.related_model is not ContentType:
                    return False
        return True

    fields = [field for field in fields if filter_field(field)]
    return fields

nautobot.apps.api.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 UUID of an object to be deleted. For example:

DELETE /api/dcim/locations/ [ {"id": "3f01f169-49b9-42d5-a526-df9118635d62"}, {"id": "c27d6c5b-7ea8-41e7-b9dd-c065efd5d9cd"} ]

Source code in nautobot/core/api/views.py
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 UUID of an object to be deleted. For example:

    DELETE /api/dcim/locations/
    [
        {"id": "3f01f169-49b9-42d5-a526-df9118635d62"},
        {"id": "c27d6c5b-7ea8-41e7-b9dd-c065efd5d9cd"}
    ]
    """

    bulk_operation_serializer_class = BulkOperationSerializer

    @extend_schema(
        request=BulkOperationSerializer(many=True),
    )
    def bulk_destroy(self, request, *args, **kwargs):
        serializer = self.bulk_operation_serializer_class(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:
                self.perform_destroy(obj)

nautobot.apps.api.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 UUID of an object to be updated as well as the attributes to be set. For example:

PATCH /api/dcim/locations/ [ { "id": "1f554d07-d099-437d-8d48-7d6e35ec8fa3", "name": "New name" }, { "id": "1f554d07-d099-437d-8d48-7d6e65ec8fa3", "status": "planned" } ]

Source code in nautobot/core/api/views.py
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 UUID of an object to be updated as well as the attributes to be set.
    For example:

    PATCH /api/dcim/locations/
    [
        {
            "id": "1f554d07-d099-437d-8d48-7d6e35ec8fa3",
            "name": "New name"
        },
        {
            "id": "1f554d07-d099-437d-8d48-7d6e65ec8fa3",
            "status": "planned"
        }
    ]
    """

    bulk_operation_serializer_class = BulkOperationSerializer

    def bulk_update(self, request, *args, **kwargs):
        partial = kwargs.pop("partial", False)
        serializer = self.bulk_operation_serializer_class(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)

        # 2.0 TODO: this should be wrapped with a paginator so as to match the same format as the list endpoint,
        # i.e. `{"results": [{instance}, {instance}, ...]}` instead of bare list `[{instance}, {instance}, ...]`
        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(str(obj.id))
                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)

nautobot.apps.api.ChoiceField

Bases: serializers.Field

Represent a ChoiceField as {'value': , 'label': }. Accepts a single value on write.

:param choices: An iterable of choices in the form (value, key). :param allow_blank: Allow blank values in addition to the listed choices.

Source code in nautobot/core/api/fields.py
class ChoiceField(serializers.Field):
    """
    Represent a ChoiceField as {'value': <DB value>, 'label': <string>}. Accepts a single value on write.

    :param choices: An iterable of choices in the form (value, key).
    :param allow_blank: Allow blank values in addition to the listed choices.
    """

    def __init__(self, choices, allow_blank=False, **kwargs):
        self.choiceset = choices
        self.allow_blank = allow_blank
        self._choices = {}

        # Unpack grouped choices
        for k, v in choices:
            if isinstance(v, (list, tuple)):
                for k2, v2 in v:
                    self._choices[k2] = v2
            else:
                self._choices[k] = v

        super().__init__(**kwargs)

    def validate_empty_values(self, data):
        # Convert null to an empty string unless allow_null == True
        if data is None:
            if self.allow_null:
                return True, None
            else:
                data = ""
        return super().validate_empty_values(data)

    def to_representation(self, obj):
        if obj == "":
            return None
        return OrderedDict([("value", obj), ("label", self._choices[obj])])

    def to_internal_value(self, data):
        if data == "":
            if self.allow_blank:
                return data
            raise ValidationError("This field may not be blank.")

        if isinstance(data, dict):
            if "value" in data:
                data = data["value"]
            else:
                raise ValidationError(
                    'Value must be passed directly (e.g. "foo": 123) '
                    'or as a dict with key "value" (e.g. "foo": {"value": 123}).'
                )

        # Provide an explicit error message if the request is trying to write a dict or list
        if isinstance(data, list):
            raise ValidationError('Value must be passed directly (e.g. "foo": 123); do not use a list.')

        # Check for string representations of boolean/integer values
        if hasattr(data, "lower"):
            if data.lower() == "true":
                data = True
            elif data.lower() == "false":
                data = False
            else:
                try:
                    data = int(data)
                except ValueError:
                    pass

        try:
            if data in self._choices:
                return data
        except TypeError:  # Input is an unhashable type
            pass

        raise ValidationError(f"{data} is not a valid choice.")

    @property
    def choices(self):
        return self._choices

nautobot.apps.api.ContentTypeField

Bases: RelatedField

Represent a ContentType as '.'

Source code in nautobot/core/api/fields.py
@extend_schema_field(str)
class ContentTypeField(RelatedField):
    """
    Represent a ContentType as '<app_label>.<model>'
    """

    default_error_messages = {
        "does_not_exist": "Invalid content type: {content_type}",
        "invalid": "Invalid value. Specify a content type as '<app_label>.<model_name>'.",
    }

    def to_internal_value(self, data):
        try:
            app_label, model = data.split(".")
            return self.queryset.get(app_label=app_label, model=model)
        except ObjectDoesNotExist:
            self.fail("does_not_exist", content_type=data)
        except (AttributeError, TypeError, ValueError):
            self.fail("invalid")
        return None

    def to_representation(self, obj):
        return f"{obj.app_label}.{obj.model}"

nautobot.apps.api.CustomFieldModelSerializerMixin

Bases: ValidatedModelSerializer

Extends ModelSerializer to render any CustomFields and their values associated with an object.

Source code in nautobot/core/api/serializers.py
class CustomFieldModelSerializerMixin(ValidatedModelSerializer):
    """
    Extends ModelSerializer to render any CustomFields and their values associated with an object.
    """

    computed_fields = SerializerMethodField(read_only=True)
    custom_fields = CustomFieldsDataField(
        source="_custom_field_data",
        default=CreateOnlyDefault(CustomFieldDefaultValues()),
    )

    @extend_schema_field(OpenApiTypes.OBJECT)
    def get_computed_fields(self, obj):
        return obj.get_computed_fields()

    def get_field_names(self, declared_fields, info):
        """Ensure that "custom_fields" and "computed_fields" are included appropriately."""
        fields = list(super().get_field_names(declared_fields, info))
        # Ensure that custom_fields field appears at the end, not the start, of the fields
        self.extend_field_names(fields, "custom_fields")
        if not self.is_nested:
            # Only include computed_fields as opt-in.
            self.extend_field_names(fields, "computed_fields", opt_in_only=True)
        else:
            # As computed fields are expensive, do not include them in nested serializers even if opted-in at the root
            if "computed_fields" in fields:
                fields.remove("computed_fields")
        return fields

get_field_names(declared_fields, info)

Ensure that "custom_fields" and "computed_fields" are included appropriately.

Source code in nautobot/core/api/serializers.py
def get_field_names(self, declared_fields, info):
    """Ensure that "custom_fields" and "computed_fields" are included appropriately."""
    fields = list(super().get_field_names(declared_fields, info))
    # Ensure that custom_fields field appears at the end, not the start, of the fields
    self.extend_field_names(fields, "custom_fields")
    if not self.is_nested:
        # Only include computed_fields as opt-in.
        self.extend_field_names(fields, "computed_fields", opt_in_only=True)
    else:
        # As computed fields are expensive, do not include them in nested serializers even if opted-in at the root
        if "computed_fields" in fields:
            fields.remove("computed_fields")
    return fields

nautobot.apps.api.CustomFieldModelViewSet

Bases: ModelViewSet

Include the applicable set of CustomFields in the ModelViewSet context.

Source code in nautobot/extras/api/views.py
class CustomFieldModelViewSet(ModelViewSet):
    """
    Include the applicable set of CustomFields in the ModelViewSet context.
    """

    def get_serializer_context(self):
        # Gather all custom fields for the model
        content_type = ContentType.objects.get_for_model(self.queryset.model)
        custom_fields = content_type.custom_fields.all()

        context = super().get_serializer_context()
        context.update(
            {
                "custom_fields": custom_fields,
            }
        )
        return context

nautobot.apps.api.GetObjectCountsView

Bases: NautobotAPIVersionMixin, APIView

Enumerate the models listed on the Nautobot home page and return data structure containing verbose_name_plural, url and count.

Source code in nautobot/core/api/views.py
class GetObjectCountsView(NautobotAPIVersionMixin, APIView):
    """
    Enumerate the models listed on the Nautobot home page and return data structure
    containing verbose_name_plural, url and count.
    """

    permission_classes = [IsAuthenticated]

    @extend_schema(exclude=True)
    def get(self, request):
        object_counts = {
            "Inventory": [
                {"model": "dcim.rack"},
                {"model": "dcim.devicetype"},
                {"model": "dcim.device"},
                {"model": "dcim.virtualchassis"},
                {"model": "dcim.deviceredundancygroup"},
                {"model": "dcim.cable"},
            ],
            "Networks": [
                {"model": "ipam.vrf"},
                {"model": "ipam.prefix"},
                {"model": "ipam.ipaddress"},
                {"model": "ipam.vlan"},
            ],
            "Security": [{"model": "extras.secret"}],
            "Platform": [
                {"model": "extras.gitrepository"},
                {"model": "extras.relationship"},
                {"model": "extras.computedfield"},
                {"model": "extras.customfield"},
                {"model": "extras.customlink"},
                {"model": "extras.tag"},
                {"model": "extras.status"},
                {"model": "extras.role"},
            ],
        }

        for entry in itertools.chain(*object_counts.values()):
            app_label, model_name = entry["model"].split(".")
            model = apps.get_model(app_label, model_name)
            permission = get_permission_for_model(model, "view")
            if not request.user.has_perm(permission):
                continue
            data = {"name": model._meta.verbose_name_plural}
            try:
                data["url"] = django_reverse(get_route_for_model(model, "list"))
            except NoReverseMatch:
                logger = logging.getLogger(__name__)
                route = get_route_for_model(model, "list")
                logger.warning(f"Handled expected exception when generating filter field: {route}")
            manager = model.objects
            if request.user.has_perm(permission):
                if hasattr(manager, "restrict"):
                    data["count"] = model.objects.restrict(request.user).count()
                else:
                    data["count"] = model.objects.count()
            entry.update(data)

        return Response(object_counts)

nautobot.apps.api.ModelViewSet

Bases: NautobotAPIVersionMixin, BulkUpdateModelMixin, BulkDestroyModelMixin, ModelViewSetMixin, ModelViewSet_

Extend DRF's ModelViewSet to support bulk update and delete functions.

Source code in nautobot/core/api/views.py
class ModelViewSet(
    NautobotAPIVersionMixin,
    BulkUpdateModelMixin,
    BulkDestroyModelMixin,
    ModelViewSetMixin,
    ModelViewSet_,
):
    """
    Extend DRF's ModelViewSet to support bulk update and delete functions.
    """

    logger = logging.getLogger(__name__ + ".ModelViewSet")

    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 isinstance(instance, 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
        else:
            # Check that the instance is matched by the view's queryset
            self.queryset.get(pk=instance.pk)

    def perform_create(self, serializer):
        model = self.queryset.model
        self.logger.info(f"Creating new {model._meta.verbose_name}")

        # Enforce object-level permissions on save()
        try:
            with transaction.atomic():
                instance = serializer.save()
                self._validate_objects(instance)
        except ObjectDoesNotExist:
            raise PermissionDenied()

    def perform_update(self, serializer):
        model = self.queryset.model
        self.logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})")

        # Enforce object-level permissions on save()
        try:
            with transaction.atomic():
                instance = serializer.save()
                self._validate_objects(instance)
        except ObjectDoesNotExist:
            raise PermissionDenied()

    def perform_destroy(self, instance):
        model = self.queryset.model
        self.logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})")

        return super().perform_destroy(instance)

nautobot.apps.api.ModelViewSetMixin

Source code in nautobot/core/api/views.py
class ModelViewSetMixin:
    logger = logging.getLogger(__name__ + ".ModelViewSet")

    # TODO: can't set lookup_value_regex globally; some models/viewsets (ContentType, Group) have integer rather than
    #       UUID PKs and also do NOT support composite-keys.
    #       The impact of NOT setting this is that per the OpenAPI schema, only UUIDs are permitted for most ViewSets;
    #       however, "secretly" due to our custom get_object() implementation below, you can actually also specify a
    #       composite_key value instead of a UUID. We're not currently documenting/using this feature, so OK for now
    # lookup_value_regex = r"[^/]+"

    def get_object(self):
        """Extend rest_framework.generics.GenericAPIView.get_object to allow "pk" lookups to use a composite-key."""
        queryset = self.filter_queryset(self.get_queryset())
        lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field

        if lookup_url_kwarg not in self.kwargs:
            raise RuntimeError(
                f"Expected view {self.__class__.__name__} to be called with a URL keyword argument named "
                f'"{lookup_url_kwarg}". Fix your URL conf, or set the `.lookup_field` attribute on the view correctly.'
            )

        if lookup_url_kwarg == "pk" and hasattr(queryset.model, "composite_key"):
            # Support lookup by either PK (UUID) or composite_key
            lookup_value = self.kwargs["pk"]
            if is_uuid(lookup_value):
                obj = get_object_or_404(queryset, pk=lookup_value)
            else:
                obj = get_object_or_404(queryset, composite_key=lookup_value)
        else:
            # Default DRF lookup behavior, just in case a viewset has overridden `lookup_url_kwarg` for its own needs
            obj = get_object_or_404(queryset, **{self.lookup_field: self.kwargs[lookup_url_kwarg]})

        self.check_object_permissions(self.request, obj)

        return obj

    def get_serializer(self, *args, **kwargs):
        # If a list of objects has been provided, initialize the serializer with many=True
        if isinstance(kwargs.get("data", {}), list):
            kwargs["many"] = True

        return super().get_serializer(*args, **kwargs)

    def get_serializer_context(self):
        context = super().get_serializer_context()
        if "text/csv" in self.request.accepted_media_type:
            # CSV rendering should always use depth 0
            context["depth"] = 0
        elif self.request.method == "GET":
            # Only allow the depth to be greater than 0 in GET requests
            depth = 0
            try:
                depth = int(self.request.query_params.get("depth", 0))
            except ValueError:
                self.logger.warning("The depth parameter must be an integer between 0 and 10")

            context["depth"] = depth
        else:
            # Use depth=0 in all write type requests.
            context["depth"] = 0

        return context

    def restrict_queryset(self, request, *args, **kwargs):
        """
        Restrict the view's queryset to allow only the permitted objects for the given request.

        Subclasses (such as nautobot.extras.api.views.JobModelViewSet) may wish to override this.

        Called by initial(), below.
        """
        # Restrict the view's QuerySet to allow only the permitted objects for the given user, if applicable
        if request.user.is_authenticated:
            http_action = HTTP_ACTIONS[request.method]
            if http_action:
                self.queryset = self.queryset.restrict(request.user, http_action)

    def initial(self, request, *args, **kwargs):
        """
        Runs anything that needs to occur prior to calling the method handler.

        Override of internal Django Rest Framework API.
        """
        super().initial(request, *args, **kwargs)

        # Django Rest Framework stores the raw API version string e.g. "1.2" as request.version.
        # For convenience we split it out into integer major/minor versions as well.
        major, minor = request.version.split(".")
        request.major_version = int(major)
        request.minor_version = int(minor)

        self.restrict_queryset(request, *args, **kwargs)

    def dispatch(self, request, *args, **kwargs):
        try:
            return super().dispatch(request, *args, **kwargs)
        except ProtectedError as e:
            protected_objects = list(e.protected_objects)
            msg = f"Unable to delete object. {len(protected_objects)} dependent objects were found: "
            msg += ", ".join([f"{obj} ({obj.pk})" for obj in protected_objects])
            self.logger.warning(msg)
            return self.finalize_response(request, Response({"detail": msg}, status=409), *args, **kwargs)

    def finalize_response(self, request, response, *args, **kwargs):
        # In the case of certain errors, we might not even get to the point of setting request.accepted_media_type
        if hasattr(request, "accepted_media_type") and "text/csv" in request.accepted_media_type:
            filename = f"{settings.BRANDING_PREPENDED_FILENAME}{self.queryset.model.__name__.lower()}_data.csv"
            response["Content-Disposition"] = f'attachment; filename="{filename}"'
        return super().finalize_response(request, response, *args, **kwargs)

get_object()

Extend rest_framework.generics.GenericAPIView.get_object to allow "pk" lookups to use a composite-key.

Source code in nautobot/core/api/views.py
def get_object(self):
    """Extend rest_framework.generics.GenericAPIView.get_object to allow "pk" lookups to use a composite-key."""
    queryset = self.filter_queryset(self.get_queryset())
    lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field

    if lookup_url_kwarg not in self.kwargs:
        raise RuntimeError(
            f"Expected view {self.__class__.__name__} to be called with a URL keyword argument named "
            f'"{lookup_url_kwarg}". Fix your URL conf, or set the `.lookup_field` attribute on the view correctly.'
        )

    if lookup_url_kwarg == "pk" and hasattr(queryset.model, "composite_key"):
        # Support lookup by either PK (UUID) or composite_key
        lookup_value = self.kwargs["pk"]
        if is_uuid(lookup_value):
            obj = get_object_or_404(queryset, pk=lookup_value)
        else:
            obj = get_object_or_404(queryset, composite_key=lookup_value)
    else:
        # Default DRF lookup behavior, just in case a viewset has overridden `lookup_url_kwarg` for its own needs
        obj = get_object_or_404(queryset, **{self.lookup_field: self.kwargs[lookup_url_kwarg]})

    self.check_object_permissions(self.request, obj)

    return obj

initial(request, *args, **kwargs)

Runs anything that needs to occur prior to calling the method handler.

Override of internal Django Rest Framework API.

Source code in nautobot/core/api/views.py
def initial(self, request, *args, **kwargs):
    """
    Runs anything that needs to occur prior to calling the method handler.

    Override of internal Django Rest Framework API.
    """
    super().initial(request, *args, **kwargs)

    # Django Rest Framework stores the raw API version string e.g. "1.2" as request.version.
    # For convenience we split it out into integer major/minor versions as well.
    major, minor = request.version.split(".")
    request.major_version = int(major)
    request.minor_version = int(minor)

    self.restrict_queryset(request, *args, **kwargs)

restrict_queryset(request, *args, **kwargs)

Restrict the view's queryset to allow only the permitted objects for the given request.

Subclasses (such as nautobot.extras.api.views.JobModelViewSet) may wish to override this.

Called by initial(), below.

Source code in nautobot/core/api/views.py
def restrict_queryset(self, request, *args, **kwargs):
    """
    Restrict the view's queryset to allow only the permitted objects for the given request.

    Subclasses (such as nautobot.extras.api.views.JobModelViewSet) may wish to override this.

    Called by initial(), below.
    """
    # Restrict the view's QuerySet to allow only the permitted objects for the given user, if applicable
    if request.user.is_authenticated:
        http_action = HTTP_ACTIONS[request.method]
        if http_action:
            self.queryset = self.queryset.restrict(request.user, http_action)

nautobot.apps.api.MultipleChoiceJSONField

Bases: serializers.MultipleChoiceField

A MultipleChoiceField that renders the received value as a JSON-compatible list rather than a set.

Source code in nautobot/extras/api/fields.py
class MultipleChoiceJSONField(serializers.MultipleChoiceField):
    """A MultipleChoiceField that renders the received value as a JSON-compatible list rather than a set."""

    def __init__(self, **kwargs):
        """Overload default choices handling to also accept a callable."""
        choices = kwargs.get("choices")
        if callable(choices):
            kwargs["choices"] = CallableChoiceIterator(choices)
        super().__init__(**kwargs)

    def to_internal_value(self, data):
        set_value = super().to_internal_value(data)
        return sorted(set_value)

__init__(**kwargs)

Overload default choices handling to also accept a callable.

Source code in nautobot/extras/api/fields.py
def __init__(self, **kwargs):
    """Overload default choices handling to also accept a callable."""
    choices = kwargs.get("choices")
    if callable(choices):
        kwargs["choices"] = CallableChoiceIterator(choices)
    super().__init__(**kwargs)

nautobot.apps.api.NautobotAutoSchema

Bases: AutoSchema

Nautobot-specific extensions to drf-spectacular's AutoSchema.

Source code in nautobot/core/api/schema.py
class NautobotAutoSchema(AutoSchema):
    """Nautobot-specific extensions to drf-spectacular's AutoSchema."""

    custom_actions = ["bulk_update", "bulk_partial_update", "bulk_destroy"]

    # Primarily, method_mapping is used to map HTTP method verbs to viewset method names,
    # which doesn't account for the fact that with our custom actions there are multiple viewset methods per verb,
    # hence why we have to override get_operation_id() below.
    # Secondarily, drf-spectacular uses method_mapping.values() to identify which methods are view methods,
    # so need to make sure these methods are represented as values in the mapping even if not under the actual verbs.
    method_mapping = AutoSchema.method_mapping.copy()
    method_mapping.update(
        {
            "_put": "bulk_update",
            "_patch": "bulk_partial_update",
            "_delete": "bulk_destroy",
        }
    )

    @property
    def is_bulk_action(self):
        """Custom property for convenience."""
        return hasattr(self.view, "action") and self.view.action in self.custom_actions

    @property
    def is_partial_action(self):
        """Custom property for convenience."""
        return hasattr(self.view, "action") and self.view.action in ["partial_update", "bulk_partial_update"]

    def _get_paginator(self):
        """Nautobot's custom bulk operations, even though they return a list of records, are NOT paginated."""
        if self.is_bulk_action:
            return None
        return super()._get_paginator()

    def get_description(self):
        """
        Get the appropriate description for a given API endpoint.

        By default, if a specific action doesn't have its own docstring, and neither does the view class,
        drf-spectacular will walk up the MRO of the view class until it finds a docstring, and use that.
        Most of our viewsets (for better or for worse) do not have docstrings, and so it'll find and use the generic
        docstring of the `NautobotModelViewSet` class, which isn't very useful to the end user. Instead of doing that,
        we only use the docstring of the view itself (ignoring its parent class docstrings), or if none exists, we
        make an attempt at rendering a basically accurate default description.
        """
        action_or_method = getattr(self.view, getattr(self.view, "action", self.method.lower()), None)
        action_doc = get_doc(action_or_method)
        if action_doc:
            return action_doc

        if self.view.__doc__:
            view_doc = get_doc(self.view.__class__)
            if view_doc:
                return view_doc

        # Fall back to a generic default description
        if hasattr(self.view, "queryset") and self.method.lower() in self.method_mapping:
            action = self.method_mapping[self.method.lower()].replace("_", " ").capitalize()
            model_name = self.view.queryset.model._meta.verbose_name
            if action == "Create":
                return f"{action} one or more {model_name} objects."
            if "{id}" in self.path:
                return f"{action} a {model_name} object."
            return f"{action} a list of {model_name} objects."

        # Give up
        return super().get_description()

    def get_filter_backends(self):
        """Nautobot's custom bulk operations, even though they return a list of records, are NOT filterable."""
        if self.is_bulk_action:
            return []
        return super().get_filter_backends()

    def get_operation(self, *args, **kwargs):
        operation = super().get_operation(*args, **kwargs)
        # drf-spectacular never generates a requestBody for DELETE operations, but our bulk-delete operations need one
        if operation is not None and "requestBody" not in operation and self.is_bulk_action and self.method == "DELETE":
            # based on drf-spectacular's `_get_request_body()`, `_get_request_for_media_type()`,
            # `_unwrap_list_serializer()`, and `_get_request_for_media_type()` methods
            request_serializer = self.get_request_serializer()

            # We skip past a number of checks from the aforementioned private methods, as this is a very specific case
            component = self.resolve_serializer(request_serializer.child, "request")

            operation["requestBody"] = {
                "content": {
                    media_type: build_media_type_object(build_array_type(component.ref))
                    for media_type in self.map_parsers()
                },
                "required": True,
            }

        # Inject a custom description for the "id" parameter since ours has custom lookup behavior.
        if operation is not None and "parameters" in operation:
            for param in operation["parameters"]:
                if param["name"] == "id" and "description" not in param:
                    param["description"] = "Unique object identifier, either a UUID primary key or a composite key."
            if self.method == "GET":
                if "depth" not in operation["parameters"]:
                    operation["parameters"].append(
                        {
                            "in": "query",
                            "name": "depth",
                            "required": False,
                            "description": "Serializer Depth",
                            "schema": {"type": "integer", "minimum": 0, "maximum": 10, "default": 1},
                        }
                    )
        return operation

    def get_operation_id(self):
        """Extend the base method to handle Nautobot's REST API bulk operations.

        Without this extension, every one of our ModelViewSet classes will result in drf-spectacular complaining
        about operationId collisions, e.g. between DELETE /api/dcim/devices/ and DELETE /api/dcim/devices/<pk>/ would
        both get resolved to the same "dcim_devices_destroy" operation-id and this would make drf-spectacular complain.

        With this extension, the bulk endpoints automatically get a different operation-id from the non-bulk endpoints.
        """
        if self.is_bulk_action:
            # Same basic sequence of calls as AutoSchema.get_operation_id,
            # except we use "self.view.action" instead of "self.method_mapping[self.method]" to get the action verb
            tokenized_path = self._tokenize_path()
            tokenized_path = [t.replace("-", "_") for t in tokenized_path]

            action = self.view.action

            if not tokenized_path:
                tokenized_path.append("root")
            if re.search(r"<drf_format_suffix\w*:\w+>", self.path_regex):
                tokenized_path.append("formatted")

            return "_".join([*tokenized_path, action])

        # For all other view actions, operation-id is the same as in the base class
        return super().get_operation_id()

    def get_request_serializer(self):
        """
        Return the request serializer (used for describing/parsing the request payload) for this endpoint.

        We override the default drf-spectacular behavior for the case where the endpoint describes a write request
        with required data (PATCH, POST, PUT). In those cases we replace FooSerializer with a dynamically-defined
        WritableFooSerializer class in order to more accurately represent the available options on write.

        We also override for the case where the endpoint is one of Nautobot's custom bulk API endpoints, which
        require a list of serializers as input, rather than a single one.
        """
        serializer = super().get_request_serializer()

        # For bulk operations, make sure we use a "many" serializer.
        many = self.is_bulk_action
        partial = self.is_partial_action

        if serializer is not None and self.method in ["PATCH", "POST", "PUT"]:
            writable_class = self.get_writable_class(serializer, bulk=many)
            if writable_class is not None:
                if hasattr(serializer, "child"):
                    child_serializer = self.get_writable_class(serializer.child, bulk=many)
                    serializer = writable_class(child=child_serializer, many=many, partial=partial)
                else:
                    serializer = writable_class(many=many, partial=partial)

        return serializer

    def get_response_serializers(self):
        """
        Return the response serializer (used for describing the response payload) for this endpoint.

        We override the default drf-spectacular behavior for the case where the endpoint describes a write request
        to a bulk endpoint, which returns a list of serializers, rather than a single one.
        """
        response_serializers = super().get_response_serializers()

        if self.is_bulk_action:
            if is_serializer(response_serializers):
                return type(response_serializers)(many=True)

        return response_serializers

    # Cache of existing dynamically-defined WritableFooSerializer classes.
    writable_serializers = {}

    def get_writable_class(self, serializer, bulk=False):
        """
        Given a FooSerializer instance, look up or construct a [Bulk]WritableFooSerializer class if necessary.

        If no [Bulk]WritableFooSerializer class is needed, returns None instead.
        """
        properties = {}
        # Does this serializer have any fields of certain special types?
        # These are the field types that are asymmetric between request (write) and response (read); if any such fields
        # are present, we should generate a distinct WritableFooSerializer to reflect that asymmetry in the schema.
        fields = {} if hasattr(serializer, "child") else serializer.fields
        for child_name, child in fields.items():
            # Don't consider read_only fields (since we're planning specifically for the writable serializer).
            if child.read_only:
                continue

            if isinstance(child, (ChoiceField, WritableNestedSerializer)):
                properties[child_name] = None
            elif isinstance(child, ManyRelatedField) and isinstance(child.child_relation, SerializedPKRelatedField):
                properties[child_name] = None

        if bulk:
            # The "id" field is always different in bulk serializers
            properties["id"] = None

        if not properties:
            # There's nothing about this serializer that requires a special WritableSerializer class to be defined.
            return None

        # Have we already created a [Bulk]WritableFooSerializer class or do we need to do so now?
        writable_name = "Writable" + type(serializer).__name__
        if bulk:
            writable_name = f"Bulk{writable_name}"
        if writable_name not in self.writable_serializers:
            # We need to create a new class to use
            # If the original serializer class has a Meta, make sure we set Meta.ref_name appropriately
            meta_class = getattr(type(serializer), "Meta", None)
            if meta_class:
                ref_name = "Writable" + self.get_serializer_ref_name(serializer)
                if bulk:
                    ref_name = f"Bulk{ref_name}"
                writable_meta = type("Meta", (meta_class,), {"ref_name": ref_name})
                properties["Meta"] = writable_meta

            # Define and cache a new [Bulk]WritableFooSerializer class
            if bulk:

                def get_fields(self):
                    """For Nautobot's bulk_update/partial_update/delete APIs, the `id` field is mandatory."""
                    new_fields = {}
                    for name, field in type(serializer)().get_fields().items():
                        if name == "id":
                            field.read_only = False
                            field.required = True
                        new_fields[name] = field
                    return new_fields

                properties["get_fields"] = get_fields

            self.writable_serializers[writable_name] = type(writable_name, (type(serializer),), properties)

        writable_class = self.writable_serializers[writable_name]
        return writable_class

    def get_serializer_ref_name(self, serializer):
        """
        Get the serializer's ref_name Meta attribute if set, or else derive a ref_name automatically.

        Based on drf_yasg.utils.get_serializer_ref_name().
        """
        serializer_meta = getattr(serializer, "Meta", None)
        if hasattr(serializer_meta, "ref_name"):
            return serializer_meta.ref_name
        serializer_name = type(serializer).__name__
        if serializer_name == "NestedSerializer" and isinstance(serializer, serializers.ModelSerializer):
            return None
        ref_name = serializer_name
        if ref_name.endswith("Serializer"):
            ref_name = ref_name[: -len("Serializer")]
        return ref_name

    def resolve_serializer(self, serializer, direction, bypass_extensions=False):
        """
        Re-add required `id` field on bulk_partial_update action.

        drf-spectacular clears the `required` list for any partial serializers in its `_map_basic_serializer()`,
        but Nautobot bulk partial updates require the `id` field to be specified for each object to update.
        """
        component = super().resolve_serializer(serializer, direction, bypass_extensions)
        if (
            component
            and component.schema is not None
            and self.is_bulk_action
            and self.is_partial_action
            and direction == "request"
        ):
            component.schema["required"] = ["id"]
        return component

is_bulk_action property

Custom property for convenience.

is_partial_action property

Custom property for convenience.

get_description()

Get the appropriate description for a given API endpoint.

By default, if a specific action doesn't have its own docstring, and neither does the view class, drf-spectacular will walk up the MRO of the view class until it finds a docstring, and use that. Most of our viewsets (for better or for worse) do not have docstrings, and so it'll find and use the generic docstring of the NautobotModelViewSet class, which isn't very useful to the end user. Instead of doing that, we only use the docstring of the view itself (ignoring its parent class docstrings), or if none exists, we make an attempt at rendering a basically accurate default description.

Source code in nautobot/core/api/schema.py
def get_description(self):
    """
    Get the appropriate description for a given API endpoint.

    By default, if a specific action doesn't have its own docstring, and neither does the view class,
    drf-spectacular will walk up the MRO of the view class until it finds a docstring, and use that.
    Most of our viewsets (for better or for worse) do not have docstrings, and so it'll find and use the generic
    docstring of the `NautobotModelViewSet` class, which isn't very useful to the end user. Instead of doing that,
    we only use the docstring of the view itself (ignoring its parent class docstrings), or if none exists, we
    make an attempt at rendering a basically accurate default description.
    """
    action_or_method = getattr(self.view, getattr(self.view, "action", self.method.lower()), None)
    action_doc = get_doc(action_or_method)
    if action_doc:
        return action_doc

    if self.view.__doc__:
        view_doc = get_doc(self.view.__class__)
        if view_doc:
            return view_doc

    # Fall back to a generic default description
    if hasattr(self.view, "queryset") and self.method.lower() in self.method_mapping:
        action = self.method_mapping[self.method.lower()].replace("_", " ").capitalize()
        model_name = self.view.queryset.model._meta.verbose_name
        if action == "Create":
            return f"{action} one or more {model_name} objects."
        if "{id}" in self.path:
            return f"{action} a {model_name} object."
        return f"{action} a list of {model_name} objects."

    # Give up
    return super().get_description()

get_filter_backends()

Nautobot's custom bulk operations, even though they return a list of records, are NOT filterable.

Source code in nautobot/core/api/schema.py
def get_filter_backends(self):
    """Nautobot's custom bulk operations, even though they return a list of records, are NOT filterable."""
    if self.is_bulk_action:
        return []
    return super().get_filter_backends()

get_operation_id()

Extend the base method to handle Nautobot's REST API bulk operations.

Without this extension, every one of our ModelViewSet classes will result in drf-spectacular complaining about operationId collisions, e.g. between DELETE /api/dcim/devices/ and DELETE /api/dcim/devices// would both get resolved to the same "dcim_devices_destroy" operation-id and this would make drf-spectacular complain.

With this extension, the bulk endpoints automatically get a different operation-id from the non-bulk endpoints.

Source code in nautobot/core/api/schema.py
def get_operation_id(self):
    """Extend the base method to handle Nautobot's REST API bulk operations.

    Without this extension, every one of our ModelViewSet classes will result in drf-spectacular complaining
    about operationId collisions, e.g. between DELETE /api/dcim/devices/ and DELETE /api/dcim/devices/<pk>/ would
    both get resolved to the same "dcim_devices_destroy" operation-id and this would make drf-spectacular complain.

    With this extension, the bulk endpoints automatically get a different operation-id from the non-bulk endpoints.
    """
    if self.is_bulk_action:
        # Same basic sequence of calls as AutoSchema.get_operation_id,
        # except we use "self.view.action" instead of "self.method_mapping[self.method]" to get the action verb
        tokenized_path = self._tokenize_path()
        tokenized_path = [t.replace("-", "_") for t in tokenized_path]

        action = self.view.action

        if not tokenized_path:
            tokenized_path.append("root")
        if re.search(r"<drf_format_suffix\w*:\w+>", self.path_regex):
            tokenized_path.append("formatted")

        return "_".join([*tokenized_path, action])

    # For all other view actions, operation-id is the same as in the base class
    return super().get_operation_id()

get_request_serializer()

Return the request serializer (used for describing/parsing the request payload) for this endpoint.

We override the default drf-spectacular behavior for the case where the endpoint describes a write request with required data (PATCH, POST, PUT). In those cases we replace FooSerializer with a dynamically-defined WritableFooSerializer class in order to more accurately represent the available options on write.

We also override for the case where the endpoint is one of Nautobot's custom bulk API endpoints, which require a list of serializers as input, rather than a single one.

Source code in nautobot/core/api/schema.py
def get_request_serializer(self):
    """
    Return the request serializer (used for describing/parsing the request payload) for this endpoint.

    We override the default drf-spectacular behavior for the case where the endpoint describes a write request
    with required data (PATCH, POST, PUT). In those cases we replace FooSerializer with a dynamically-defined
    WritableFooSerializer class in order to more accurately represent the available options on write.

    We also override for the case where the endpoint is one of Nautobot's custom bulk API endpoints, which
    require a list of serializers as input, rather than a single one.
    """
    serializer = super().get_request_serializer()

    # For bulk operations, make sure we use a "many" serializer.
    many = self.is_bulk_action
    partial = self.is_partial_action

    if serializer is not None and self.method in ["PATCH", "POST", "PUT"]:
        writable_class = self.get_writable_class(serializer, bulk=many)
        if writable_class is not None:
            if hasattr(serializer, "child"):
                child_serializer = self.get_writable_class(serializer.child, bulk=many)
                serializer = writable_class(child=child_serializer, many=many, partial=partial)
            else:
                serializer = writable_class(many=many, partial=partial)

    return serializer

get_response_serializers()

Return the response serializer (used for describing the response payload) for this endpoint.

We override the default drf-spectacular behavior for the case where the endpoint describes a write request to a bulk endpoint, which returns a list of serializers, rather than a single one.

Source code in nautobot/core/api/schema.py
def get_response_serializers(self):
    """
    Return the response serializer (used for describing the response payload) for this endpoint.

    We override the default drf-spectacular behavior for the case where the endpoint describes a write request
    to a bulk endpoint, which returns a list of serializers, rather than a single one.
    """
    response_serializers = super().get_response_serializers()

    if self.is_bulk_action:
        if is_serializer(response_serializers):
            return type(response_serializers)(many=True)

    return response_serializers

get_serializer_ref_name(serializer)

Get the serializer's ref_name Meta attribute if set, or else derive a ref_name automatically.

Based on drf_yasg.utils.get_serializer_ref_name().

Source code in nautobot/core/api/schema.py
def get_serializer_ref_name(self, serializer):
    """
    Get the serializer's ref_name Meta attribute if set, or else derive a ref_name automatically.

    Based on drf_yasg.utils.get_serializer_ref_name().
    """
    serializer_meta = getattr(serializer, "Meta", None)
    if hasattr(serializer_meta, "ref_name"):
        return serializer_meta.ref_name
    serializer_name = type(serializer).__name__
    if serializer_name == "NestedSerializer" and isinstance(serializer, serializers.ModelSerializer):
        return None
    ref_name = serializer_name
    if ref_name.endswith("Serializer"):
        ref_name = ref_name[: -len("Serializer")]
    return ref_name

get_writable_class(serializer, bulk=False)

Given a FooSerializer instance, look up or construct a [Bulk]WritableFooSerializer class if necessary.

If no [Bulk]WritableFooSerializer class is needed, returns None instead.

Source code in nautobot/core/api/schema.py
def get_writable_class(self, serializer, bulk=False):
    """
    Given a FooSerializer instance, look up or construct a [Bulk]WritableFooSerializer class if necessary.

    If no [Bulk]WritableFooSerializer class is needed, returns None instead.
    """
    properties = {}
    # Does this serializer have any fields of certain special types?
    # These are the field types that are asymmetric between request (write) and response (read); if any such fields
    # are present, we should generate a distinct WritableFooSerializer to reflect that asymmetry in the schema.
    fields = {} if hasattr(serializer, "child") else serializer.fields
    for child_name, child in fields.items():
        # Don't consider read_only fields (since we're planning specifically for the writable serializer).
        if child.read_only:
            continue

        if isinstance(child, (ChoiceField, WritableNestedSerializer)):
            properties[child_name] = None
        elif isinstance(child, ManyRelatedField) and isinstance(child.child_relation, SerializedPKRelatedField):
            properties[child_name] = None

    if bulk:
        # The "id" field is always different in bulk serializers
        properties["id"] = None

    if not properties:
        # There's nothing about this serializer that requires a special WritableSerializer class to be defined.
        return None

    # Have we already created a [Bulk]WritableFooSerializer class or do we need to do so now?
    writable_name = "Writable" + type(serializer).__name__
    if bulk:
        writable_name = f"Bulk{writable_name}"
    if writable_name not in self.writable_serializers:
        # We need to create a new class to use
        # If the original serializer class has a Meta, make sure we set Meta.ref_name appropriately
        meta_class = getattr(type(serializer), "Meta", None)
        if meta_class:
            ref_name = "Writable" + self.get_serializer_ref_name(serializer)
            if bulk:
                ref_name = f"Bulk{ref_name}"
            writable_meta = type("Meta", (meta_class,), {"ref_name": ref_name})
            properties["Meta"] = writable_meta

        # Define and cache a new [Bulk]WritableFooSerializer class
        if bulk:

            def get_fields(self):
                """For Nautobot's bulk_update/partial_update/delete APIs, the `id` field is mandatory."""
                new_fields = {}
                for name, field in type(serializer)().get_fields().items():
                    if name == "id":
                        field.read_only = False
                        field.required = True
                    new_fields[name] = field
                return new_fields

            properties["get_fields"] = get_fields

        self.writable_serializers[writable_name] = type(writable_name, (type(serializer),), properties)

    writable_class = self.writable_serializers[writable_name]
    return writable_class

resolve_serializer(serializer, direction, bypass_extensions=False)

Re-add required id field on bulk_partial_update action.

drf-spectacular clears the required list for any partial serializers in its _map_basic_serializer(), but Nautobot bulk partial updates require the id field to be specified for each object to update.

Source code in nautobot/core/api/schema.py
def resolve_serializer(self, serializer, direction, bypass_extensions=False):
    """
    Re-add required `id` field on bulk_partial_update action.

    drf-spectacular clears the `required` list for any partial serializers in its `_map_basic_serializer()`,
    but Nautobot bulk partial updates require the `id` field to be specified for each object to update.
    """
    component = super().resolve_serializer(serializer, direction, bypass_extensions)
    if (
        component
        and component.schema is not None
        and self.is_bulk_action
        and self.is_partial_action
        and direction == "request"
    ):
        component.schema["required"] = ["id"]
    return component

nautobot.apps.api.NautobotCSVParser

Bases: BaseParser

Counterpart to NautobotCSVRenderer - import CSV data.

Source code in nautobot/core/api/parsers.py
class NautobotCSVParser(BaseParser):
    """Counterpart to NautobotCSVRenderer - import CSV data."""

    media_type = "text/csv"

    def parse(self, stream, media_type=None, parser_context=None):
        parser_context = parser_context or {}
        encoding = parser_context.get("encoding", "UTF-8")
        try:
            if "serializer_class" in parser_context:
                # UI bulk-import case
                serializer_class = parser_context["serializer_class"]
            else:
                # REST API case
                serializer_class = parser_context["view"].get_serializer_class()
        except (KeyError, AttributeError):
            raise ParseError("No serializer_class was provided by the parser_context")
        if serializer_class is None:
            raise ParseError("Serializer class for this parser_context is None, unable to proceed")

        serializer = serializer_class(context={"request": parser_context.get("request", None), "depth": 0})

        try:
            text = stream.read().decode(encoding)
            reader = csv.DictReader(StringIO(text))

            data = []
            for counter, row in enumerate(reader, start=1):
                data.append(self.row_elements_to_data(counter, row, serializer=serializer))

            if "pk" in parser_context.get("kwargs", {}):
                # Single-object update, not bulk update - strip it so that we get the expected input and return format
                data = data[0]
            # Note that we can't distinguish between single-create and bulk-create with a list of one object,
            # as both would have the same CSV representation. Therefore create via CSV **always** acts as bulk-create,
            # and the response will always be a list of created objects, never a single object

            if settings.DEBUG:
                logger.debug("CSV loaded into data:\n%s", json.dumps(data, indent=2))
            return data
        except ParseError:
            raise
        except Exception as exc:
            raise ParseError(str(exc)) from exc

    def _group_data_by_field_name(self, data):
        """
        Converts a dictionary with flat keys separated by '__' into a nested dictionary structure suitable for serialization.

        Example:
            Input:
                {
                    'type': 'virtual',
                    'name': 'Interface 4',
                    'device__name': 'Device 1',
                    'device__tenant__name': '',
                    'device__location': 'Test+Location+1',
                    'status': 'Active',
                }

            Output:
                {
                    'type': 'virtual',
                    'name': 'Interface 4',
                    'device': {
                        'name': 'Device 1',
                        'location': 'Test+Location+1',
                        "tenant":{
                            "name": "",
                        }
                    },
                    'status': 'Active'
                }
        """

        def insert_nested_dict(keys, value, current_dict):
            key = keys[0]
            if len(keys) == 1:
                current_dict[key] = None if value in [CSV_NO_OBJECT, CSV_NULL_TYPE] else value
            else:
                current_dict[key] = current_dict.get(key, {})
                insert_nested_dict(keys[1:], value, current_dict[key])

        result_dict = {}
        for original_key, original_value in data.items():
            split_keys = original_key.split("__")
            insert_nested_dict(split_keys, original_value, result_dict)

        return result_dict

    def _field_lookups_not_empty(self, field_lookups):
        """Check if all values of the field lookups dict are not all NoObject"""
        return any(value != CSV_NO_OBJECT for value in field_lookups.values())

    def _remove_object_not_found_values(self, data):
        """Remove all `CSV_NO_OBJECT` field lookups from the given data, and swap out `CSV_NULL_TYPE` and
        'CSV_NO_OBJECT' values for `None`.

        If all the lookups for a field are 'CSV_NO_OBJECT', it indicates that the field does not exist,
        and it needs to be removed to prevent unnecessary database queries.

        Args:
            data (dict): A dictionary containing field natural key lookups and their corresponding values.

        Returns:
            dict: A modified dictionary with field lookups of 'CSV_NO_OBJECT' values removed, and 'CSV_NULL_TYPE' and 'CSV_NO_OBJECT' swapped for `None`.
        """
        lookup_grouped_by_field_name = {}
        for lookup, lookup_value in data.items():
            field_name = lookup.split("__", 1)[0]
            lookup_grouped_by_field_name.setdefault(field_name, {}).update({lookup: lookup_value})

        # Ignore lookup groups which has all its values set to NoObject
        # These lookups fields do not exists
        data_without_missing_field_lookups_values = {
            lookup: lookup_value
            for lookup_group in lookup_grouped_by_field_name.values()
            for lookup, lookup_value in lookup_group.items()
            if self._field_lookups_not_empty(lookup_group)
        }

        return data_without_missing_field_lookups_values

    def row_elements_to_data(self, counter, row, serializer):
        """
        Parse a single row of CSV data (represented as a dict) into a dict suitable for consumption by the serializer.

        TODO: it would be more elegant if our serializer fields knew how to deserialize the CSV data themselves;
        could we then literally have the parser just return list(reader) and not need this function at all?
        """
        data = {}
        valid_row_data = self._remove_object_not_found_values(row)
        fields_value_mapping = self._group_data_by_field_name(valid_row_data)
        for column, key in enumerate(fields_value_mapping.keys(), start=1):
            if not key:
                raise ParseError(f"Row {counter}: Column {column}: missing/empty header for this column")

            value = fields_value_mapping[key]
            if key.startswith("cf_"):
                # Custom field
                if value == "":
                    value = None
                data.setdefault("custom_fields", {})[key[3:]] = value
                continue

            serializer_field = serializer.fields.get(key, None)
            if serializer_field is None:
                # The REST API normally just ignores any columns the serializer doesn't understand
                logger.debug('Skipping unknown column "%s"', key)
                continue

            if serializer_field.read_only and key != "id":
                # Deserializing read-only fields is tricky, especially for things like SerializerMethodFields that
                # can potentially render as anything. We don't strictly need such fields (except "id" for bulk PATCH),
                # so let's just skip it.
                continue

            if isinstance(serializer_field, serializers.ManyRelatedField):
                # A list of related objects, represented as a list of composite-keys
                if value:
                    value = value.split(",")
                else:
                    value = []
            elif isinstance(serializer_field, serializers.RelatedField):
                # A single related object, represented by its composite-key
                if value:
                    pass
                else:
                    value = None
            elif isinstance(serializer_field, (serializers.ListField, serializers.MultipleChoiceField)):
                if value:
                    value = value.split(",")
                else:
                    value = []
            elif isinstance(serializer_field, (serializers.DictField, serializers.JSONField)):
                # We currently only store lists or dicts in JSONFields, never bare ints/strings.
                # On the CSV write side, we only render dicts to JSON
                if value is not None:
                    if value.startswith(("{", "[")):
                        value = json.loads(value)
                    else:
                        value = value.split(",")
                        try:
                            # We have some cases where it's a list of integers, such as in RackReservation.units
                            value = [int(v) for v in value]
                        except ValueError:
                            # Guess not!
                            pass

            # CSV doesn't provide a ready distinction between blank and null, so in this case we have to pick one.
            # This does mean that for a nullable AND blankable field, there's no way for CSV to set it to blank string.
            # See corresponding logic in NautobotCSVRenderer.
            if value == "" and serializer_field.allow_null:
                value = None

            data[key] = value

        return data

row_elements_to_data(counter, row, serializer)

Parse a single row of CSV data (represented as a dict) into a dict suitable for consumption by the serializer.

TODO: it would be more elegant if our serializer fields knew how to deserialize the CSV data themselves; could we then literally have the parser just return list(reader) and not need this function at all?

Source code in nautobot/core/api/parsers.py
def row_elements_to_data(self, counter, row, serializer):
    """
    Parse a single row of CSV data (represented as a dict) into a dict suitable for consumption by the serializer.

    TODO: it would be more elegant if our serializer fields knew how to deserialize the CSV data themselves;
    could we then literally have the parser just return list(reader) and not need this function at all?
    """
    data = {}
    valid_row_data = self._remove_object_not_found_values(row)
    fields_value_mapping = self._group_data_by_field_name(valid_row_data)
    for column, key in enumerate(fields_value_mapping.keys(), start=1):
        if not key:
            raise ParseError(f"Row {counter}: Column {column}: missing/empty header for this column")

        value = fields_value_mapping[key]
        if key.startswith("cf_"):
            # Custom field
            if value == "":
                value = None
            data.setdefault("custom_fields", {})[key[3:]] = value
            continue

        serializer_field = serializer.fields.get(key, None)
        if serializer_field is None:
            # The REST API normally just ignores any columns the serializer doesn't understand
            logger.debug('Skipping unknown column "%s"', key)
            continue

        if serializer_field.read_only and key != "id":
            # Deserializing read-only fields is tricky, especially for things like SerializerMethodFields that
            # can potentially render as anything. We don't strictly need such fields (except "id" for bulk PATCH),
            # so let's just skip it.
            continue

        if isinstance(serializer_field, serializers.ManyRelatedField):
            # A list of related objects, represented as a list of composite-keys
            if value:
                value = value.split(",")
            else:
                value = []
        elif isinstance(serializer_field, serializers.RelatedField):
            # A single related object, represented by its composite-key
            if value:
                pass
            else:
                value = None
        elif isinstance(serializer_field, (serializers.ListField, serializers.MultipleChoiceField)):
            if value:
                value = value.split(",")
            else:
                value = []
        elif isinstance(serializer_field, (serializers.DictField, serializers.JSONField)):
            # We currently only store lists or dicts in JSONFields, never bare ints/strings.
            # On the CSV write side, we only render dicts to JSON
            if value is not None:
                if value.startswith(("{", "[")):
                    value = json.loads(value)
                else:
                    value = value.split(",")
                    try:
                        # We have some cases where it's a list of integers, such as in RackReservation.units
                        value = [int(v) for v in value]
                    except ValueError:
                        # Guess not!
                        pass

        # CSV doesn't provide a ready distinction between blank and null, so in this case we have to pick one.
        # This does mean that for a nullable AND blankable field, there's no way for CSV to set it to blank string.
        # See corresponding logic in NautobotCSVRenderer.
        if value == "" and serializer_field.allow_null:
            value = None

        data[key] = value

    return data

nautobot.apps.api.NautobotHyperlinkedRelatedField

Bases: WritableSerializerMixin, serializers.HyperlinkedRelatedField

Extend HyperlinkedRelatedField to include URL namespace-awareness, add 'object_type' field, and read composite-keys.

Source code in nautobot/core/api/fields.py
@extend_schema_field(
    {
        "type": "object",
        "properties": {
            "id": {
                "oneOf": [
                    {"type": "string", "format": "uuid"},
                    {"type": "integer"},
                ]
            },
            "object_type": {
                "type": "string",
                "pattern": "^[a-z][a-z0-9_]+\\.[a-z][a-z0-9_]+$",
                "example": "app_label.modelname",
            },
            "url": {
                "type": "string",
                "format": "uri",
            },
        },
    }
)
class NautobotHyperlinkedRelatedField(WritableSerializerMixin, serializers.HyperlinkedRelatedField):
    """
    Extend HyperlinkedRelatedField to include URL namespace-awareness, add 'object_type' field, and read composite-keys.
    """

    def __init__(self, *args, **kwargs):
        """Override DRF's namespace-unaware default view_name logic for HyperlinkedRelatedField.

        DRF defaults to '{model_name}-detail' instead of '{app_label}:{model_name}-detail'.
        """
        if "view_name" not in kwargs or (kwargs["view_name"].endswith("-detail") and ":" not in kwargs["view_name"]):
            if "queryset" not in kwargs:
                logger.warning(
                    '"view_name=%r" is probably incorrect for this related API field; '
                    'unable to determine the correct "view_name" as "queryset" wasn\'t specified',
                    kwargs["view_name"],
                )
            else:
                kwargs["view_name"] = get_route_for_model(kwargs["queryset"].model, "detail", api=True)
        super().__init__(*args, **kwargs)

    @property
    def _related_model(self):
        """The model class that this field is referencing to."""
        if self.queryset is not None:
            return self.queryset.model
        # Foreign key where the destination is referenced by string rather than by Python class
        if getattr(self.parent.Meta.model, self.source, False):
            return getattr(self.parent.Meta.model, self.source).field.model

        logger.warning(
            "Unable to determine model for related field %r; "
            "ensure that either the field defines a 'queryset' or the Meta defines the related 'model'.",
            self.field_name,
        )
        return None

    def to_internal_value(self, data):
        """Convert potentially nested representation to a model instance."""
        if isinstance(data, dict):
            if "url" in data:
                return super().to_internal_value(data["url"])
            elif "id" in data:
                return super().to_internal_value(data["id"])
        if isinstance(data, str) and not is_uuid(data) and not is_url(data):
            # Maybe it's a composite-key?
            related_model = self._related_model
            if related_model is not None and hasattr(related_model, "natural_key_args_to_kwargs"):
                try:
                    data = related_model.natural_key_args_to_kwargs(deconstruct_composite_key(data))
                except ValueError as err:
                    # Not a correctly constructed composite key?
                    raise ValidationError(f"Related object not found using provided composite-key: {data}") from err
            elif related_model is not None and related_model.label_lower == "auth.group":
                # auth.Group is a base Django model and so doesn't implement our natural_key_args_to_kwargs() method
                data = {"name": deconstruct_composite_key(data)}
        return super().to_internal_value(data)

    def to_representation(self, value):
        """Convert URL representation to a brief nested representation."""
        url = super().to_representation(value)

        # nested serializer provides an instance
        if isinstance(value, Model):
            model = type(value)
        else:
            model = self._related_model

        if model is None:
            return {"id": value.pk, "object_type": "unknown.unknown", "url": url}
        return {"id": value.pk, "object_type": model._meta.label_lower, "url": url}

__init__(*args, **kwargs)

Override DRF's namespace-unaware default view_name logic for HyperlinkedRelatedField.

DRF defaults to '{model_name}-detail' instead of '{app_label}:{model_name}-detail'.

Source code in nautobot/core/api/fields.py
def __init__(self, *args, **kwargs):
    """Override DRF's namespace-unaware default view_name logic for HyperlinkedRelatedField.

    DRF defaults to '{model_name}-detail' instead of '{app_label}:{model_name}-detail'.
    """
    if "view_name" not in kwargs or (kwargs["view_name"].endswith("-detail") and ":" not in kwargs["view_name"]):
        if "queryset" not in kwargs:
            logger.warning(
                '"view_name=%r" is probably incorrect for this related API field; '
                'unable to determine the correct "view_name" as "queryset" wasn\'t specified',
                kwargs["view_name"],
            )
        else:
            kwargs["view_name"] = get_route_for_model(kwargs["queryset"].model, "detail", api=True)
    super().__init__(*args, **kwargs)

to_internal_value(data)

Convert potentially nested representation to a model instance.

Source code in nautobot/core/api/fields.py
def to_internal_value(self, data):
    """Convert potentially nested representation to a model instance."""
    if isinstance(data, dict):
        if "url" in data:
            return super().to_internal_value(data["url"])
        elif "id" in data:
            return super().to_internal_value(data["id"])
    if isinstance(data, str) and not is_uuid(data) and not is_url(data):
        # Maybe it's a composite-key?
        related_model = self._related_model
        if related_model is not None and hasattr(related_model, "natural_key_args_to_kwargs"):
            try:
                data = related_model.natural_key_args_to_kwargs(deconstruct_composite_key(data))
            except ValueError as err:
                # Not a correctly constructed composite key?
                raise ValidationError(f"Related object not found using provided composite-key: {data}") from err
        elif related_model is not None and related_model.label_lower == "auth.group":
            # auth.Group is a base Django model and so doesn't implement our natural_key_args_to_kwargs() method
            data = {"name": deconstruct_composite_key(data)}
    return super().to_internal_value(data)

to_representation(value)

Convert URL representation to a brief nested representation.

Source code in nautobot/core/api/fields.py
def to_representation(self, value):
    """Convert URL representation to a brief nested representation."""
    url = super().to_representation(value)

    # nested serializer provides an instance
    if isinstance(value, Model):
        model = type(value)
    else:
        model = self._related_model

    if model is None:
        return {"id": value.pk, "object_type": "unknown.unknown", "url": url}
    return {"id": value.pk, "object_type": model._meta.label_lower, "url": url}

nautobot.apps.api.NautobotModelSerializer

Bases: RelationshipModelSerializerMixin, CustomFieldModelSerializerMixin, NotesSerializerMixin, ValidatedModelSerializer

Base class to use for serializers based on OrganizationalModel or PrimaryModel.

Can also be used for models derived from BaseModel, so long as they support custom fields, notes, and relationships.

Source code in nautobot/core/api/serializers.py
class NautobotModelSerializer(
    RelationshipModelSerializerMixin, CustomFieldModelSerializerMixin, NotesSerializerMixin, ValidatedModelSerializer
):
    """Base class to use for serializers based on OrganizationalModel or PrimaryModel.

    Can also be used for models derived from BaseModel, so long as they support custom fields, notes, and relationships.
    """

nautobot.apps.api.NautobotModelViewSet

Bases: NotesViewSetMixin, CustomFieldModelViewSet

Base class to use for API ViewSets based on OrganizationalModel or PrimaryModel.

Can also be used for models derived from BaseModel, so long as they support Notes.

Source code in nautobot/extras/api/views.py
class NautobotModelViewSet(NotesViewSetMixin, CustomFieldModelViewSet):
    """Base class to use for API ViewSets based on OrganizationalModel or PrimaryModel.

    Can also be used for models derived from BaseModel, so long as they support Notes.
    """

nautobot.apps.api.NotesSerializerMixin

Bases: BaseModelSerializer

Extend Serializer with a notes field.

Source code in nautobot/core/api/serializers.py
class NotesSerializerMixin(BaseModelSerializer):
    """Extend Serializer with a `notes` field."""

    notes_url = serializers.SerializerMethodField()

    def get_field_names(self, declared_fields, info):
        """Ensure that fields includes "notes_url" field if applicable."""
        fields = list(super().get_field_names(declared_fields, info))
        if hasattr(self.Meta.model, "notes"):
            # Make sure the field is at the end of fields, instead of the beginning
            self.extend_field_names(fields, "notes_url")
        return fields

    @extend_schema_field(serializers.URLField())
    def get_notes_url(self, instance):
        try:
            notes_url = get_route_for_model(instance, "notes", api=True)
            return reverse(notes_url, args=[instance.id], request=self.context["request"])
        except NoReverseMatch:
            model_name = type(instance).__name__
            logger.warning(
                (
                    f"Notes feature is not available for model {model_name}. "
                    "Please make sure to: "
                    f"1. Include NotesMixin from nautobot.extras.model.mixins in the {model_name} class definition "
                    f"2. Include NotesViewSetMixin from nautobot.extras.api.views in the {model_name}ViewSet "
                    "before including NotesSerializerMixin in the model serializer"
                )
            )

            return None

get_field_names(declared_fields, info)

Ensure that fields includes "notes_url" field if applicable.

Source code in nautobot/core/api/serializers.py
def get_field_names(self, declared_fields, info):
    """Ensure that fields includes "notes_url" field if applicable."""
    fields = list(super().get_field_names(declared_fields, info))
    if hasattr(self.Meta.model, "notes"):
        # Make sure the field is at the end of fields, instead of the beginning
        self.extend_field_names(fields, "notes_url")
    return fields

nautobot.apps.api.NotesViewSetMixin

Source code in nautobot/extras/api/views.py
class NotesViewSetMixin:
    def restrict_queryset(self, request, *args, **kwargs):
        """
        Apply "view" permissions on the POST /notes/ endpoint, otherwise as ModelViewSetMixin.
        """
        if request.user.is_authenticated and self.action == "notes":
            self.queryset = self.queryset.restrict(request.user, "view")
        else:
            super().restrict_queryset(request, *args, **kwargs)

    class CreateNotePermissions(TokenPermissions):
        """As nautobot.core.api.authentication.TokenPermissions, but enforcing add_note permission."""

        perms_map = {
            "GET": ["%(app_label)s.view_%(model_name)s", "extras.view_note"],
            "POST": ["%(app_label)s.view_%(model_name)s", "extras.add_note"],
        }

    @extend_schema(methods=["get"], filters=False, responses={200: serializers.NoteSerializer(many=True)})
    @extend_schema(
        methods=["post"],
        request=serializers.NoteInputSerializer,
        responses={201: serializers.NoteSerializer(many=False)},
    )
    @action(detail=True, url_path="notes", methods=["get", "post"], permission_classes=[CreateNotePermissions])
    def notes(self, request, *args, **kwargs):
        """
        API methods for returning or creating notes on an object.
        """
        obj = get_object_or_404(
            self.queryset, **{self.lookup_field: self.kwargs[self.lookup_url_kwarg or self.lookup_field]}
        )
        if request.method == "POST":
            content_type = ContentType.objects.get_for_model(obj)
            data = request.data
            data["assigned_object_id"] = obj.pk
            data["assigned_object_type"] = f"{content_type.app_label}.{content_type.model}"
            serializer = serializers.NoteSerializer(data=data, context={"request": request})

            # Create the new Note.
            serializer.is_valid(raise_exception=True)
            serializer.save(user=request.user)
            return Response(serializer.data, status=status.HTTP_201_CREATED)

        else:
            notes = self.paginate_queryset(obj.notes)
            serializer = serializers.NoteSerializer(notes, many=True, context={"request": request})

        return self.get_paginated_response(serializer.data)

CreateNotePermissions

Bases: TokenPermissions

As nautobot.core.api.authentication.TokenPermissions, but enforcing add_note permission.

Source code in nautobot/extras/api/views.py
class CreateNotePermissions(TokenPermissions):
    """As nautobot.core.api.authentication.TokenPermissions, but enforcing add_note permission."""

    perms_map = {
        "GET": ["%(app_label)s.view_%(model_name)s", "extras.view_note"],
        "POST": ["%(app_label)s.view_%(model_name)s", "extras.add_note"],
    }

notes(request, *args, **kwargs)

API methods for returning or creating notes on an object.

Source code in nautobot/extras/api/views.py
@extend_schema(methods=["get"], filters=False, responses={200: serializers.NoteSerializer(many=True)})
@extend_schema(
    methods=["post"],
    request=serializers.NoteInputSerializer,
    responses={201: serializers.NoteSerializer(many=False)},
)
@action(detail=True, url_path="notes", methods=["get", "post"], permission_classes=[CreateNotePermissions])
def notes(self, request, *args, **kwargs):
    """
    API methods for returning or creating notes on an object.
    """
    obj = get_object_or_404(
        self.queryset, **{self.lookup_field: self.kwargs[self.lookup_url_kwarg or self.lookup_field]}
    )
    if request.method == "POST":
        content_type = ContentType.objects.get_for_model(obj)
        data = request.data
        data["assigned_object_id"] = obj.pk
        data["assigned_object_type"] = f"{content_type.app_label}.{content_type.model}"
        serializer = serializers.NoteSerializer(data=data, context={"request": request})

        # Create the new Note.
        serializer.is_valid(raise_exception=True)
        serializer.save(user=request.user)
        return Response(serializer.data, status=status.HTTP_201_CREATED)

    else:
        notes = self.paginate_queryset(obj.notes)
        serializer = serializers.NoteSerializer(notes, many=True, context={"request": request})

    return self.get_paginated_response(serializer.data)

restrict_queryset(request, *args, **kwargs)

Apply "view" permissions on the POST /notes/ endpoint, otherwise as ModelViewSetMixin.

Source code in nautobot/extras/api/views.py
def restrict_queryset(self, request, *args, **kwargs):
    """
    Apply "view" permissions on the POST /notes/ endpoint, otherwise as ModelViewSetMixin.
    """
    if request.user.is_authenticated and self.action == "notes":
        self.queryset = self.queryset.restrict(request.user, "view")
    else:
        super().restrict_queryset(request, *args, **kwargs)

nautobot.apps.api.ObjectTypeField

Bases: serializers.CharField

Represent the ContentType of this serializer's model as ".".

Source code in nautobot/core/api/fields.py
@extend_schema_field(
    {
        "type": "string",
        "pattern": "^[a-z][a-z0-9_]+\\.[a-z][a-z0-9_]+$",
        "example": "app_label.modelname",
    }
)
class ObjectTypeField(serializers.CharField):
    """
    Represent the ContentType of this serializer's model as "<app_label>.<model>".
    """

    def __init__(self, *args, read_only=True, source="*", **kwargs):  # pylint: disable=useless-parent-delegation
        """Default read_only to True as this should never be a writable field."""
        super().__init__(*args, read_only=read_only, source=source, **kwargs)

    def to_representation(self, _value):
        """
        Get the content-type of this serializer's model.

        Implemented this way because `_value` may be None when generating the schema.
        """
        model = self.parent.Meta.model
        return model._meta.label_lower

__init__(*args, read_only=True, source='*', **kwargs)

Default read_only to True as this should never be a writable field.

Source code in nautobot/core/api/fields.py
def __init__(self, *args, read_only=True, source="*", **kwargs):  # pylint: disable=useless-parent-delegation
    """Default read_only to True as this should never be a writable field."""
    super().__init__(*args, read_only=read_only, source=source, **kwargs)

to_representation(_value)

Get the content-type of this serializer's model.

Implemented this way because _value may be None when generating the schema.

Source code in nautobot/core/api/fields.py
def to_representation(self, _value):
    """
    Get the content-type of this serializer's model.

    Implemented this way because `_value` may be None when generating the schema.
    """
    model = self.parent.Meta.model
    return model._meta.label_lower

nautobot.apps.api.OptInFieldsMixin

A serializer mixin that takes an additional opt_in_fields argument that controls which fields should be displayed.

Source code in nautobot/core/api/serializers.py
class OptInFieldsMixin:
    """
    A serializer mixin that takes an additional `opt_in_fields` argument that controls
    which fields should be displayed.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.__pruned_fields = None

    @property
    def fields(self):
        """
        Removes all serializer fields specified in a serializers `opt_in_fields` list that aren't specified in the
        `include` query parameter.

        As an example, if the serializer specifies that `opt_in_fields = ["computed_fields"]`
        but `computed_fields` is not specified in the `?include` query parameter, `computed_fields` will be popped
        from the list of fields.
        """
        if self.__pruned_fields is None:
            fields = dict(super().fields)
            serializer_opt_in_fields = getattr(self.Meta, "opt_in_fields", None)

            if not serializer_opt_in_fields:
                # This serializer has no defined opt_in_fields, so we never need to go further than this
                self.__pruned_fields = fields
                return self.__pruned_fields

            if not hasattr(self, "_context"):
                # We are being called before a request cycle
                return fields

            try:
                request = self.context["request"]
            except KeyError:
                # No available request?
                return fields

            # opt-in fields only applies on GET requests, for other methods we support these fields regardless
            if request is not None and request.method != "GET":
                return fields

            # NOTE: drf test framework builds a request object where the query
            # parameters are found under the GET attribute.
            params = normalize_querydict(getattr(request, "query_params", getattr(request, "GET", None)))

            try:
                user_opt_in_fields = params.get("include", [])
            except AttributeError:
                # include parameter was not specified
                user_opt_in_fields = []

            # Drop any fields that are not specified in the users opt in fields
            for field in serializer_opt_in_fields:
                if field not in user_opt_in_fields:
                    fields.pop(field, None)

            self.__pruned_fields = fields

        return self.__pruned_fields

fields property

Removes all serializer fields specified in a serializers opt_in_fields list that aren't specified in the include query parameter.

As an example, if the serializer specifies that opt_in_fields = ["computed_fields"] but computed_fields is not specified in the ?include query parameter, computed_fields will be popped from the list of fields.

nautobot.apps.api.OrderedDefaultRouter

Bases: DefaultRouter

Source code in nautobot/core/api/routers.py
class OrderedDefaultRouter(DefaultRouter):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Extend the list view mappings to support the DELETE operation
        self.routes[0].mapping.update(
            {
                "put": "bulk_update",
                "patch": "bulk_partial_update",
                "delete": "bulk_destroy",
            }
        )

    def get_api_root_view(self, api_urls=None):
        """
        Wrap DRF's DefaultRouter to return an alphabetized list of endpoints.
        """
        api_root_dict = OrderedDict()
        list_name = self.routes[0].name
        for prefix, _viewset, basename in sorted(self.registry, key=lambda x: x[0]):
            api_root_dict[prefix] = list_name.format(basename=basename)

        return self.APIRootView.as_view(api_root_dict=api_root_dict)

get_api_root_view(api_urls=None)

Wrap DRF's DefaultRouter to return an alphabetized list of endpoints.

Source code in nautobot/core/api/routers.py
def get_api_root_view(self, api_urls=None):
    """
    Wrap DRF's DefaultRouter to return an alphabetized list of endpoints.
    """
    api_root_dict = OrderedDict()
    list_name = self.routes[0].name
    for prefix, _viewset, basename in sorted(self.registry, key=lambda x: x[0]):
        api_root_dict[prefix] = list_name.format(basename=basename)

    return self.APIRootView.as_view(api_root_dict=api_root_dict)

nautobot.apps.api.ReadOnlyModelViewSet

Bases: NautobotAPIVersionMixin, ModelViewSetMixin, ReadOnlyModelViewSet_

Extend DRF's ReadOnlyModelViewSet to support queryset restriction.

Source code in nautobot/core/api/views.py
class ReadOnlyModelViewSet(NautobotAPIVersionMixin, ModelViewSetMixin, ReadOnlyModelViewSet_):
    """
    Extend DRF's ReadOnlyModelViewSet to support queryset restriction.
    """

nautobot.apps.api.RelationshipModelSerializerMixin

Bases: ValidatedModelSerializer

Extend ValidatedModelSerializer with a relationships field.

Source code in nautobot/core/api/serializers.py
class RelationshipModelSerializerMixin(ValidatedModelSerializer):
    """Extend ValidatedModelSerializer with a `relationships` field."""

    # TODO # 3024 need to change this as well to show just pks in depth=0
    relationships = RelationshipsDataField(required=False, source="*")

    def create(self, validated_data):
        relationships_data = validated_data.pop("relationships", {})
        required_relationships_errors = self.Meta().model.required_related_objects_errors(
            output_for="api", initial_data=relationships_data
        )
        if required_relationships_errors:
            raise ValidationError({"relationships": required_relationships_errors})
        instance = super().create(validated_data)
        if relationships_data:
            try:
                self._save_relationships(instance, relationships_data)
            except DjangoValidationError as error:
                raise ValidationError(str(error)) from error
        return instance

    def update(self, instance, validated_data):
        relationships_key_specified = "relationships" in self.context["request"].data
        relationships_data = validated_data.pop("relationships", {})
        required_relationships_errors = self.Meta().model.required_related_objects_errors(
            output_for="api",
            initial_data=relationships_data,
            relationships_key_specified=relationships_key_specified,
            instance=instance,
        )
        if required_relationships_errors:
            raise ValidationError({"relationships": required_relationships_errors})

        instance = super().update(instance, validated_data)
        if relationships_data:
            try:
                self._save_relationships(instance, relationships_data)
            except DjangoValidationError as error:
                raise ValidationError(str(error)) from error
        return instance

    def _save_relationships(self, instance, relationships):
        """Create/update RelationshipAssociations corresponding to a model instance."""
        # relationships has already passed RelationshipsDataField.to_internal_value(), so we can skip some try/excepts
        logger.debug("_save_relationships: %s : %s", instance, relationships)
        for relationship, relationship_data in relationships.items():
            for other_side in ["source", "destination", "peer"]:
                if other_side not in relationship_data:
                    continue

                other_type = getattr(relationship, f"{other_side}_type")
                other_side_model = other_type.model_class()

                expected_objects_data = relationship_data[other_side]
                expected_objects = [
                    other_side_model.objects.get(**object_data) for object_data in expected_objects_data
                ]

                this_side = RelationshipSideChoices.OPPOSITE[other_side]

                if this_side != RelationshipSideChoices.SIDE_PEER:
                    existing_associations = relationship.relationship_associations.filter(
                        **{f"{this_side}_id": instance.pk}
                    )
                    existing_objects = [assoc.get_peer(instance) for assoc in existing_associations]
                else:
                    existing_associations_1 = relationship.relationship_associations.filter(source_id=instance.pk)
                    existing_objects_1 = [assoc.get_peer(instance) for assoc in existing_associations_1]
                    existing_associations_2 = relationship.relationship_associations.filter(destination_id=instance.pk)
                    existing_objects_2 = [assoc.get_peer(instance) for assoc in existing_associations_2]
                    existing_associations = list(existing_associations_1) + list(existing_associations_2)
                    existing_objects = existing_objects_1 + existing_objects_2

                add_objects = []
                remove_assocs = []

                for obj, assoc in zip(existing_objects, existing_associations):
                    if obj not in expected_objects:
                        remove_assocs.append(assoc)
                for obj in expected_objects:
                    if obj not in existing_objects:
                        add_objects.append(obj)

                for add_object in add_objects:
                    if "request" in self.context and not self.context["request"].user.has_perm(
                        "extras.add_relationshipassociation"
                    ):
                        raise PermissionDenied("This user does not have permission to create RelationshipAssociations.")
                    if other_side != RelationshipSideChoices.SIDE_SOURCE:
                        assoc = RelationshipAssociation(
                            relationship=relationship,
                            source_type=relationship.source_type,
                            source_id=instance.id,
                            destination_type=relationship.destination_type,
                            destination_id=add_object.id,
                        )
                    else:
                        assoc = RelationshipAssociation(
                            relationship=relationship,
                            source_type=relationship.source_type,
                            source_id=add_object.id,
                            destination_type=relationship.destination_type,
                            destination_id=instance.id,
                        )
                    assoc.validated_save()  # enforce relationship filter logic, etc.
                    logger.debug("Created %s", assoc)

                for remove_assoc in remove_assocs:
                    if "request" in self.context and not self.context["request"].user.has_perm(
                        "extras.delete_relationshipassociation"
                    ):
                        raise PermissionDenied("This user does not have permission to delete RelationshipAssociations.")
                    logger.debug("Deleting %s", remove_assoc)
                    remove_assoc.delete()

    def get_field_names(self, declared_fields, info):
        """Ensure that "relationships" is included as an opt-in field on root serializers."""
        fields = list(super().get_field_names(declared_fields, info))
        if not self.is_nested:
            # Only include relationships as opt-in.
            self.extend_field_names(fields, "relationships", opt_in_only=True)
        else:
            # As relationships are expensive, do not include them on nested serializers even if opted in.
            if "relationships" in fields:
                fields.remove("relationships")
        return fields

get_field_names(declared_fields, info)

Ensure that "relationships" is included as an opt-in field on root serializers.

Source code in nautobot/core/api/serializers.py
def get_field_names(self, declared_fields, info):
    """Ensure that "relationships" is included as an opt-in field on root serializers."""
    fields = list(super().get_field_names(declared_fields, info))
    if not self.is_nested:
        # Only include relationships as opt-in.
        self.extend_field_names(fields, "relationships", opt_in_only=True)
    else:
        # As relationships are expensive, do not include them on nested serializers even if opted in.
        if "relationships" in fields:
            fields.remove("relationships")
    return fields

nautobot.apps.api.SerializedPKRelatedField

Bases: PrimaryKeyRelatedField

Extends PrimaryKeyRelatedField to return a serialized object on read. This is useful for representing related objects in a ManyToManyField while still allowing a set of primary keys to be written.

Source code in nautobot/core/api/fields.py
class SerializedPKRelatedField(PrimaryKeyRelatedField):
    """
    Extends PrimaryKeyRelatedField to return a serialized object on read. This is useful for representing related
    objects in a ManyToManyField while still allowing a set of primary keys to be written.
    """

    def __init__(self, serializer, **kwargs):
        self.serializer = serializer
        self.pk_field = kwargs.pop("pk_field", None)
        super().__init__(**kwargs)

    def to_representation(self, value):
        return self.serializer(value, context={"request": self.context["request"]}).data

nautobot.apps.api.TaggedModelSerializerMixin

Bases: BaseModelSerializer

Source code in nautobot/extras/api/mixins.py
class TaggedModelSerializerMixin(BaseModelSerializer):
    def get_field_names(self, declared_fields, info):
        """Ensure that 'tags' field is always present except on nested serializers."""
        fields = list(super().get_field_names(declared_fields, info))
        if not self.is_nested:
            self.extend_field_names(fields, "tags")
        return fields

    def create(self, validated_data):
        tags = validated_data.pop("tags", None)
        instance = super().create(validated_data)

        if tags is not None:
            return self._save_tags(instance, tags)
        return instance

    def update(self, instance, validated_data):
        tags = validated_data.pop("tags", None)

        # Cache tags on instance for change logging
        instance._tags = tags or []

        instance = super().update(instance, validated_data)

        if tags is not None:
            return self._save_tags(instance, tags)
        return instance

    def _save_tags(self, instance, tags):
        if tags:
            instance.tags.set([t.name for t in tags])
        else:
            instance.tags.clear()

        return instance

    def to_representation(self, instance):
        data = super().to_representation(instance)
        if self._is_csv_request() and data.get("tags"):
            # Export tag names for CSV
            data["tags"] = list(instance.tags.values_list("name", flat=True))
        return data

get_field_names(declared_fields, info)

Ensure that 'tags' field is always present except on nested serializers.

Source code in nautobot/extras/api/mixins.py
def get_field_names(self, declared_fields, info):
    """Ensure that 'tags' field is always present except on nested serializers."""
    fields = list(super().get_field_names(declared_fields, info))
    if not self.is_nested:
        self.extend_field_names(fields, "tags")
    return fields

nautobot.apps.api.TimeZoneSerializerField

Bases: TimeZoneSerializerField_

Represents a time zone as a string.

Source code in nautobot/core/api/fields.py
@extend_schema_field(str)
class TimeZoneSerializerField(TimeZoneSerializerField_):
    """Represents a time zone as a string."""

nautobot.apps.api.TreeModelSerializerMixin

Bases: BaseModelSerializer

Add a tree_depth field to non-nested model serializers based on django-tree-queries.

Source code in nautobot/core/api/serializers.py
class TreeModelSerializerMixin(BaseModelSerializer):
    """Add a `tree_depth` field to non-nested model serializers based on django-tree-queries."""

    tree_depth = serializers.SerializerMethodField(read_only=True)

    @extend_schema_field(serializers.IntegerField(allow_null=True))
    def get_tree_depth(self, obj):
        """The `tree_depth` is not a database field, but an annotation automatically added by django-tree-queries."""
        return getattr(obj, "tree_depth", None)

    def get_field_names(self, declared_fields, info):
        """Ensure that "tree_depth" is included on root serializers only, as nested objects are not annotated."""
        fields = list(super().get_field_names(declared_fields, info))
        if self.is_nested and "tree_depth" in fields:
            fields.remove("tree_depth")
        return fields

get_field_names(declared_fields, info)

Ensure that "tree_depth" is included on root serializers only, as nested objects are not annotated.

Source code in nautobot/core/api/serializers.py
def get_field_names(self, declared_fields, info):
    """Ensure that "tree_depth" is included on root serializers only, as nested objects are not annotated."""
    fields = list(super().get_field_names(declared_fields, info))
    if self.is_nested and "tree_depth" in fields:
        fields.remove("tree_depth")
    return fields

get_tree_depth(obj)

The tree_depth is not a database field, but an annotation automatically added by django-tree-queries.

Source code in nautobot/core/api/serializers.py
@extend_schema_field(serializers.IntegerField(allow_null=True))
def get_tree_depth(self, obj):
    """The `tree_depth` is not a database field, but an annotation automatically added by django-tree-queries."""
    return getattr(obj, "tree_depth", None)

nautobot.apps.api.ValidatedModelSerializer

Bases: BaseModelSerializer

Extends the built-in ModelSerializer to enforce calling full_clean() on a copy of the associated instance during validation. (DRF does not do this by default; see https://github.com/encode/django-rest-framework/issues/3144)

Source code in nautobot/core/api/serializers.py
class ValidatedModelSerializer(BaseModelSerializer):
    """
    Extends the built-in ModelSerializer to enforce calling full_clean() on a copy of the associated instance during
    validation. (DRF does not do this by default; see https://github.com/encode/django-rest-framework/issues/3144)
    """

    def validate(self, data):
        # Remove custom fields data and tags (if any) prior to model validation
        attrs = data.copy()
        attrs.pop("custom_fields", None)
        attrs.pop("relationships", None)
        attrs.pop("tags", None)

        # Skip ManyToManyFields
        for field in self.Meta.model._meta.get_fields():
            if isinstance(field, models.ManyToManyField):
                attrs.pop(field.name, None)

        # Run clean() on an instance of the model
        if self.instance is None:
            instance = self.Meta.model(**attrs)
        else:
            instance = self.instance
            for k, v in attrs.items():
                setattr(instance, k, v)
        instance.full_clean()
        return data

nautobot.apps.api.WritableNestedSerializer

Bases: BaseModelSerializer

Returns a nested representation of an object on read, but accepts either the nested representation or the primary key value on write operations.

Note that subclasses will always have a read-only object_type field, which represents the content-type of this serializer's associated model (e.g. "dcim.device"). This is required as the OpenAPI schema, using the PolymorphicProxySerializer class defined below, relies upon this field as a way to identify to the client which of several possible nested serializers are in use for a given attribute.

Source code in nautobot/core/api/serializers.py
class WritableNestedSerializer(BaseModelSerializer):
    """
    Returns a nested representation of an object on read, but accepts either the nested representation or the
    primary key value on write operations.

    Note that subclasses will always have a read-only `object_type` field, which represents the content-type of this
    serializer's associated model (e.g. "dcim.device"). This is required as the OpenAPI schema, using the
    PolymorphicProxySerializer class defined below, relies upon this field as a way to identify to the client
    which of several possible nested serializers are in use for a given attribute.
    """

    def get_queryset(self):
        return self.Meta.model.objects

    def to_internal_value(self, data):
        if data is None:
            return None

        # Dictionary of related object attributes
        if isinstance(data, dict):
            params = dict_to_filter_params(data)

            # Make output from a WritableNestedSerializer "round-trip" capable by automatically stripping from the
            # data any serializer fields that do not correspond to a specific model field
            for field_name, field_instance in self.fields.items():
                if field_name in params and field_instance.source == "*":
                    logger.debug("Discarding non-database field %s", field_name)
                    del params[field_name]

            queryset = self.get_queryset()
            try:
                return queryset.get(**params)
            except ObjectDoesNotExist:
                raise ValidationError(f"Related object not found using the provided attributes: {params}")
            except MultipleObjectsReturned:
                raise ValidationError(f"Multiple objects match the provided attributes: {params}")
            except FieldError as e:
                raise ValidationError(e)

        queryset = self.get_queryset()
        pk = None

        if isinstance(self.Meta.model._meta.pk, models.AutoField):
            # PK is an int for this model. This is usually the User model
            try:
                pk = int(data)
            except (TypeError, ValueError):
                raise ValidationError(
                    "Related objects must be referenced by ID or by dictionary of attributes. Received an "
                    f"unrecognized value: {data}"
                )

        else:
            # We assume a type of UUIDField for all other models

            # PK of related object
            try:
                # Ensure the pk is a valid UUID
                pk = uuid.UUID(str(data))
            except (TypeError, ValueError):
                raise ValidationError(
                    "Related objects must be referenced by ID or by dictionary of attributes. Received an "
                    f"unrecognized value: {data}"
                )

        try:
            return queryset.get(pk=pk)
        except ObjectDoesNotExist:
            raise ValidationError(f"Related object not found using the provided ID: {pk}")

nautobot.apps.api.WritableSerializerMixin

WritableSerializerMixin provides the to_internal_value() function. The function provides the ability to write API requests that identify unique objects based on combinations of fields other than the primary key. e.g: "parent": { "location_type__parent": {"name": "Campus"}, "parent__name": "Campus-29" } vs "parent": "10dff139-7333-46b0-bef6-f6a5a7b5497c"

Source code in nautobot/core/api/mixins.py
class WritableSerializerMixin:
    """
    WritableSerializerMixin provides the to_internal_value() function.
    The function provides the ability to write API requests that identify unique objects based on
    combinations of fields other than the primary key.
    e.g:
    "parent": { "location_type__parent": {"name": "Campus"}, "parent__name": "Campus-29" }
    vs
    "parent": "10dff139-7333-46b0-bef6-f6a5a7b5497c"
    """

    def remove_non_filter_fields(self, filter_params):
        """
        Make output from a WritableSerializer "round-trip" capable by automatically stripping from the
        data any serializer fields that do not correspond to a specific model field
        """
        if hasattr(self, "fields"):
            for field_name, field_instance in self.fields.items():
                if field_name in filter_params and field_instance.source == "*":
                    logger.debug("Discarding non-filter field %s", field_name)
                    del filter_params[field_name]
        return filter_params

    def get_queryset_filter_params(self, data, queryset):
        """
        Data could be a dictionary and an int (for the User model) or a str that represents the primary key.
        If it is a dictionary, we return it after remove non-filter fields.
        If it is a primary key, we return a dictionary object formatted like this {"pk": pk}
        """

        if isinstance(data, dict):
            params = dict_to_filter_params(data)
            return self.remove_non_filter_fields(params)

        # Account for the fact that HyperlinkedIdentityFields might pass in URLs.
        if is_url(data):
            # Strip the trailing slash and split on slashes, taking the last value as the PK.
            data = data.rstrip("/").split("/")[-1]

        try:
            # The int case here is taking into account for the User model we inherit from django
            pk = int(data) if isinstance(queryset.model._meta.pk, AutoField) else uuid.UUID(str(data))
        except (TypeError, ValueError) as e:
            raise ValidationError(
                "Related objects must be referenced by ID or by dictionary of attributes. Received an "
                f"unrecognized value: {data}"
            ) from e
        return {"pk": pk}

    def get_object(self, data, queryset):
        """
        Retrieve an unique object based on a dictionary of data attributes and raise errors accordingly if the object is not found.
        """
        filter_params = self.get_queryset_filter_params(data=data, queryset=queryset)
        try:
            return queryset.get(**filter_params)
        except ObjectDoesNotExist as e:
            raise ValidationError(f"Related object not found using the provided attributes: {filter_params}") from e
        except MultipleObjectsReturned as e:
            raise ValidationError(f"Multiple objects match the provided attributes: {filter_params}") from e
        except FieldError as e:
            raise ValidationError(e) from e

    def to_internal_value(self, data):
        """
        Return an object or a list of objects based on a dictionary of data attributes or an UUID.
        """
        if data is None:
            return None
        if hasattr(self, "queryset"):
            queryset = self.queryset
        else:
            queryset = self.Meta.model.objects
        if isinstance(data, list):
            return [self.get_object(data=entry, queryset=queryset) for entry in data]
        return self.get_object(data=data, queryset=queryset)

get_object(data, queryset)

Retrieve an unique object based on a dictionary of data attributes and raise errors accordingly if the object is not found.

Source code in nautobot/core/api/mixins.py
def get_object(self, data, queryset):
    """
    Retrieve an unique object based on a dictionary of data attributes and raise errors accordingly if the object is not found.
    """
    filter_params = self.get_queryset_filter_params(data=data, queryset=queryset)
    try:
        return queryset.get(**filter_params)
    except ObjectDoesNotExist as e:
        raise ValidationError(f"Related object not found using the provided attributes: {filter_params}") from e
    except MultipleObjectsReturned as e:
        raise ValidationError(f"Multiple objects match the provided attributes: {filter_params}") from e
    except FieldError as e:
        raise ValidationError(e) from e

get_queryset_filter_params(data, queryset)

Data could be a dictionary and an int (for the User model) or a str that represents the primary key. If it is a dictionary, we return it after remove non-filter fields. If it is a primary key, we return a dictionary object formatted like this {"pk": pk}

Source code in nautobot/core/api/mixins.py
def get_queryset_filter_params(self, data, queryset):
    """
    Data could be a dictionary and an int (for the User model) or a str that represents the primary key.
    If it is a dictionary, we return it after remove non-filter fields.
    If it is a primary key, we return a dictionary object formatted like this {"pk": pk}
    """

    if isinstance(data, dict):
        params = dict_to_filter_params(data)
        return self.remove_non_filter_fields(params)

    # Account for the fact that HyperlinkedIdentityFields might pass in URLs.
    if is_url(data):
        # Strip the trailing slash and split on slashes, taking the last value as the PK.
        data = data.rstrip("/").split("/")[-1]

    try:
        # The int case here is taking into account for the User model we inherit from django
        pk = int(data) if isinstance(queryset.model._meta.pk, AutoField) else uuid.UUID(str(data))
    except (TypeError, ValueError) as e:
        raise ValidationError(
            "Related objects must be referenced by ID or by dictionary of attributes. Received an "
            f"unrecognized value: {data}"
        ) from e
    return {"pk": pk}

remove_non_filter_fields(filter_params)

Make output from a WritableSerializer "round-trip" capable by automatically stripping from the data any serializer fields that do not correspond to a specific model field

Source code in nautobot/core/api/mixins.py
def remove_non_filter_fields(self, filter_params):
    """
    Make output from a WritableSerializer "round-trip" capable by automatically stripping from the
    data any serializer fields that do not correspond to a specific model field
    """
    if hasattr(self, "fields"):
        for field_name, field_instance in self.fields.items():
            if field_name in filter_params and field_instance.source == "*":
                logger.debug("Discarding non-filter field %s", field_name)
                del filter_params[field_name]
    return filter_params

to_internal_value(data)

Return an object or a list of objects based on a dictionary of data attributes or an UUID.

Source code in nautobot/core/api/mixins.py
def to_internal_value(self, data):
    """
    Return an object or a list of objects based on a dictionary of data attributes or an UUID.
    """
    if data is None:
        return None
    if hasattr(self, "queryset"):
        queryset = self.queryset
    else:
        queryset = self.Meta.model.objects
    if isinstance(data, list):
        return [self.get_object(data=entry, queryset=queryset) for entry in data]
    return self.get_object(data=data, queryset=queryset)

nautobot.apps.api.dict_to_filter_params(d, prefix='')

Translate a dictionary of attributes to a nested set of parameters suitable for QuerySet filtering. For example:

{
    "name": "Foo",
    "rack": {
        "facility_id": "R101"
    }
}
Becomes

{ "name": "Foo", "rack__facility_id": "R101" }

And can be employed as filter parameters

Device.objects.filter(**dict_to_filter(attrs_dict))

Source code in nautobot/core/api/utils.py
def dict_to_filter_params(d, prefix=""):
    """
    Translate a dictionary of attributes to a nested set of parameters suitable for QuerySet filtering. For example:

        {
            "name": "Foo",
            "rack": {
                "facility_id": "R101"
            }
        }

    Becomes:

        {
            "name": "Foo",
            "rack__facility_id": "R101"
        }

    And can be employed as filter parameters:

        Device.objects.filter(**dict_to_filter(attrs_dict))
    """
    params = {}
    for key, val in d.items():
        k = prefix + key
        if isinstance(val, dict):
            params.update(dict_to_filter_params(val, k + "__"))
        else:
            params[k] = val
    return params

nautobot.apps.api.dynamic_import(name)

Dynamically import a class from an absolute path string

Source code in nautobot/core/api/utils.py
def dynamic_import(name):
    """
    Dynamically import a class from an absolute path string
    """
    components = name.split(".")
    mod = __import__(components[0])
    for comp in components[1:]:
        mod = getattr(mod, comp)
    return mod

nautobot.apps.api.get_api_version_serializer(serializer_choices, api_version)

Returns the serializer of an api_version

Parameters:

Name Type Description Default
serializer_choices tuple

list of SerializerVersions

required
api_version str

Request API version

required

Returns:

Type Description
Serializer

the serializer for the api_version if found in serializer_choices else None

Source code in nautobot/core/api/utils.py
def get_api_version_serializer(serializer_choices, api_version):
    """Returns the serializer of an api_version

    Args:
        serializer_choices (tuple): list of SerializerVersions
        api_version (str): Request API version

    Returns:
        (Serializer): the serializer for the api_version if found in serializer_choices else None
    """
    for versions, serializer in serializer_choices:
        if api_version in versions:
            return serializer
    return None

nautobot.apps.api.get_serializer_for_model(model, prefix='')

Dynamically resolve and return the appropriate serializer for a model.

Raises:

Type Description
SerializerNotFound

if the requested serializer cannot be located.

Source code in nautobot/core/api/utils.py
def get_serializer_for_model(model, prefix=""):
    """
    Dynamically resolve and return the appropriate serializer for a model.

    Raises:
        SerializerNotFound: if the requested serializer cannot be located.
    """
    app_label, model_name = model._meta.label.split(".")
    if app_label == "contenttypes" and model_name == "ContentType":
        app_path = "nautobot.extras"
    # Serializers for Django's auth models are in the users app
    elif app_label == "auth":
        app_path = "nautobot.users"
    else:
        app_path = apps.get_app_config(app_label).name
    serializer_name = f"{app_path}.api.serializers.{prefix}{model_name}Serializer"
    try:
        return dynamic_import(serializer_name)
    except AttributeError as exc:
        raise exceptions.SerializerNotFound(
            f"Could not determine serializer for {app_label}.{model_name} with prefix '{prefix}'"
        ) from exc

nautobot.apps.api.get_view_name(view, suffix=None)

Derive the view name from its associated model, if it has one. Fall back to DRF's built-in get_view_name.

Source code in nautobot/core/api/utils.py
def get_view_name(view, suffix=None):
    """
    Derive the view name from its associated model, if it has one. Fall back to DRF's built-in `get_view_name`.
    """
    if hasattr(view, "queryset"):
        # Determine the model name from the queryset.
        name = view.queryset.model._meta.verbose_name
        name = " ".join([w[0].upper() + w[1:] for w in name.split()])  # Capitalize each word

    else:
        # Replicate DRF's built-in behavior.
        name = view.__class__.__name__
        name = formatting.remove_trailing_string(name, "View")
        name = formatting.remove_trailing_string(name, "ViewSet")
        name = formatting.camelcase_to_spaces(name)

    if suffix:
        name += " " + suffix

    return name

nautobot.apps.api.is_api_request(request)

Return True of the request is being made via the REST API.

Source code in nautobot/core/api/utils.py
def is_api_request(request):
    """
    Return True of the request is being made via the REST API.
    """
    api_path = reverse("api-root")
    return request.path_info.startswith(api_path)

nautobot.apps.api.rest_api_server_error(request, *args, **kwargs)

Handle exceptions and return a useful error message for REST API requests.

Source code in nautobot/core/api/utils.py
def rest_api_server_error(request, *args, **kwargs):
    """
    Handle exceptions and return a useful error message for REST API requests.
    """
    type_, error, _traceback = sys.exc_info()
    data = {
        "error": str(error),
        "exception": type_.__name__,
        "nautobot_version": settings.VERSION,
        "python_version": platform.python_version(),
    }
    return JsonResponse(data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

nautobot.apps.api.versioned_serializer_selector(obj, serializer_choices, default_serializer)

Returns appropriate serializer class depending on request api_version, and swagger_fake_view

Parameters:

Name Type Description Default
obj ViewSet instance
required
serializer_choices tuple

Tuple of SerializerVersions

required
default_serializer Serializer

Default Serializer class

required
Source code in nautobot/core/api/utils.py
def versioned_serializer_selector(obj, serializer_choices, default_serializer):
    """Returns appropriate serializer class depending on request api_version, and swagger_fake_view

    Args:
        obj (ViewSet instance):
        serializer_choices (tuple): Tuple of SerializerVersions
        default_serializer (Serializer): Default Serializer class
    """
    if not getattr(obj, "swagger_fake_view", False) and hasattr(obj.request, "major_version"):
        api_version = f"{obj.request.major_version}.{obj.request.minor_version}"
        serializer = get_api_version_serializer(serializer_choices, api_version)
        if serializer is not None:
            return serializer
    return default_serializer