Converted console/power import views to new scheme

netbox/dcim/forms.py

@@ -14,8 +14,8 @@ from tenancy.forms import TenancyForm
 from tenancy.models import Tenant
 from utilities.forms import (
     APISelect, add_blank_choice, ArrayFieldSelectMultiple, BootstrapMixin, BulkEditForm, BulkEditNullBooleanSelect,
-    BulkImportForm, ChainedFieldsMixin, ChainedModelChoiceField, CommentField, CSVDataField, ExpandableNameField,
+    ChainedFieldsMixin, ChainedModelChoiceField, CommentField, ExpandableNameField, FilterChoiceField,
-    FilterChoiceField, FlexibleModelChoiceField, Livesearch, SelectWithDisabled, SmallTextarea, SlugField,
+    FlexibleModelChoiceField, Livesearch, SelectWithDisabled, SmallTextarea, SlugField,
     FilterTreeNodeMultipleChoiceField,
 )
 from .formfields import MACAddressFormField
@@ -945,75 +945,81 @@ class ConsolePortCreateForm(DeviceComponentForm):
     name_pattern = ExpandableNameField(label='Name')
 
 
-class ConsoleConnectionCSVForm(forms.Form):
+class ConsoleConnectionCSVForm(forms.ModelForm):
     console_server = FlexibleModelChoiceField(
         queryset=Device.objects.filter(device_type__is_console_server=True),
         to_field_name='name',
+        help_text='Console server name or PK',
         error_messages={
             'invalid_choice': 'Console server not found',
         }
     )
-    cs_port = forms.CharField()
+    cs_port = forms.CharField(
-    device = FlexibleModelChoiceField(queryset=Device.objects.all(), to_field_name='name',
+        help_text='Console server port name'
-                                      error_messages={'invalid_choice': 'Device not found'})
+    )
-    console_port = forms.CharField()
+    device = FlexibleModelChoiceField(
+        queryset=Device.objects.all(),
+        to_field_name='name',
+        help_text='Device name or PK',
+        error_messages={
+            'invalid_choice': 'Device not found',
+        }
+    )
+    console_port = forms.CharField(
+        help_text='Console port name'
+    )
     status = forms.CharField(validators=[validate_connection_status])
 
-    def clean(self):
+    class Meta:
+        model = ConsolePort
+        fields = ['console_server', 'cs_port', 'device', 'console_port']
 
-        # Validate console server port
+    def clean_console_port(self):
-        if self.cleaned_data.get('console_server'):
+
-            try:
+        console_port_name = self.cleaned_data.get('console_port')
-                cs_port = ConsoleServerPort.objects.get(device=self.cleaned_data['console_server'],
+        if not self.cleaned_data.get('device') or not console_port_name:
-                                                        name=self.cleaned_data['cs_port'])
+            return None
-                if ConsolePort.objects.filter(cs_port=cs_port):
-                    raise forms.ValidationError("Console server port is already occupied (by {} {})"
-                                                .format(cs_port.connected_console.device, cs_port.connected_console))
-            except ConsoleServerPort.DoesNotExist:
-                raise forms.ValidationError("Invalid console server port ({} {})"
-                                            .format(self.cleaned_data['console_server'], self.cleaned_data['cs_port']))
 
-        # Validate console port
-        if self.cleaned_data.get('device'):
         try:
-                console_port = ConsolePort.objects.get(device=self.cleaned_data['device'],
+            # Retrieve console port by name
-                                                       name=self.cleaned_data['console_port'])
+            consoleport = ConsolePort.objects.get(
-                if console_port.cs_port:
+                device=self.cleaned_data['device'], name=console_port_name
-                    raise forms.ValidationError("Console port is already connected (to {} {})"
+            )
-                                                .format(console_port.cs_port.device, console_port.cs_port))
+            # Check if the console port is already connected
+            if consoleport.cs_port is not None:
+                raise forms.ValidationError("{} {} is already connected".format(
+                    self.cleaned_data['device'], console_port_name
+                ))
         except ConsolePort.DoesNotExist:
-                raise forms.ValidationError("Invalid console port ({} {})"
+            raise forms.ValidationError("Invalid console port ({} {})".format(
-                                            .format(self.cleaned_data['device'], self.cleaned_data['console_port']))
+                self.cleaned_data['device'], console_port_name
+            ))
 
+        self.instance = consoleport
+        return consoleport
 
