Fixes #23644: Centralize dependent field queryset narrowing in CSV import forms

Import forms built filter parameters such as {f"site__{self.fields['site'].to_field_name}": value} by hand in many __init__() methods to limit one field's queryset by the value of another. Add CSVModelForm.get_related_filter() to build these in one place, and filter_by_value() on CSVModelChoiceField and CSVModelMultipleChoiceField for looking up objects by a single CSV value.

The header loop in CSVModelForm.__init__() now names the field name and field separately for readability.

netbox/dcim/forms/bulk_import.py

@@ -197,7 +197,7 @@ class LocationImportForm(NestedGroupModelImportForm):
 
         if data:
             # Limit location queryset by assigned site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)
 
 
@@ -369,7 +369,7 @@ class RackImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit location queryset by assigned site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
 
     def clean(self):
@@ -432,13 +432,13 @@ class RackReservationImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit location queryset by assigned site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
 
             # Limit rack queryset by assigned site and group
             params = {
-                f"site__{self.fields['site'].to_field_name}": data.get('site'),
-                f"location__{self.fields['location'].to_field_name}": data.get('location'),
+                **self.get_related_filter('site', 'site', data.get('site')),
+                **self.get_related_filter('location', 'location', data.get('location')),
             }
             self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)
 
@@ -691,7 +691,7 @@ class BaseDeviceImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit device type queryset by manufacturer
-            params = {f"manufacturer__{self.fields['manufacturer'].to_field_name}": data.get('manufacturer')}
+            params = self.get_related_filter('manufacturer', 'manufacturer', data.get('manufacturer'))
             self.fields['device_type'].queryset = self.fields['device_type'].queryset.filter(**params)
 
 
@@ -770,29 +770,25 @@ class DeviceImportForm(BaseDeviceImportForm):
         if data:
 
             # Limit location queryset by assigned site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
             self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)
 
             # Limit rack queryset by assigned site and location
-            params = {
-                f"site__{self.fields['site'].to_field_name}": data.get('site'),
-            }
+            params = self.get_related_filter('site', 'site', data.get('site'))
             if location := data.get('location'):
-                params.update({
-                    f"location__{self.fields['location'].to_field_name}": location,
-                })
+                params.update(self.get_related_filter('location', 'location', location))
             self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)
 
             # Limit platform queryset by manufacturer
-            params = {f"manufacturer__{self.fields['manufacturer'].to_field_name}": data.get('manufacturer')}
+            params = self.get_related_filter('manufacturer', 'manufacturer', data.get('manufacturer'))
             self.fields['platform'].queryset = self.fields['platform'].queryset.filter(
                 Q(**params) | Q(manufacturer=None)
             )
 
             # Limit device bay queryset by parent device
             if parent := data.get('parent'):
-                params = {f"device__{self.fields['parent'].to_field_name}": parent}
+                params = self.get_related_filter('device', 'parent', parent)
                 self.fields['device_bay'].queryset = self.fields['device_bay'].queryset.filter(**params)
 
     def clean(self):
@@ -855,7 +851,7 @@ class ModuleImportForm(ModuleCommonForm, PrimaryModelImportForm):
 
         if data:
             # Limit module_bay queryset by assigned device
-            params = {f"device__{self.fields['device'].to_field_name}": data.get('device')}
+            params = self.get_related_filter('device', 'device', data.get('device'))
             self.fields['module_bay'].queryset = self.fields['module_bay'].queryset.filter(**params)
 
     def clean_replicate_components(self):
@@ -1262,9 +1258,7 @@ class InterfaceImportForm(OwnerCSVMixin, NetBoxModelImportForm):
         if data:
             # Limit choices for parent, bridge, and LAG interfaces to the assigned device
             if device := data.get('device'):
-                params = {
-                    f"device__{self.fields['device'].to_field_name}": device
-                }
+                params = self.get_related_filter('device', 'device', device)
                 self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)
                 self.fields['bridge'].queryset = self.fields['bridge'].queryset.filter(**params)
                 self.fields['lag'].queryset = self.fields['lag'].queryset.filter(**params)
@@ -1272,7 +1266,7 @@ class InterfaceImportForm(OwnerCSVMixin, NetBoxModelImportForm):
 
             # Limit choices for VLANs to the assigned VLAN group
             if vlan_group := data.get('vlan_group'):
