diff --git a/netbox/dcim/forms.py b/netbox/dcim/forms.py
index ce6fed468..bff89a3b5 100644
--- a/netbox/dcim/forms.py
+++ b/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,
- FilterChoiceField, FlexibleModelChoiceField, Livesearch, SelectWithDisabled, SmallTextarea, SlugField,
+ ChainedFieldsMixin, ChainedModelChoiceField, CommentField, ExpandableNameField, FilterChoiceField,
+ 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()
- device = FlexibleModelChoiceField(queryset=Device.objects.all(), to_field_name='name',
- error_messages={'invalid_choice': 'Device not found'})
- console_port = forms.CharField()
+ cs_port = forms.CharField(
+ help_text='Console server port name'
+ )
+ 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
- if self.cleaned_data.get('console_server'):
- try:
- cs_port = ConsoleServerPort.objects.get(device=self.cleaned_data['console_server'],
- name=self.cleaned_data['cs_port'])
- 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']))
+ def clean_console_port(self):
- # Validate console port
- if self.cleaned_data.get('device'):
- try:
- console_port = ConsolePort.objects.get(device=self.cleaned_data['device'],
- name=self.cleaned_data['console_port'])
- if console_port.cs_port:
- raise forms.ValidationError("Console port is already connected (to {} {})"
- .format(console_port.cs_port.device, console_port.cs_port))
- except ConsolePort.DoesNotExist:
- raise forms.ValidationError("Invalid console port ({} {})"
- .format(self.cleaned_data['device'], self.cleaned_data['console_port']))
+ console_port_name = self.cleaned_data.get('console_port')
+ if not self.cleaned_data.get('device') or not console_port_name:
+ return None
+ try:
+ # Retrieve console port by name
+ consoleport = ConsolePort.objects.get(
+ device=self.cleaned_data['device'], name=console_port_name
+ )
+ # 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 ({} {})".format(
+ self.cleaned_data['device'], console_port_name
+ ))
-class ConsoleConnectionImportForm(BootstrapMixin, BulkImportForm):
- csv = CSVDataField(csv_form=ConsoleConnectionCSVForm)
+ self.instance = consoleport
+ return consoleport
- def clean(self):
- records = self.cleaned_data.get('csv')
- if not records:
- return
+ def clean_cs_port(self):
- connection_list = []
+ cs_port_name = self.cleaned_data.get('cs_port')
+ if not self.cleaned_data.get('console_server') or not cs_port_name:
+ return None
- for i, record in enumerate(records, start=1):
- 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))
+ 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
+ ))
- self.cleaned_data['csv'] = connection_list
+ return cs_port
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()
- device = FlexibleModelChoiceField(queryset=Device.objects.all(), to_field_name='name',
- error_messages={'invalid_choice': 'Device not found'})
- power_port = forms.CharField()
+ power_outlet = forms.CharField(
+ help_text='Power outlet name'
+ )
+ 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
- if self.cleaned_data.get('pdu'):
- try:
- power_outlet = PowerOutlet.objects.get(device=self.cleaned_data['pdu'],
- name=self.cleaned_data['power_outlet'])
- 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']))
+ def clean_power_port(self):
- # Validate power port
- if self.cleaned_data.get('device'):
- try:
- power_port = PowerPort.objects.get(device=self.cleaned_data['device'],
- name=self.cleaned_data['power_port'])
- if power_port.power_outlet:
- raise forms.ValidationError("Power port is already connected (to {} {})"
- .format(power_port.power_outlet.device, power_port.power_outlet))
- except PowerPort.DoesNotExist:
- raise forms.ValidationError("Invalid power port ({} {})"
- .format(self.cleaned_data['device'], self.cleaned_data['power_port']))
+ power_port_name = self.cleaned_data.get('power_port')
+ if not self.cleaned_data.get('device') or not power_port_name:
+ return None
+ try:
+ # Retrieve power port by name
+ powerport = PowerPort.objects.get(
+ device=self.cleaned_data['device'], name=power_port_name
+ )
+ # 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 ({} {})".format(
+ self.cleaned_data['device'], power_port_name
+ ))
-class PowerConnectionImportForm(BootstrapMixin, BulkImportForm):
- csv = CSVDataField(csv_form=PowerConnectionCSVForm)
+ self.instance = powerport
+ return powerport
- def clean(self):
- records = self.cleaned_data.get('csv')
- if not records:
- return
+ def clean_power_outlet(self):
- connection_list = []
+ power_outlet_name = self.cleaned_data.get('power_outlet')
+ if not self.cleaned_data.get('pdu') or not power_outlet_name:
+ return None
- for i, record in enumerate(records, start=1):
- 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))
+ 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
+ ))
- self.cleaned_data['csv'] = connection_list
+ return power_outlet
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():
diff --git a/netbox/dcim/views.py b/netbox/dcim/views.py
index 58ea57580..0aade9cb3 100644
--- a/netbox/dcim/views.py
+++ b/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'
diff --git a/netbox/templates/dcim/console_connections_import.html b/netbox/templates/dcim/console_connections_import.html
deleted file mode 100644
index 3dc0f96a4..000000000
--- a/netbox/templates/dcim/console_connections_import.html
+++ /dev/null
@@ -1,45 +0,0 @@
-{% extends 'utilities/obj_import.html' %}
-
-{% block title %}Console Connections Import{% endblock %}
-
-{% block instructions %}
- CSV Format
-
-
-
- | Field |
- Description |
- Example |
-
-
-
-
- | Console server |
- Device name or {ID} |
- abc1-cs3 |
-
-
- | Console server port |
- Full CS port name |
- Port 35 |
-
-
- | Device |
- Device name or {ID} |
- abc1-switch7 |
-
-
- | Console Port |
- Console port name |
- Console |
-
-
- | Connection Status |
- "planned" or "connected" |
- planned |
-
-
-
- Example
- abc1-cs3,Port 35,abc1-switch7,Console,planned
-{% endblock %}
diff --git a/netbox/templates/dcim/interface_connections_import.html b/netbox/templates/dcim/interface_connections_import.html
deleted file mode 100644
index 4cfd82e58..000000000
--- a/netbox/templates/dcim/interface_connections_import.html
+++ /dev/null
@@ -1,45 +0,0 @@
-{% extends 'utilities/obj_import.html' %}
-
-{% block title %}Interface Connections Import{% endblock %}
-
-{% block instructions %}
- CSV Format
-
-
-
- | Field |
- Description |
- Example |
-
-
-
-
- | Device A |
- Device name or {ID} |
- abc1-core1 |
-
-
- | Interface A |
- Interface name |
- xe-0/0/6 |
-
-
- | Device B |
- Device name or {ID} |
- abc1-switch7 |
-
-
- | Interface B |
- Interface name |
- xe-0/0/0 |
-
-
- | Connection Status |
- "planned" or "connected" |
- planned |
-
-
-
- Example
- abc1-core1,xe-0/0/6,abc1-switch7,xe-0/0/0,planned
-{% endblock %}
diff --git a/netbox/templates/dcim/power_connections_import.html b/netbox/templates/dcim/power_connections_import.html
deleted file mode 100644
index bd60834da..000000000
--- a/netbox/templates/dcim/power_connections_import.html
+++ /dev/null
@@ -1,45 +0,0 @@
-{% extends 'utilities/obj_import.html' %}
-
-{% block title %}Power Connections Import{% endblock %}
-
-{% block instructions %}
- CSV Format
-
-
-
- | Field |
- Description |
- Example |
-
-
-
-
- | PDU |
- Device name or {ID} |
- abc1-pdu1 |
-
-
- | Power Outlet |
- Power outlet name |
- AC4 |
-
-
- | Device |
- Device name or {ID} |
- abc1-switch7 |
-
-
- | Power Port |
- Power port name |
- PSU0 |
-
-
- | Connection Status |
- "planned" or "connected" |
- connected |
-
-
-
- Example
- abc1-pdu1,AC4,abc1-switch7,PSU0,connected
-{% endblock %}
diff --git a/netbox/utilities/forms.py b/netbox/utilities/forms.py
index 6c07aad7f..56b7c24b7 100644
--- a/netbox/utilities/forms.py
+++ b/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
diff --git a/netbox/utilities/views.py b/netbox/utilities/views.py
index 766b0676d..529244c48 100644
--- a/netbox/utilities/views.py
+++ b/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).