-class ConsoleConnectionImportForm(BootstrapMixin, BulkImportForm):
+    def clean_cs_port(self):
-    csv = CSVDataField(csv_form=ConsoleConnectionCSVForm)
 
-    def clean(self):
+        cs_port_name = self.cleaned_data.get('cs_port')
-        records = self.cleaned_data.get('csv')
+        if not self.cleaned_data.get('console_server') or not cs_port_name:
-        if not records:
+            return None
-            return
 
-        connection_list = []
+        try:
+            # Retrieve console server port by name
+            cs_port = ConsoleServerPort.objects.get(
+                device=self.cleaned_data['console_server'], name=cs_port_name
+            )
+            # Check if the console server port is already connected
+            if ConsolePort.objects.filter(cs_port=cs_port).count():
+                raise forms.ValidationError("{} {} is already connected".format(
+                    self.cleaned_data['console_server'], cs_port_name
+                ))
+        except ConsoleServerPort.DoesNotExist:
+            raise forms.ValidationError("Invalid console server port ({} {})".format(
+                self.cleaned_data['console_server'], cs_port_name
+            ))
 
-        for i, record in enumerate(records, start=1):
+        return cs_port
-            form = self.fields['csv'].csv_form(data=record)
-            if form.is_valid():
-                console_port = ConsolePort.objects.get(device=form.cleaned_data['device'],
-                                                       name=form.cleaned_data['console_port'])
-                console_port.cs_port = ConsoleServerPort.objects.get(device=form.cleaned_data['console_server'],
-                                                                     name=form.cleaned_data['cs_port'])
-                if form.cleaned_data['status'] == 'planned':
-                    console_port.connection_status = CONNECTION_STATUS_PLANNED
-                else:
-                    console_port.connection_status = CONNECTION_STATUS_CONNECTED
-                connection_list.append(console_port)
-            else:
-                for field, errors in form.errors.items():
-                    for e in errors:
-                        self.add_error('csv', "Record {} {}: {}".format(i, field, e))
-
-        self.cleaned_data['csv'] = connection_list
 
 
 class ConsolePortConnectionForm(BootstrapMixin, ChainedFieldsMixin, forms.ModelForm):
@@ -1193,76 +1199,81 @@ class PowerPortCreateForm(DeviceComponentForm):
     name_pattern = ExpandableNameField(label='Name')
 
 
-class PowerConnectionCSVForm(forms.Form):
+class PowerConnectionCSVForm(forms.ModelForm):
     pdu = FlexibleModelChoiceField(
         queryset=Device.objects.filter(device_type__is_pdu=True),
         to_field_name='name',
+        help_text='PDU name or PK',
         error_messages={
             'invalid_choice': 'PDU not found.',
         }
     )
-    power_outlet = forms.CharField()
+    power_outlet = forms.CharField(
-    device = FlexibleModelChoiceField(queryset=Device.objects.all(), to_field_name='name',
+        help_text='Power outlet name'
-                                      error_messages={'invalid_choice': 'Device not found'})
+    )
-    power_port = forms.CharField()
+    device = FlexibleModelChoiceField(
+        queryset=Device.objects.all(),
+        to_field_name='name',
+        help_text='Device name or PK',
+        error_messages={
+            'invalid_choice': 'Device not found',
+        }
+    )
+    power_port = forms.CharField(
+        help_text='Power port name'
+    )
     status = forms.CharField(validators=[validate_connection_status])
 
-    def clean(self):
+    class Meta:
+        model = PowerPort
+        fields = ['pdu', 'power_outlet', 'device', 'power_port']
 
-        # Validate power outlet
+    def clean_power_port(self):
-        if self.cleaned_data.get('pdu'):
+
-            try:
+        power_port_name = self.cleaned_data.get('power_port')
-                power_outlet = PowerOutlet.objects.get(device=self.cleaned_data['pdu'],
+        if not self.cleaned_data.get('device') or not power_port_name:
-                                                       name=self.cleaned_data['power_outlet'])
+            return None
-                if PowerPort.objects.filter(power_outlet=power_outlet):
-                    raise forms.ValidationError("Power outlet is already occupied (by {} {})"
-                                                .format(power_outlet.connected_port.device,
-                                                        power_outlet.connected_port))
-            except PowerOutlet.DoesNotExist:
-                raise forms.ValidationError("Invalid PDU port ({} {})"
-                                            .format(self.cleaned_data['pdu'], self.cleaned_data['power_outlet']))
 
