Consolidate create() and update() logic into PortSerializer base class

netbox/dcim/api/serializers_/base.py

@@ -2,10 +2,12 @@ from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
 
+from dcim.models import FrontPort, FrontPortTemplate, PortMapping, PortTemplateMapping, RearPort, RearPortTemplate
 from utilities.api import get_serializer_for_model
 
 __all__ = (
     'ConnectedEndpointsSerializer',
+    'PortSerializer',
 )
 
 
@@ -35,3 +37,53 @@ class ConnectedEndpointsSerializer(serializers.ModelSerializer):
     @extend_schema_field(serializers.BooleanField)
     def get_connected_endpoints_reachable(self, obj):
         return obj._path and obj._path.is_complete and obj._path.is_active
+
+
+class PortSerializer(serializers.ModelSerializer):
+    """
+    Base serializer for front & rear port and port templates.
+    """
+    @property
+    def _mapper(self):
+        """
+        Return the model and ForeignKey field name used to track port mappings for this model.
+        """
+        if self.Meta.model is FrontPort:
+            return PortMapping, 'front_port'
+        if self.Meta.model is RearPort:
+            return PortMapping, 'rear_port'
+        if self.Meta.model is FrontPortTemplate:
+            return PortTemplateMapping, 'front_port'
+        if self.Meta.model is RearPortTemplate:
+            return PortTemplateMapping, 'rear_port'
+        raise ValueError(f"Could not determine mapping details for {self.__class__}")
+
+    def create(self, validated_data):
+        mappings = validated_data.pop('mappings', [])
+        instance = super().create(validated_data)
+
+        # Create port mappings
+        mapping_model, fk_name = self._mapper
+        for attrs in mappings:
+            mapping_model.objects.create(**{
+                fk_name: instance,
+                **attrs,
+            })
+
+        return instance
+
+    def update(self, instance, validated_data):
+        mappings = validated_data.pop('mappings', None)
+        instance = super().update(instance, validated_data)
+
+        if mappings is not None:
+            # Update port mappings
+            mapping_model, fk_name = self._mapper
+            mapping_model.objects.filter(**{fk_name: instance}).delete()
+            for attrs in mappings:
+                mapping_model.objects.create(**{
+                    fk_name: instance,
+                    **attrs,
+                })
+
+        return instance

netbox/dcim/api/serializers_/device_components.py

@@ -19,7 +19,7 @@ from wireless.api.serializers_.nested import NestedWirelessLinkSerializer
 from wireless.api.serializers_.wirelesslans import WirelessLANSerializer
 from wireless.choices import *
 from wireless.models import WirelessLAN
-from .base import ConnectedEndpointsSerializer
+from .base import ConnectedEndpointsSerializer, PortSerializer
 from .cables import CabledObjectSerializer
 from .devices import DeviceSerializer, MACAddressSerializer, ModuleSerializer, VirtualDeviceContextSerializer
 from .manufacturers import ManufacturerSerializer
@@ -307,7 +307,7 @@ class RearPortMappingSerializer(serializers.ModelSerializer):
         fields = ('position', 'front_port', 'front_port_position')
 
 
-class RearPortSerializer(NetBoxModelSerializer, CabledObjectSerializer):
+class RearPortSerializer(NetBoxModelSerializer, CabledObjectSerializer, PortSerializer):
     device = DeviceSerializer(nested=True)
     module = ModuleSerializer(
         nested=True,
@@ -331,28 +331,6 @@ class RearPortSerializer(NetBoxModelSerializer, CabledObjectSerializer):
         ]
         brief_fields = ('id', 'url', 'display', 'device', 'name', 'description', 'cable', '_occupied')
 