-                params = {f"group__{self.fields['vlan_group'].to_field_name}": vlan_group}
+                params = self.get_related_filter('group', 'vlan_group', vlan_group)
                 self.fields['untagged_vlan'].queryset = self.fields['untagged_vlan'].queryset.filter(**params)
                 self.fields['tagged_vlans'].queryset = self.fields['tagged_vlans'].queryset.filter(**params)
                 self.fields['qinq_svlan'].queryset = self.fields['qinq_svlan'].queryset.filter(**params)
@@ -1656,13 +1650,13 @@ class MACAddressImportForm(PrimaryModelImportForm):
             # Limit interface queryset by assigned device
             if data.get('device'):
                 self.fields['interface'].queryset = Interface.objects.filter(
-                    **{f"device__{self.fields['device'].to_field_name}": data['device']}
+                    **self.get_related_filter('device', 'device', data['device'])
                 )
 
             # Limit interface queryset by assigned device
             elif data.get('virtual_machine'):
                 self.fields['interface'].queryset = VMInterface.objects.filter(
-                    **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
+                    **self.get_related_filter('virtual_machine', 'virtual_machine', data['virtual_machine'])
                 )
 
     def clean(self):
@@ -1875,7 +1869,7 @@ class CableImportForm(PrimaryModelImportForm):
         if data:
             # Limit choices for side_a_device to the assigned side_a_site
             if side_a_site := data.get('side_a_site'):
-                side_a_parent_params = {f'site__{self.fields['side_a_site'].to_field_name}': side_a_site}
+                side_a_parent_params = self.get_related_filter('site', 'side_a_site', side_a_site)
                 self.fields['side_a_device'].queryset = self.fields['side_a_device'].queryset.filter(
                     **side_a_parent_params
                 )
@@ -1885,7 +1879,7 @@ class CableImportForm(PrimaryModelImportForm):
 
             # Limit choices for side_b_device to the assigned side_b_site
             if side_b_site := data.get('side_b_site'):
-                side_b_parent_params = {f'site__{self.fields['side_b_site'].to_field_name}': side_b_site}
+                side_b_parent_params = self.get_related_filter('site', 'side_b_site', side_b_site)
                 self.fields['side_b_device'].queryset = self.fields['side_b_device'].queryset.filter(
                     **side_b_parent_params
                 )
@@ -1958,12 +1952,11 @@ class CableImportForm(PrimaryModelImportForm):
             # The parent field has already raised its own validation error
             return None
         field = self.fields[field_name]
-        to_field_name = field.to_field_name or 'pk'
 
         parents = []
         for value in self._split_side_values(self.data.get(field_name)):
             try:
-                parents.append(field.queryset.get(**{to_field_name: value}))
+                parents.append(field.filter_by_value(value).get())
             except ObjectDoesNotExist:
                 self.add_error(field_name, _("Object not found: {value}").format(value=value))
                 return None
@@ -2199,7 +2192,7 @@ class PowerPanelImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit group queryset by assigned site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
 
 
@@ -2270,17 +2263,17 @@ class PowerFeedImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit power_panel queryset by site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['power_panel'].queryset = self.fields['power_panel'].queryset.filter(**params)
 
             # Limit location queryset by site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
 
             # Limit rack queryset by site and group
             params = {
-                f"site__{self.fields['site'].to_field_name}": data.get('site'),
-                f"location__{self.fields['location'].to_field_name}": data.get('location'),
+                **self.get_related_filter('site', 'site', data.get('site')),
+                **self.get_related_filter('location', 'location', data.get('location')),
             }
             self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)
 
@@ -2328,7 +2321,7 @@ class CoolingSourceImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit location queryset by assigned site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
 
 
@@ -2391,17 +2384,17 @@ class CoolingFeedImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit cooling_source queryset by site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['cooling_source'].queryset = self.fields['cooling_source'].queryset.filter(**params)
 
             # Limit location queryset by site