-        # Validate power port
-        if self.cleaned_data.get('device'):
         try:
-                power_port = PowerPort.objects.get(device=self.cleaned_data['device'],
+            # Retrieve power port by name
-                                                   name=self.cleaned_data['power_port'])
+            powerport = PowerPort.objects.get(
-                if power_port.power_outlet:
+                device=self.cleaned_data['device'], name=power_port_name
-                    raise forms.ValidationError("Power port is already connected (to {} {})"
+            )
-                                                .format(power_port.power_outlet.device, power_port.power_outlet))
+            # Check if the power port is already connected
+            if powerport.power_outlet is not None:
+                raise forms.ValidationError("{} {} is already connected".format(
+                    self.cleaned_data['device'], power_port_name
+                ))
         except PowerPort.DoesNotExist:
-                raise forms.ValidationError("Invalid power port ({} {})"
+            raise forms.ValidationError("Invalid power port ({} {})".format(
-                                            .format(self.cleaned_data['device'], self.cleaned_data['power_port']))
+                self.cleaned_data['device'], power_port_name
+            ))
 
+        self.instance = powerport
+        return powerport
 
-class PowerConnectionImportForm(BootstrapMixin, BulkImportForm):
+    def clean_power_outlet(self):
-    csv = CSVDataField(csv_form=PowerConnectionCSVForm)
 
-    def clean(self):
+        power_outlet_name = self.cleaned_data.get('power_outlet')
-        records = self.cleaned_data.get('csv')
+        if not self.cleaned_data.get('pdu') or not power_outlet_name:
-        if not records:
+            return None
-            return
 
-        connection_list = []
+        try:
+            # Retrieve power outlet by name
+            power_outlet = PowerOutlet.objects.get(
+                device=self.cleaned_data['pdu'], name=power_outlet_name
+            )
+            # Check if the power outlet is already connected
+            if PowerPort.objects.filter(power_outlet=power_outlet).count():
+                raise forms.ValidationError("{} {} is already connected".format(
+                    self.cleaned_data['pdu'], power_outlet_name
+                ))
+        except PowerOutlet.DoesNotExist:
+            raise forms.ValidationError("Invalid power outlet ({} {})".format(
+                self.cleaned_data['pdu'], power_outlet_name
+            ))
 
-        for i, record in enumerate(records, start=1):
+        return power_outlet
-            form = self.fields['csv'].csv_form(data=record)
-            if form.is_valid():
-                power_port = PowerPort.objects.get(device=form.cleaned_data['device'],
-                                                   name=form.cleaned_data['power_port'])
-                power_port.power_outlet = PowerOutlet.objects.get(device=form.cleaned_data['pdu'],
-                                                                  name=form.cleaned_data['power_outlet'])
-                if form.cleaned_data['status'] == 'planned':
-                    power_port.connection_status = CONNECTION_STATUS_PLANNED
-                else:
-                    power_port.connection_status = CONNECTION_STATUS_CONNECTED
-                connection_list.append(power_port)
-            else:
-                for field, errors in form.errors.items():
-                    for e in errors:
-                        self.add_error('csv', "Record {} {}: {}".format(i, field, e))
-
-        self.cleaned_data['csv'] = connection_list
 
 
 class PowerPortConnectionForm(BootstrapMixin, ChainedFieldsMixin, forms.ModelForm):
@@ -1629,7 +1640,7 @@ class InterfaceConnectionCSVForm(forms.ModelForm):
         try:
             # Retrieve interface by name
             interface = Interface.objects.get(
-                device=self.cleaned_data['device_a'], name=self.cleaned_data['interface_a']
+                device=self.cleaned_data['device_a'], name=interface_name
             )
             # Check for an existing connection to this interface
             if InterfaceConnection.objects.filter(Q(interface_a=interface) | Q(interface_b=interface)).count():
@@ -1652,7 +1663,7 @@ class InterfaceConnectionCSVForm(forms.ModelForm):
         try:
             # Retrieve interface by name
             interface = Interface.objects.get(
-                device=self.cleaned_data['device_b'], name=self.cleaned_data['interface_b']
+                device=self.cleaned_data['device_b'], name=interface_name
             )
             # Check for an existing connection to this interface
             if InterfaceConnection.objects.filter(Q(interface_a=interface) | Q(interface_b=interface)).count():

netbox/dcim/views.py