-    def create(self, validated_data):
-        mappings = validated_data.pop('mappings', [])
-        instance = super().create(validated_data)
-
-        # Create FrontPort mappings
-        for attrs in mappings:
-            PortMapping.objects.create(rear_port=instance, **attrs)
-
-        return instance
-
-    def update(self, instance, validated_data):
-        mappings = validated_data.pop('mappings', None)
-        instance = super().update(instance, validated_data)
-
-        if mappings is not None:
-            # Update FrontPort mappings
-            PortMapping.objects.filter(rear_port=instance).delete()
-            for attrs in mappings:
-                PortMapping.objects.create(rear_port=instance, **attrs)
-
-        return instance
-
 
 class FrontPortMappingSerializer(serializers.ModelSerializer):
     position = serializers.IntegerField(
@@ -367,7 +345,7 @@ class FrontPortMappingSerializer(serializers.ModelSerializer):
         fields = ('position', 'rear_port', 'rear_port_position')
 
 
-class FrontPortSerializer(NetBoxModelSerializer, CabledObjectSerializer):
+class FrontPortSerializer(NetBoxModelSerializer, CabledObjectSerializer, PortSerializer):
     device = DeviceSerializer(nested=True)
     module = ModuleSerializer(
         nested=True,
@@ -391,28 +369,6 @@ class FrontPortSerializer(NetBoxModelSerializer, CabledObjectSerializer):
         ]
         brief_fields = ('id', 'url', 'display', 'device', 'name', 'description', 'cable', '_occupied')
 
-    def create(self, validated_data):
-        mappings = validated_data.pop('mappings', [])
-        instance = super().create(validated_data)
-
-        # Create RearPort mappings
-        for attrs in mappings:
-            PortMapping.objects.create(front_port=instance, **attrs)
-
-        return instance
-
-    def update(self, instance, validated_data):
-        mappings = validated_data.pop('mappings', None)
-        instance = super().update(instance, validated_data)
-
-        if mappings is not None:
-            # Update RearPort mappings
-            PortMapping.objects.filter(front_port=instance).delete()
-            for attrs in mappings:
-                PortMapping.objects.create(front_port=instance, **attrs)
-
-        return instance
-
 
 class ModuleBaySerializer(NetBoxModelSerializer):
     device = DeviceSerializer(nested=True)

netbox/dcim/api/serializers_/devicetype_components.py

@@ -12,6 +12,7 @@ from netbox.api.fields import ChoiceField, ContentTypeField
 from netbox.api.gfk_fields import GFKSerializerField
 from netbox.api.serializers import ChangeLogMessageSerializer, ValidatedModelSerializer
 from wireless.choices import *
+from .base import PortSerializer
 from .devicetypes import DeviceTypeSerializer, ModuleTypeSerializer
 from .manufacturers import ManufacturerSerializer
 from .nested import NestedInterfaceTemplateSerializer
@@ -219,7 +220,7 @@ class RearPortTemplateMappingSerializer(serializers.ModelSerializer):
         fields = ('position', 'front_port', 'front_port_position')
 
 
-class RearPortTemplateSerializer(ComponentTemplateSerializer):
+class RearPortTemplateSerializer(ComponentTemplateSerializer, PortSerializer):
     device_type = DeviceTypeSerializer(
         required=False,
         nested=True,
@@ -247,28 +248,6 @@ class RearPortTemplateSerializer(ComponentTemplateSerializer):
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'description')
 
-    def create(self, validated_data):
-        mappings = validated_data.pop('mappings', [])
-        instance = super().create(validated_data)
-
-        # Create FrontPort mappings
-        for attrs in mappings:
-            PortTemplateMapping.objects.create(rear_port=instance, **attrs)
-
-        return instance
-
-    def update(self, instance, validated_data):
-        mappings = validated_data.pop('mappings', None)
-        instance = super().update(instance, validated_data)
-
-        if mappings is not None:
-            # Update FrontPort mappings
-            PortTemplateMapping.objects.filter(rear_port=instance).delete()
-            for attrs in mappings:
-                PortTemplateMapping.objects.create(rear_port=instance, **attrs)
-
-        return instance
-
 
 class FrontPortTemplateMappingSerializer(serializers.ModelSerializer):
     position = serializers.IntegerField(
@@ -283,7 +262,7 @@ class FrontPortTemplateMappingSerializer(serializers.ModelSerializer):
         fields = ('position', 'rear_port', 'rear_port_position')
 
 
-class FrontPortTemplateSerializer(ComponentTemplateSerializer):
+class FrontPortTemplateSerializer(ComponentTemplateSerializer, PortSerializer):
     device_type = DeviceTypeSerializer(
         nested=True,
         required=False,
@@ -311,28 +290,6 @@ class FrontPortTemplateSerializer(ComponentTemplateSerializer):
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'description')
 
-    def create(self, validated_data):
-        mappings = validated_data.pop('mappings', [])
-        instance = super().create(validated_data)
-
-        # Create RearPort mappings
-        for attrs in mappings:
-            PortTemplateMapping.objects.create(front_port=instance, **attrs)
-
-        return instance
-
-    def update(self, instance, validated_data):
-        mappings = validated_data.pop('mappings', None)
-        instance = super().update(instance, validated_data)
-
-        if mappings is not None:
-            # Update RearPort assignments
-            PortTemplateMapping.objects.filter(front_port=instance).delete()
-            for attrs in mappings:
-                PortTemplateMapping.objects.create(front_port=instance, **attrs)
-
-        return instance
-
 
 class ModuleBayTemplateSerializer(ComponentTemplateSerializer):
     device_type = DeviceTypeSerializer(