-            params = {f"site__{self.fields['site'].to_field_name}": data.get('site')}
+            params = self.get_related_filter('site', 'site', data.get('site'))
             self.fields['location'].queryset = self.fields['location'].queryset.filter(**params)
 
             # Limit rack queryset by site and group
             params = {
-                f"site__{self.fields['site'].to_field_name}": data.get('site'),
-                f"location__{self.fields['location'].to_field_name}": data.get('location'),
+                **self.get_related_filter('site', 'site', data.get('site')),
+                **self.get_related_filter('location', 'location', data.get('location')),
             }
             self.fields['rack'].queryset = self.fields['rack'].queryset.filter(**params)
 
@@ -2451,6 +2444,6 @@ class VirtualDeviceContextImportForm(PrimaryModelImportForm):
         if data:
 
             # Limit primary_ip4/ip6 querysets by assigned device
-            params = {f"interface__device__{self.fields['device'].to_field_name}": data.get('device')}
+            params = self.get_related_filter('interface__device', 'device', data.get('device'))
             self.fields['primary_ip4'].queryset = self.fields['primary_ip4'].queryset.filter(**params)
             self.fields['primary_ip6'].queryset = self.fields['primary_ip6'].queryset.filter(**params)

netbox/ipam/forms/bulk_import.py

@@ -241,14 +241,10 @@ class PrefixImportForm(ScopedImportForm, PrimaryModelImportForm):
         query = Q()
 
         if vlan_site:
-            query |= Q(**{
-                f"site__{self.fields['vlan_site'].to_field_name}": vlan_site
-            })
+            query |= Q(**self.get_related_filter('site', 'vlan_site', vlan_site))
 
         if vlan_group:
-            query &= Q(**{
-                f"group__{self.fields['vlan_group'].to_field_name}": vlan_group
-            })
+            query &= Q(**self.get_related_filter('group', 'vlan_group', vlan_group))
 
         queryset = self.fields['vlan'].queryset.filter(query)
         self.fields['vlan'].queryset = queryset
@@ -370,13 +366,13 @@ class IPAddressImportForm(PrimaryModelImportForm):
             # Limit interface queryset by assigned device
             if data.get('device'):
                 self.fields['interface'].queryset = Interface.objects.filter(
-                    **{f"device__{self.fields['device'].to_field_name}": data['device']}
+                    **self.get_related_filter('device', 'device', data['device'])
                 )
 
             # Limit interface queryset by assigned VM
             elif data.get('virtual_machine'):
                 self.fields['interface'].queryset = VMInterface.objects.filter(
-                    **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
+                    **self.get_related_filter('virtual_machine', 'virtual_machine', data['virtual_machine'])
                 )
 
     def clean_is_primary(self):

netbox/utilities/forms/fields/csv.py

@@ -69,7 +69,18 @@ class CSVTypedChoiceField(forms.TypedChoiceField):
     widget = CSVSelectWidget
 
 
-class CSVModelChoiceField(forms.ModelChoiceField):
+class CSVQuerysetMixin:
+    """
+    Provides lookup of objects in a CSV field's queryset by a single CSV value.
+    """
+    def filter_by_value(self, value):
+        """
+        Return the objects in the field's queryset matching a single CSV value.
+        """
+        return self.queryset.filter(**{self.to_field_name or 'pk': value})
+
+
+class CSVModelChoiceField(CSVQuerysetMixin, forms.ModelChoiceField):
     """
     Extends Django's `ModelChoiceField` to provide additional validation for CSV values.
     """
@@ -90,7 +101,7 @@ class CSVModelChoiceField(forms.ModelChoiceField):
             )
 
 
-class CSVModelMultipleChoiceField(forms.ModelMultipleChoiceField):
+class CSVModelMultipleChoiceField(CSVQuerysetMixin, forms.ModelMultipleChoiceField):
     """
     Extends Django's `ModelMultipleChoiceField` to support comma-separated values.
     """

netbox/utilities/forms/forms.py

@@ -126,9 +126,18 @@ class CSVModelForm(forms.ModelForm):
         super().__init__(*args, **kwargs)
 
         # Modify the model form to accommodate any customized to_field_name properties
-        for field, to_field in self.headers.items():
+        for field_name, to_field in self.headers.items():
             if to_field is not None:
-                self.fields[field].to_field_name = to_field
+                field = self.fields[field_name]
+                field.to_field_name = to_field
+
+    def get_related_filter(self, path, field_name, value):
+        """
+        Return filter parameters matching objects related through `path` to the object identified by `value` in the
+        field `field_name`. Used to limit one field's queryset by the value of another.
+        """
+        field = self.fields[field_name]
+        return {f'{path}__{field.to_field_name}': value}
 
     def clean(self):
         # Flag any invalid CSV headers

