#22644 Add ObjectChange to PortMapping (#22645)

netbox/dcim/api/serializers_/base.py

@@ -3,6 +3,7 @@ from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
 
 from dcim.models import FrontPort, FrontPortTemplate, PortMapping, PortTemplateMapping, RearPort, RearPortTemplate
+from dcim.utils import reconcile_port_mappings
 from utilities.api import get_serializer_for_model
 
 __all__ = (
@@ -68,17 +69,25 @@ class PortSerializer(serializers.ModelSerializer):
             return PortTemplateMapping, 'rear_port'
         raise ValueError(f"Could not determine mapping details for {self.__class__}")
 
+    def _reconcile_mappings(self, instance, mappings):
+        mapping_model, fk_name = self._mapper
+        other_field = 'rear_port' if fk_name == 'front_port' else 'front_port'
+
+        # Normalize the opposite-port FK from a model instance to its id so the mappings can be
+        # reconciled by value.
+        desired = []
+        for attrs in mappings:
+            attrs = dict(attrs)
+            if other_field in attrs:
+                attrs[f'{other_field}_id'] = attrs.pop(other_field).pk
+            desired.append(attrs)
+
+        reconcile_port_mappings(mapping_model, parent_field=fk_name, parent=instance, desired=desired)
+
     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,
-            })
+        self._reconcile_mappings(instance, mappings)
 
         return instance
 
@@ -86,14 +95,9 @@ class PortSerializer(serializers.ModelSerializer):
         mappings = validated_data.pop('mappings', None)
         instance = super().update(instance, validated_data)
 
+        # Only reconcile when the client supplied rear_ports; a PATCH that omits it leaves the
+        # existing mappings untouched.
         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,
-                })
+            self._reconcile_mappings(instance, mappings)
 
         return instance

netbox/dcim/forms/mixins.py

@@ -1,12 +1,11 @@
 from django import forms
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist, ValidationError
-from django.db import connection
-from django.db.models.signals import post_save
 from django.utils.translation import gettext_lazy as _
 
 from dcim.constants import LOCATION_SCOPE_TYPES