@@ -23,7 +23,7 @@ from extras.models import Graph, TopologyMap, GRAPH_TYPE_INTERFACE, GRAPH_TYPE_S
 from utilities.forms import ConfirmationForm
 from utilities.paginator import EnhancedPaginator
 from utilities.views import (
-    BulkDeleteView, BulkEditView, BulkImportView, BulkImportView2, ObjectDeleteView, ObjectEditView, ObjectListView,
+    BulkDeleteView, BulkEditView, BulkImportView2, ObjectDeleteView, ObjectEditView, ObjectListView,
 )
 from . import filters, forms, tables
 from .models import (
@@ -1012,11 +1012,10 @@ class ConsolePortBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
     parent_cls = Device
 
 
-class ConsoleConnectionsBulkImportView(PermissionRequiredMixin, BulkImportView):
+class ConsoleConnectionsBulkImportView(PermissionRequiredMixin, BulkImportView2):
     permission_required = 'dcim.change_consoleport'
-    form = forms.ConsoleConnectionImportForm
+    model_form = forms.ConsoleConnectionCSVForm
     table = tables.ConsoleConnectionTable
-    template_name = 'dcim/console_connections_import.html'
     default_return_url = 'dcim:console_connections_list'
 
 
@@ -1235,11 +1234,10 @@ class PowerPortBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
     parent_cls = Device
 
 
-class PowerConnectionsBulkImportView(PermissionRequiredMixin, BulkImportView):
+class PowerConnectionsBulkImportView(PermissionRequiredMixin, BulkImportView2):
     permission_required = 'dcim.change_powerport'
-    form = forms.PowerConnectionImportForm
+    model_form = forms.PowerConnectionCSVForm
     table = tables.PowerConnectionTable
-    template_name = 'dcim/power_connections_import.html'
     default_return_url = 'dcim:power_connections_list'
 
 

netbox/templates/dcim/console_connections_import.html

@@ -1,45 +0,0 @@
-{% extends 'utilities/obj_import.html' %}
-
-{% block title %}Console Connections Import{% endblock %}
-
-{% block instructions %}
-    <h4>CSV Format</h4>
-    <table class="table">
-        <thead>
-            <tr>
-                <th>Field</th>
-                <th>Description</th>
-                <th>Example</th>
-            </tr>
-        </thead>
-        <tbody>
-            <tr>
-                <td>Console server</td>
-                <td>Device name or {ID}</td>
-                <td>abc1-cs3</td>
-            </tr>
-            <tr>
-                <td>Console server port</td>
-                <td>Full CS port name</td>
-                <td>Port 35</td>
-            </tr>
-            <tr>
-                <td>Device</td>
-                <td>Device name or {ID}</td>
-                <td>abc1-switch7</td>
-            </tr>
-            <tr>
-                <td>Console Port</td>
-                <td>Console port name</td>
-                <td>Console</td>
-            </tr>
-            <tr>
-                <td>Connection Status</td>
-                <td>"planned" or "connected"</td>
-                <td>planned</td>
-            </tr>
-        </tbody>
-    </table>
-    <h4>Example</h4>
-    <pre>abc1-cs3,Port 35,abc1-switch7,Console,planned</pre>
-{% endblock %}

netbox/templates/dcim/interface_connections_import.html

@@ -1,45 +0,0 @@
-{% extends 'utilities/obj_import.html' %}
-
-{% block title %}Interface Connections Import{% endblock %}
-
-{% block instructions %}
-    <h4>CSV Format</h4>
-    <table class="table">
-        <thead>
-            <tr>
-                <th>Field</th>
-                <th>Description</th>
-                <th>Example</th>
-            </tr>
-        </thead>
-        <tbody>
-            <tr>
-                <td>Device A</td>
-                <td>Device name or {ID}</td>
-                <td>abc1-core1</td>
-            </tr>
-            <tr>
-                <td>Interface A</td>
-                <td>Interface name</td>
-                <td>xe-0/0/6</td>
-            </tr>
-            <tr>
-                <td>Device B</td>
-                <td>Device name or {ID}</td>
-                <td>abc1-switch7</td>
-            </tr>
-            <tr>
-                <td>Interface B</td>
-                <td>Interface name</td>
-                <td>xe-0/0/0</td>
-            </tr>
-            <tr>
-                <td>Connection Status</td>
-                <td>"planned" or "connected"</td>
-                <td>planned</td>
-            </tr>
-        </tbody>
-    </table>
-    <h4>Example</h4>
-    <pre>abc1-core1,xe-0/0/6,abc1-switch7,xe-0/0/0,planned</pre>
-{% endblock %}

netbox/templates/dcim/power_connections_import.html

@@ -1,45 +0,0 @@
-{% extends 'utilities/obj_import.html' %}
-
-{% block title %}Power Connections Import{% endblock %}
-
-{% block instructions %}
-    <h4>CSV Format</h4>
-    <table class="table">
-        <thead>
-            <tr>
-                <th>Field</th>
-                <th>Description</th>
-                <th>Example</th>
-            </tr>
-        </thead>
-        <tbody>
-            <tr>
-                <td>PDU</td>
-                <td>Device name or {ID}</td>
-                <td>abc1-pdu1</td>
-            </tr>
-            <tr>
-                <td>Power Outlet</td>
-                <td>Power outlet name</td>
-                <td>AC4</td>
-            </tr>
-            <tr>
-                <td>Device</td>
-                <td>Device name or {ID}</td>
-                <td>abc1-switch7</td>
-            </tr>
-            <tr>
-                <td>Power Port</td>
-                <td>Power port name</td>
-                <td>PSU0</td>
-            </tr>
-            <tr>
-                <td>Connection Status</td>
-                <td>"planned" or "connected"</td>
-                <td>connected</td>
-            </tr>
-        </tbody>
-    </table>
-    <h4>Example</h4>
-    <pre>abc1-pdu1,AC4,abc1-switch7,PSU0,connected</pre>
-{% endblock %}

netbox/utilities/forms.py

@@ -539,28 +539,3 @@ class BulkEditForm(forms.Form):
             self.nullable_fields = [field for field in self.Meta.nullable_fields]
         else:
             self.nullable_fields = []
-
-
-class BulkImportForm(forms.Form):
-
-    def clean(self):
-        fields, records = self.cleaned_data.get('csv').split('\n', 1)
-        if not records:
-            return
-
-        obj_list = []
-
-        for i, record in enumerate(records, start=1):
-            obj_form = self.fields['csv'].csv_form(data=record)
-            if obj_form.is_valid():
-                obj = obj_form.save(commit=False)
-                obj_list.append(obj)
-            else:
-                for field, errors in obj_form.errors.items():
-                    for e in errors:
-                        if field == '__all__':
-                            self.add_error('csv', "Record {}: {}".format(i, e))
-                        else:
-                            self.add_error('csv', "Record {} ({}): {}".format(i, field, e))
-
-        self.cleaned_data['csv'] = obj_list

netbox/utilities/views.py

@@ -369,61 +369,6 @@ class BulkAddView(View):
         })
 
 