netbox/utilities/tests/test_forms.py

@@ -12,7 +12,7 @@ from netbox.choices import ImportFormatChoices
 from utilities.choices import Choice, ChoiceSet
 from utilities.forms.bulk_import import BulkImportForm
 from utilities.forms.fields import ChoiceField, MultipleChoiceField, TypedChoiceField
-from utilities.forms.fields.csv import CSVSelectWidget
+from utilities.forms.fields.csv import CSVModelChoiceField, CSVModelMultipleChoiceField, CSVSelectWidget
 from utilities.forms.fields.dynamic import (
     DynamicChoiceField,
     DynamicModelChoiceField,
@@ -20,7 +20,7 @@ from utilities.forms.fields.dynamic import (
     DynamicMultipleChoiceField,
 )
 from utilities.forms.fields.generic import GenericObjectChoiceField
-from utilities.forms.forms import BulkRenameForm
+from utilities.forms.forms import BulkRenameForm, CSVModelForm
 from utilities.forms.mixins import GenericObjectFormMixin
 from utilities.forms.rendering import FieldSet
 from utilities.forms.utils import (
@@ -567,6 +567,60 @@ class CSVSelectWidgetTestCase(TestCase):
         self.assertFalse(widget.value_omitted_from_data(data, {}, 'test_field'))
 
 
+class CSVModelFieldFilterByValueTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        Site.objects.bulk_create((
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+        ))
+
+    def test_filter_by_value_with_to_field_name(self):
+        site = Site.objects.get(slug='site-1')
+        for field_class in (CSVModelChoiceField, CSVModelMultipleChoiceField):
+            with self.subTest(field_class=field_class.__name__):
+                field = field_class(queryset=Site.objects.all(), to_field_name='slug')
+                self.assertQuerySetEqual(field.filter_by_value('site-1'), [site])
+
+    def test_filter_by_value_without_to_field_name(self):
+        site = Site.objects.get(slug='site-2')
+        for field_class in (CSVModelChoiceField, CSVModelMultipleChoiceField):
+            with self.subTest(field_class=field_class.__name__):
+                field = field_class(queryset=Site.objects.all())
+                self.assertQuerySetEqual(field.filter_by_value(site.pk), [site])
+
+    def test_filter_by_value_no_match(self):
+        field = CSVModelChoiceField(queryset=Site.objects.all(), to_field_name='name')
+        self.assertFalse(field.filter_by_value('Site 3').exists())
+
+
+class CSVModelFormGetRelatedFilterTestCase(TestCase):
+
+    class RackImportTestForm(CSVModelForm):
+        site = CSVModelChoiceField(queryset=Site.objects.all(), to_field_name='name')
+
+        class Meta:
+            model = Rack
+            fields = ('site',)
+
+    def test_get_related_filter(self):
+        form = self.RackImportTestForm()
+        self.assertEqual(form.get_related_filter('site', 'site', 'Site 1'), {'site__name': 'Site 1'})
+
+    def test_get_related_filter_nested_path(self):
+        form = self.RackImportTestForm()
+        self.assertEqual(
+            form.get_related_filter('rack__site', 'site', 'Site 1'),
+            {'rack__site__name': 'Site 1'}
+        )
+
+    def test_get_related_filter_with_csv_header(self):
+        # A CSV header such as "site.slug" overrides the field's to_field_name
+        form = self.RackImportTestForm(headers={'site': 'slug'})
+        self.assertEqual(form.get_related_filter('site', 'site', 'site-1'), {'site__slug': 'site-1'})
+
+
 class SelectMultipleWidgetTestCase(TestCase):
     """
     Validate filtering behavior of AvailableOptions and SelectedOptions widgets.

netbox/virtualization/forms/bulk_import.py

@@ -262,15 +262,13 @@ class VMInterfaceImportForm(OwnerCSVMixin, NetBoxModelImportForm):
         if data:
             # Limit interface choices for parent & bridge interfaces to the assigned VM
             if virtual_machine := data.get('virtual_machine'):
-                params = {
-                    f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": virtual_machine
-                }
+                params = self.get_related_filter('virtual_machine', 'virtual_machine', virtual_machine)
                 self.fields['parent'].queryset = self.fields['parent'].queryset.filter(**params)
                 self.fields['bridge'].queryset = self.fields['bridge'].queryset.filter(**params)
 
             # Limit choices for VLANs to the assigned VLAN group
             if vlan_group := data.get('vlan_group'):
-                params = {f"group__{self.fields['vlan_group'].to_field_name}": vlan_group}
+                params = self.get_related_filter('group', 'vlan_group', vlan_group)
                 self.fields['untagged_vlan'].queryset = self.fields['untagged_vlan'].queryset.filter(**params)
                 self.fields['tagged_vlans'].queryset = self.fields['tagged_vlans'].queryset.filter(**params)
                 self.fields['qinq_svlan'].queryset = self.fields['qinq_svlan'].queryset.filter(**params)

netbox/vpn/forms/bulk_import.py

@@ -123,11 +123,11 @@ class TunnelTerminationImportForm(NetBoxModelImportForm):
             # Limit termination queryset by assigned device/VM
             if data.get('device'):
                 self.fields['termination'].queryset = Interface.objects.filter(
-                    **{f"device__{self.fields['device'].to_field_name}": data['device']}
+                    **self.get_related_filter('device', 'device', data['device'])
                 )
             elif data.get('virtual_machine'):
                 self.fields['termination'].queryset = VMInterface.objects.filter(
-                    **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
+                    **self.get_related_filter('virtual_machine', 'virtual_machine', data['virtual_machine'])
                 )
 
     def save(self, *args, **kwargs):
@@ -325,11 +325,11 @@ class L2VPNTerminationImportForm(NetBoxModelImportForm):
             # Limit interface queryset by device or VM
             if data.get('device'):
                 self.fields['interface'].queryset = Interface.objects.filter(
-                    **{f"device__{self.fields['device'].to_field_name}": data['device']}
+                    **self.get_related_filter('device', 'device', data['device'])
                 )
             elif data.get('virtual_machine'):
                 self.fields['interface'].queryset = VMInterface.objects.filter(
-                    **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
+                    **self.get_related_filter('virtual_machine', 'virtual_machine', data['virtual_machine'])
                 )
 
     def clean(self):

netbox/wireless/forms/bulk_import.py

@@ -171,19 +171,19 @@ class WirelessLinkImportForm(PrimaryModelImportForm):
 
         if data:
             # Limit choices for interface_a to the assigned device_a
-            interface_a_params = {f'device__{self.fields["device_a"].to_field_name}': data.get('device_a')}
+            interface_a_params = self.get_related_filter('device', 'device_a', data.get('device_a'))
             # Limit choices for device_a to the assigned site_a
             if site_a := data.get('site_a'):
-                device_a_params = {f'site__{self.fields["site_a"].to_field_name}': site_a}
+                device_a_params = self.get_related_filter('site', 'site_a', site_a)
                 self.fields['device_a'].queryset = self.fields['device_a'].queryset.filter(**device_a_params)
-                interface_a_params.update({f'device__site__{self.fields["site_a"].to_field_name}': site_a})
+                interface_a_params.update(self.get_related_filter('device__site', 'site_a', site_a))
             self.fields['interface_a'].queryset = self.fields['interface_a'].queryset.filter(**interface_a_params)
 
             # Limit choices for interface_b to the assigned device_b
-            interface_b_params = {f'device__{self.fields["device_b"].to_field_name}': data.get('device_b')}
+            interface_b_params = self.get_related_filter('device', 'device_b', data.get('device_b'))
             # Limit choices for device_b to the assigned site_b
             if site_b := data.get('site_b'):
-                device_b_params = {f'site__{self.fields["site_b"].to_field_name}': site_b}
+                device_b_params = self.get_related_filter('site', 'site_b', site_b)
                 self.fields['device_b'].queryset = self.fields['device_b'].queryset.filter(**device_b_params)
-                interface_b_params.update({f'device__site__{self.fields["site_b"].to_field_name}': site_b})
+                interface_b_params.update(self.get_related_filter('device__site', 'site_b', site_b))
             self.fields['interface_b'].queryset = self.fields['interface_b'].queryset.filter(**interface_b_params)