-from dcim.models import PortMapping, PortTemplateMapping, Site
+from dcim.models import Site
+from dcim.utils import reconcile_port_mappings
 from utilities.forms import get_field_value
 from utilities.forms.fields import (
     ContentTypeChoiceField,
@@ -204,43 +203,24 @@ class FrontPortFormMixin(forms.Form):
     def _save_m2m(self):
         super()._save_m2m()
 
-        # TODO: Can this be made more efficient?
-        # Delete existing rear port mappings
-        self.port_mapping_model.objects.filter(front_port_id=self.instance.pk).delete()
-
-        # Create new rear port mappings
-        mappings = []
-        if self.port_mapping_model is PortTemplateMapping:
-            params = {
-                'device_type_id': self.instance.device_type_id,
-                'module_type_id': self.instance.module_type_id,
-            }
-        else:
-            params = {
-                'device_id': self.instance.device_id,
-            }
+        # Build the desired set of mappings from the submitted rear port pairs, assigning front port
+        # positions in order. reconcile_port_mappings() then writes only the difference, so re-saving
+        # a front port without changing its wiring produces no writes (and no changelog churn).
+        desired = []
         for i, rp_position in enumerate(self.cleaned_data['rear_ports'], start=1):
             rear_port_id, rear_port_position = rp_position.split(':')
-            mappings.append(
-                self.port_mapping_model(**{
-                    **params,
-                    'front_port_id': self.instance.pk,
-                    'front_port_position': i,
-                    'rear_port_id': rear_port_id,
-                    'rear_port_position': rear_port_position,
-                })
-            )
-        self.port_mapping_model.objects.bulk_create(mappings)
-        # Send post_save signals
-        for mapping in mappings:
-            post_save.send(
-                sender=PortMapping,
-                instance=mapping,
-                created=True,
-                raw=False,
-                using=connection,
-                update_fields=None
-            )
+            desired.append({
+                'front_port_position': i,
+                'rear_port_id': int(rear_port_id),
+                'rear_port_position': int(rear_port_position),
+            })
+
+        reconcile_port_mappings(
+            self.port_mapping_model,
+            parent_field='front_port',
+            parent=self.instance,
+            desired=desired,
+        )
 
     def _get_rear_port_choices(self, parent_filter, front_port):
         """

netbox/dcim/migrations/0239_add_portmapping_objectchange.py

@@ -0,0 +1,31 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("dcim", "0238_alter_cable__abs_length"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="portmapping",
+            name="created",
+            field=models.DateTimeField(auto_now_add=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="portmapping",
+            name="last_updated",
+            field=models.DateTimeField(auto_now=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="porttemplatemapping",
+            name="created",
+            field=models.DateTimeField(auto_now_add=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="porttemplatemapping",
+            name="last_updated",
+            field=models.DateTimeField(auto_now=True, null=True),
+        ),
+    ]

netbox/dcim/models/base.py

@@ -29,6 +29,8 @@ class PortMappingBase(models.Model):
         ),
     )
 
+    # Change-logged but private: no public API/URL, and the delete-time UPDATE cascade onto the
+    # parent ports stays suppressed (see #21390, #22270).
     _netbox_private = True
 
     class Meta:
@@ -44,6 +46,9 @@ class PortMappingBase(models.Model):
             ),
         )
 
+    def __str__(self):
+        return f'{self.front_port}:{self.front_port_position} to {self.rear_port}:{self.rear_port_position}'
+
     def clean(self):
         super().clean()
 

netbox/dcim/models/device_component_templates.py

@@ -11,6 +11,7 @@ from dcim.models.base import PortMappingBase
 from dcim.models.mixins import InterfaceValidationMixin
 from dcim.utils import get_module_bay_positions, resolve_module_placeholder
 from netbox.models import ChangeLoggedModel
+from netbox.models.features import ChangeLoggingMixin
 from utilities.fields import ColorField, NaturalOrderingField
 from utilities.mptt import TreeManager
 from utilities.ordering import naturalize_interface
@@ -538,7 +539,7 @@ class InterfaceTemplate(InterfaceValidationMixin, ModularComponentTemplateModel)
         }
 
 
-class PortTemplateMapping(PortMappingBase):
+class PortTemplateMapping(ChangeLoggingMixin, PortMappingBase):
     """
     Maps a FrontPortTemplate & position to a RearPortTemplate & position.
     """
@@ -567,6 +568,10 @@ class PortTemplateMapping(PortMappingBase):
         related_name='mappings',
     )
 
+    class Meta(PortMappingBase.Meta):
+        # Inherit the unique constraints from PortMappingBase.Meta.
+        pass
+
     def clean(self):
         super().clean()
 

netbox/dcim/models/device_components.py

@@ -15,6 +15,7 @@ from dcim.models.base import PortMappingBase
 from dcim.models.mixins import InterfaceValidationMixin
 from netbox.choices import ColorChoices
 from netbox.models import NetBoxModel, OrganizationalModel
+from netbox.models.features import ChangeLoggingMixin
 from netbox.models.mixins import OwnerMixin
 from utilities.fields import ColorField, NaturalOrderingField
 from utilities.mptt import TreeManager
@@ -1196,7 +1197,7 @@ class Interface(
 # Pass-through ports
 #
 
-class PortMapping(PortMappingBase):
+class PortMapping(ChangeLoggingMixin, PortMappingBase):
     """
     Maps a FrontPort & position to a RearPort & position.
     """
@@ -1216,6 +1217,10 @@ class PortMapping(PortMappingBase):
         related_name='mappings',
     )
 
+    class Meta(PortMappingBase.Meta):
+        # Inherit the unique constraints from PortMappingBase.Meta.
+        pass
+
     def clean(self):
         super().clean()
 

netbox/dcim/tests/test_port_mappings.py

@@ -0,0 +1,361 @@
+import uuid
+
+from django.contrib.contenttypes.models import ContentType
+from django.test import RequestFactory, TestCase, tag
+from django.urls import reverse
+from rest_framework import status
+
+from core.choices import ObjectChangeActionChoices
+from core.models import ObjectChange
+from dcim.choices import PortTypeChoices
+from dcim.models import (
+    Device,
+    DeviceRole,
+    DeviceType,
+    FrontPort,
+    FrontPortTemplate,
+    Manufacturer,
+    PortMapping,
+    PortTemplateMapping,
+    RearPort,
+    RearPortTemplate,
+    Site,
+)
+from dcim.utils import reconcile_port_mappings
+from netbox.context_managers import event_tracking
+from users.models import User
+from utilities.testing import APITestCase
+
+
+def _build_request(user):
+    request = RequestFactory().get('/')
+    request.id = uuid.uuid4()
+    request.user = user
+    return request
+
+
+class ReconcilePortMappingsTestCase(TestCase):
+    """
+    Exercise dcim.utils.reconcile_port_mappings and confirm that PortMapping now participates in
+    change logging (#22644): only the difference is written, so unchanged mappings keep their PK and
+    emit no ObjectChange.
+    """
+    @classmethod
+    def setUpTestData(cls):
+        cls.user = User.objects.create_user(username='testuser', password='pw')
+
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1')
+        role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1', color='ff0000')
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        cls.device = Device.objects.create(device_type=device_type, role=role, name='Device 1', site=site)
+
+        cls.front_port = FrontPort.objects.create(
+            device=cls.device, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=4
+        )
+        cls.rear_ports = [
+            RearPort.objects.create(
+                device=cls.device, name=f'Rear Port {i}', type=PortTypeChoices.TYPE_8P8C, positions=4
+            )
+            for i in range(1, 4)
+        ]
+
+    def _desired(self, *pairs):
+        # Each pair is (front_port_position, rear_port, rear_port_position).
+        return [
+            {'front_port_position': fpp, 'rear_port_id': rp.pk, 'rear_port_position': rpp}
+            for fpp, rp, rpp in pairs
+        ]
+
+    def _reconcile(self, desired):
+        request = _build_request(self.user)
+        with event_tracking(request):
+            reconcile_port_mappings(PortMapping, parent_field='front_port', parent=self.front_port, desired=desired)
+
+    def _mapping_changes(self, action=None):
+        changes = ObjectChange.objects.filter(changed_object_type=ContentType.objects.get_for_model(PortMapping))
+        if action is not None:
+            changes = changes.filter(action=action)
+        return changes
+
+    def _current_pks(self):
+        return set(PortMapping.objects.filter(front_port=self.front_port).values_list('pk', flat=True))
+
+    def test_create_records_objectchange(self):
+        self._reconcile(self._desired(
+            (1, self.rear_ports[0], 1),
+            (2, self.rear_ports[1], 1),
+        ))
+
+        self.assertEqual(PortMapping.objects.filter(front_port=self.front_port).count(), 2)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 2)
+
+    @tag('regression')  # Ref: #22644
+    def test_noop_resave_writes_nothing(self):
+        # Re-saving a front port without changing its wiring must not mint new PKs or emit changelog
+        # entries — this is what previously broke branch merges (colliding on the unique constraint
+        # when both sides replayed DELETE(old-pk) + CREATE(new-pk)).
+        desired = self._desired(
+            (1, self.rear_ports[0], 1),
+            (2, self.rear_ports[1], 1),
+        )
+        self._reconcile(desired)
+        original_pks = self._current_pks()
+        ObjectChange.objects.all().delete()
+
+        self._reconcile(desired)
+
+        self.assertEqual(self._mapping_changes().count(), 0)
+        self.assertEqual(self._current_pks(), original_pks)
+
+    def test_repointing_a_slot_deletes_and_creates(self):
+        self._reconcile(self._desired((1, self.rear_ports[0], 1)))
+        original_pk = self._current_pks().pop()
+        ObjectChange.objects.all().delete()
+
+        # Same front port position, different rear port: the slot is re-pointed.
+        self._reconcile(self._desired((1, self.rear_ports[1], 1)))
+
+        mappings = PortMapping.objects.filter(front_port=self.front_port)
+        self.assertEqual(mappings.count(), 1)
+        mapping = mappings.first()
+        self.assertEqual(mapping.rear_port_id, self.rear_ports[1].pk)
+        self.assertNotEqual(mapping.pk, original_pk)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).count(), 1)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 1)
+
+    def test_unchanged_rows_survive_alongside_changed_rows(self):
+        self._reconcile(self._desired(
+            (1, self.rear_ports[0], 1),
+            (2, self.rear_ports[1], 1),
+        ))
+        unchanged_pk = PortMapping.objects.get(front_port=self.front_port, front_port_position=1).pk
+        ObjectChange.objects.all().delete()
+
+        # Position 1 is untouched; position 2 is re-pointed to a third rear port.
+        self._reconcile(self._desired(
+            (1, self.rear_ports[0], 1),
+            (2, self.rear_ports[2], 1),
+        ))
+
+        self.assertEqual(PortMapping.objects.get(front_port=self.front_port, front_port_position=1).pk, unchanged_pk)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 1)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).count(), 1)
+
+    def test_removing_a_mapping_records_delete(self):
+        self._reconcile(self._desired(
+            (1, self.rear_ports[0], 1),
+            (2, self.rear_ports[1], 1),
+        ))
+        ObjectChange.objects.all().delete()
+
+        self._reconcile(self._desired((1, self.rear_ports[0], 1)))
+
+        self.assertEqual(PortMapping.objects.filter(front_port=self.front_port).count(), 1)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).count(), 1)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 0)
+
+    def test_swapping_positions_preserves_constraints(self):
+        # Two front port positions pointing at the same rear port's positions 1 and 2.
+        self._reconcile(self._desired(
+            (1, self.rear_ports[0], 1),
+            (2, self.rear_ports[0], 2),
+        ))
+
+        # Swap their rear port positions. Reconcile deletes both changed rows before recreating, so
+        # the transient state never violates the (rear_port, rear_port_position) unique constraint.
+        self._reconcile(self._desired(
+            (1, self.rear_ports[0], 2),
+            (2, self.rear_ports[0], 1),
+        ))
+
+        self.assertEqual(
+            PortMapping.objects.get(front_port=self.front_port, front_port_position=1).rear_port_position, 2
+        )
+        self.assertEqual(
+            PortMapping.objects.get(front_port=self.front_port, front_port_position=2).rear_port_position, 1
+        )
+
+    def test_direct_delete_records_objectchange(self):
+        self._reconcile(self._desired((1, self.rear_ports[0], 1)))
+        mapping = PortMapping.objects.get(front_port=self.front_port)
+        mapping_pk = mapping.pk
+        ObjectChange.objects.all().delete()
+
+        request = _build_request(self.user)
+        with event_tracking(request):
+            mapping.delete()
+
+        self.assertTrue(
+            self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).filter(changed_object_id=mapping_pk).exists()
+        )
+
+
+class ReconcilePortTemplateMappingsTestCase(TestCase):
+    """
+    Confirm reconcile_port_mappings and change logging behave identically for PortTemplateMapping.
+    """
+    @classmethod
+    def setUpTestData(cls):
+        cls.user = User.objects.create_user(username='testuser', password='pw')
+
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        cls.device_type = DeviceType.objects.create(
+            manufacturer=manufacturer, model='Device Type 1', slug='device-type-1'
+        )
+
+        cls.front_port = FrontPortTemplate.objects.create(
+            device_type=cls.device_type, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=4
+        )
+        cls.rear_ports = [
+            RearPortTemplate.objects.create(
+                device_type=cls.device_type, name=f'Rear Port {i}', type=PortTypeChoices.TYPE_8P8C, positions=4
+            )
+            for i in range(1, 3)
+        ]
+
+    def _desired(self, *pairs):
+        return [
+            {'front_port_position': fpp, 'rear_port_id': rp.pk, 'rear_port_position': rpp}
+            for fpp, rp, rpp in pairs
+        ]
+
+    def _reconcile(self, desired):
+        request = _build_request(self.user)
+        with event_tracking(request):
+            reconcile_port_mappings(
+                PortTemplateMapping,
+                parent_field='front_port',
+                parent=self.front_port,
+                desired=desired,
+            )
+
+    def _mapping_changes(self, action=None):
+        changes = ObjectChange.objects.filter(
+            changed_object_type=ContentType.objects.get_for_model(PortTemplateMapping)
+        )
+        if action is not None:
+            changes = changes.filter(action=action)
+        return changes
+
+    def test_create_records_objectchange(self):
+        self._reconcile(self._desired((1, self.rear_ports[0], 1)))
+
+        mapping = PortTemplateMapping.objects.get(front_port=self.front_port)
+        self.assertEqual(mapping.device_type_id, self.device_type.pk)
+        self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 1)
+
+    @tag('regression')  # Ref: #22644
+    def test_noop_resave_writes_nothing(self):
+        desired = self._desired((1, self.rear_ports[0], 1))
+        self._reconcile(desired)
+        original_pk = PortTemplateMapping.objects.get(front_port=self.front_port).pk
+        ObjectChange.objects.all().delete()
+
+        self._reconcile(desired)
+
+        self.assertEqual(self._mapping_changes().count(), 0)
+        self.assertEqual(PortTemplateMapping.objects.get(front_port=self.front_port).pk, original_pk)
+
+
+class PortMappingAPITestCase(APITestCase):
+    """
+    Exercise the reconcile behaviour through PortSerializer.create()/update() over the REST API,
+    covering the model-instance -> _id normalization in PortSerializer._reconcile_mappings.
+    """
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1')
+        role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1', color='ff0000')
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        cls.device = Device.objects.create(device_type=device_type, role=role, name='Device 1', site=site)
+
+        cls.rear_ports = [
+            RearPort.objects.create(
+                device=cls.device, name=f'Rear Port {i}', type=PortTypeChoices.TYPE_8P8C, positions=4
+            )
+            for i in range(1, 4)
+        ]
+
+        # An existing front port with two mappings, for the update/PATCH tests.
+        cls.front_port = FrontPort.objects.create(
+            device=cls.device, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=2
+        )
+        PortMapping.objects.bulk_create([
+            PortMapping(
+                device=cls.device, front_port=cls.front_port, front_port_position=1,
+                rear_port=cls.rear_ports[0], rear_port_position=1,
+            ),
+            PortMapping(
+                device=cls.device, front_port=cls.front_port, front_port_position=2,
+                rear_port=cls.rear_ports[1], rear_port_position=1,
+            ),
+        ])
+
+    def _mapping_pks(self, front_port):
+        return set(PortMapping.objects.filter(front_port=front_port).values_list('pk', flat=True))
+
+    def _mapping_creates(self):
+        return ObjectChange.objects.filter(
+            changed_object_type=ContentType.objects.get_for_model(PortMapping),
+            action=ObjectChangeActionChoices.ACTION_CREATE,
+        )
+
+    def test_create_records_mappings_and_changelog(self):
+        # Confirms PortSerializer.create() + _reconcile_mappings normalization (rear_port instance ->
+        # rear_port_id) writes the mappings and records their ObjectChanges.
+        self.add_permissions('dcim.add_frontport', 'dcim.view_frontport', 'dcim.view_rearport', 'dcim.view_device')
+        data = {
+            'device': self.device.pk,
+            'name': 'Front Port 2',
+            'type': PortTypeChoices.TYPE_8P8C,
+            'positions': 2,
+            'rear_ports': [
+                {'position': 1, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 1},
+                {'position': 2, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 2},
+            ],
+        }
+        response = self.client.post(reverse('dcim-api:frontport-list'), data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        front_port = FrontPort.objects.get(pk=response.data['id'])
+        mappings = PortMapping.objects.filter(front_port=front_port).order_by('front_port_position')
+        self.assertEqual(mappings.count(), 2)
+        self.assertEqual(mappings[0].rear_port_id, self.rear_ports[2].pk)
+        self.assertEqual(mappings[1].rear_port_position, 2)
+        self.assertEqual(
+            self._mapping_creates().filter(changed_object_id__in=mappings.values_list('pk', flat=True)).count(), 2
+        )
+
+    def test_update_omitting_rear_ports_preserves_mappings(self):
+        # A PATCH that omits rear_ports must leave the existing mappings untouched.
+        self.add_permissions('dcim.change_frontport', 'dcim.view_frontport')
+        url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.front_port.pk})
+        original_pks = self._mapping_pks(self.front_port)
+
+        response = self.client.patch(url, {'description': 'Updated'}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        self.assertEqual(self._mapping_pks(self.front_port), original_pks)
+
+    def test_update_rewires_mappings(self):
+        # Re-point both slots to new rear port pairs; reconcile deletes the old rows and creates the
+        # replacements.
+        self.add_permissions('dcim.change_frontport', 'dcim.view_frontport', 'dcim.view_rearport', 'dcim.view_device')
+        url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.front_port.pk})
+        original_pks = self._mapping_pks(self.front_port)
+
+        data = {
+            'rear_ports': [
+                {'position': 1, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 3},
+                {'position': 2, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 4},
+            ],
+        }
+        response = self.client.patch(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        mappings = PortMapping.objects.filter(front_port=self.front_port).order_by('front_port_position')
+        self.assertEqual([m.rear_port_id for m in mappings], [self.rear_ports[2].pk, self.rear_ports[2].pk])
+        self.assertEqual([m.rear_port_position for m in mappings], [3, 4])
+        self.assertTrue(self._mapping_pks(self.front_port).isdisjoint(original_pks))

netbox/dcim/utils.py

@@ -168,4 +168,53 @@ def create_port_mappings(device, device_or_module_type, module=None):
                 rear_port_position=template.rear_port_position,
             )
         )
+    # Bulk-created (no per-mapping ObjectChange) to match how every other component is instantiated.
     PortMapping.objects.bulk_create(mappings)
+
+
+def reconcile_port_mappings(mapping_model, parent_field, parent, desired):
+    """
+    Reconcile a parent port's mappings against `desired`, writing only the difference so unchanged
+    mappings keep their PK (and emit no changelog entry). Changed/removed rows are deleted before
+    replacements are created, all in one transaction, so position swaps don't trip the unique
+    constraint. Per-row create()/delete() let the change-logging signals fire naturally.
+
+    Args:
+        mapping_model: PortMapping or PortTemplateMapping.
+        parent_field: 'front_port' or 'rear_port' — the side being edited; its '<parent_field>_position'
+            is each mapping's stable identity within the set.
+        parent: the parent instance (FrontPort/RearPort or their templates).
+        desired: iterable of dicts of mapping field values EXCLUDING the parent FK, using '<field>_id'
+            for the opposite-port FK, e.g. {'front_port_position': 1, 'rear_port_id': 5,
+            'rear_port_position': 2}. save() derives device/device_type/module_type from the front port.
+    """
+    key_field = f'{parent_field}_position'
+    other_field = 'rear_port' if parent_field == 'front_port' else 'front_port'
+    value_fields = (f'{other_field}_id', f'{other_field}_position')
+
+    def target(source):
+        # The comparable "value" of a mapping: the opposite port and its position. Two mappings with
+        # the same parent-side position but a different target represent a re-pointing of that slot.
+        get = source.get if isinstance(source, dict) else lambda f: getattr(source, f)
+        return tuple(get(f) for f in value_fields)
+
+    desired_by_key = {d[key_field]: d for d in desired}
+
+    with transaction.atomic(using=router.db_for_write(mapping_model)):
+        # Lock the parent's existing mappings for the duration of the reconcile. Two requests editing
+        # the same port would otherwise read the same snapshot and race, the second colliding on a
+        # unique constraint when it recreates rows the first has already committed.
+        existing = {
+            getattr(m, key_field): m
+            for m in mapping_model.objects.filter(**{parent_field: parent}).select_for_update()
+        }
+
+        # Delete rows that no longer exist or whose target changed (before creating, to free the slots)
+        for key, mapping in existing.items():
+            if key not in desired_by_key or target(mapping) != target(desired_by_key[key]):
+                mapping.delete()
+
+        # Create rows that are new or whose target changed
+        for key, attrs in desired_by_key.items():
+            if key not in existing or target(existing[key]) != target(attrs):
+                mapping_model.objects.create(**{parent_field: parent, **attrs})