-class BulkImportView(View):
-    """
-    Import objects in bulk (CSV format).
-
-    form: Form class
-    table: The django-tables2 Table used to render the list of imported objects
-    template_name: The name of the template
-    default_return_url: The name of the URL to use for the cancel button
-    """
-    form = None
-    table = None
-    template_name = None
-    default_return_url = None
-
-    def get(self, request):
-
-        return render(request, self.template_name, {
-            'form': self.form(),
-            'return_url': self.default_return_url,
-        })
-
-    def post(self, request):
-
-        form = self.form(request.POST)
-        if form.is_valid():
-            new_objs = []
-            try:
-                with transaction.atomic():
-                    for obj in form.cleaned_data['csv']:
-                        self.save_obj(obj)
-                        new_objs.append(obj)
-
-                obj_table = self.table(new_objs)
-                if new_objs:
-                    msg = 'Imported {} {}'.format(len(new_objs), new_objs[0]._meta.verbose_name_plural)
-                    messages.success(request, msg)
-                    UserAction.objects.log_import(request.user, ContentType.objects.get_for_model(new_objs[0]), msg)
-
-                return render(request, "import_success.html", {
-                    'table': obj_table,
-                    'return_url': self.default_return_url,
-                })
-
-            except IntegrityError as e:
-                form.add_error('csv', "Record {}: {}".format(len(new_objs) + 1, e.__cause__))
-
-        return render(request, self.template_name, {
-            'form': form,
-            'return_url': self.default_return_url,
-        })
-
-    def save_obj(self, obj):
-        obj.save()
-
-
 class BulkImportView2(View):
     """
     Import objects in bulk (CSV format).