-
+
+
+ {# DataFile Form #}
+
{% if fields %}
diff --git a/netbox/utilities/choices.py b/netbox/utilities/choices.py
index ea4c5f951..5ad05b989 100644
--- a/netbox/utilities/choices.py
+++ b/netbox/utilities/choices.py
@@ -203,6 +203,18 @@ class ButtonColorChoices(ChoiceSet):
# Import Choices
#
+class ImportMethodChoices(ChoiceSet):
+ DIRECT = 'direct'
+ UPLOAD = 'upload'
+ DATA_FILE = 'datafile'
+
+ CHOICES = [
+ (DIRECT, 'Direct'),
+ (UPLOAD, 'Upload'),
+ (DATA_FILE, 'Data file'),
+ ]
+
+
class ImportFormatChoices(ChoiceSet):
AUTO = 'auto'
CSV = 'csv'
diff --git a/netbox/utilities/forms/bulk_import.py b/netbox/utilities/forms/bulk_import.py
new file mode 100644
index 000000000..369a87f3f
--- /dev/null
+++ b/netbox/utilities/forms/bulk_import.py
@@ -0,0 +1,141 @@
+import csv
+import json
+from io import StringIO
+
+import yaml
+from django import forms
+from django.utils.translation import gettext as _
+
+from extras.forms.mixins import SyncedDataMixin
+from utilities.choices import ImportFormatChoices
+from utilities.forms.utils import parse_csv
+from ..choices import ImportMethodChoices
+from .forms import BootstrapMixin
+
+
+class ImportForm(BootstrapMixin, SyncedDataMixin, forms.Form):
+ import_method = forms.ChoiceField(
+ choices=ImportMethodChoices
+ )
+ data = forms.CharField(
+ required=False,
+ widget=forms.Textarea(attrs={'class': 'font-monospace'}),
+ help_text=_("Enter object data in CSV, JSON or YAML format.")
+ )
+ upload_file = forms.FileField(
+ label="Data file",
+ required=False
+ )
+ format = forms.ChoiceField(
+ choices=ImportFormatChoices,
+ initial=ImportFormatChoices.AUTO
+ )
+
+ data_field = 'data'
+
+ def clean(self):
+ super().clean()
+
+ # Determine import method
+ import_method = self.cleaned_data['import_method']
+
+ # Determine whether we're reading from form data or an uploaded file
+ if self.cleaned_data['data'] and import_method != ImportMethodChoices.DIRECT:
+ raise forms.ValidationError("Form data must be empty when uploading/selecting a file.")
+ if import_method == ImportMethodChoices.UPLOAD:
+ self.upload_file = 'upload_file'
+ file = self.files.get('upload_file')
+ data = file.read().decode('utf-8-sig')
+ elif import_method == ImportMethodChoices.DATA_FILE:
+ data = self.cleaned_data['data_file'].data_as_string
+ else:
+ data = self.cleaned_data['data']
+
+ # Determine the data format
+ if self.cleaned_data['format'] == ImportFormatChoices.AUTO:
+ format = self._detect_format(data)
+ else:
+ format = self.cleaned_data['format']
+
+ # Process data according to the selected format
+ if format == ImportFormatChoices.CSV:
+ self.cleaned_data['data'] = self._clean_csv(data)
+ elif format == ImportFormatChoices.JSON:
+ self.cleaned_data['data'] = self._clean_json(data)
+ elif format == ImportFormatChoices.YAML:
+ self.cleaned_data['data'] = self._clean_yaml(data)
+ else:
+ raise forms.ValidationError(f"Unknown data format: {format}")
+
+ def _detect_format(self, data):
+ """
+ Attempt to automatically detect the format (CSV, JSON, or YAML) of the given data, or raise
+ a ValidationError.
+ """
+ try:
+ if data[0] in ('{', '['):
+ return ImportFormatChoices.JSON
+ if data.startswith('---') or data.startswith('- '):
+ return ImportFormatChoices.YAML
+ if ',' in data.split('\n', 1)[0]:
+ return ImportFormatChoices.CSV
+ except IndexError:
+ pass
+ raise forms.ValidationError({
+ 'format': _('Unable to detect data format. Please specify.')
+ })
+
+ def _clean_csv(self, data):
+ """
+ Clean CSV-formatted data. The first row will be treated as column headers.
+ """
+ stream = StringIO(data.strip())
+ reader = csv.reader(stream)
+ headers, records = parse_csv(reader)
+
+ # Set CSV headers for reference by the model form
+ self._csv_headers = headers
+
+ return records
+
+ def _clean_json(self, data):
+ """
+ Clean JSON-formatted data. If only a single object is defined, it will be encapsulated as a list.
+ """
+ try:
+ data = json.loads(data)
+ # Accommodate for users entering single objects
+ if type(data) is not list:
+ data = [data]
+ return data
+ except json.decoder.JSONDecodeError as err:
+ raise forms.ValidationError({
+ self.data_field: f"Invalid JSON data: {err}"
+ })
+
+ def _clean_yaml(self, data):
+ """
+ Clean YAML-formatted data. Data must be either
+ a) A single document comprising a list of dictionaries (each representing an object), or
+ b) Multiple documents, separated with the '---' token
+ """
+ records = []
+ try:
+ for data in yaml.load_all(data, Loader=yaml.SafeLoader):
+ if type(data) == list:
+ records.extend(data)
+ elif type(data) == dict:
+ records.append(data)
+ else:
+ raise forms.ValidationError({
+ self.data_field: _(
+ "Invalid YAML data. Data must be in the form of multiple documents, or a single document "
+ "comprising a list of dictionaries."
+ )
+ })
+ except yaml.error.YAMLError as err:
+ raise forms.ValidationError({
+ self.data_field: f"Invalid YAML data: {err}"
+ })
+
+ return records
diff --git a/netbox/utilities/forms/forms.py b/netbox/utilities/forms/forms.py
index bb9aaef0f..b06fb2e48 100644
--- a/netbox/utilities/forms/forms.py
+++ b/netbox/utilities/forms/forms.py
@@ -1,14 +1,8 @@
-import csv
-import json
import re
-from io import StringIO
-import yaml
from django import forms
from django.utils.translation import gettext as _
-from utilities.choices import ImportFormatChoices
-from utilities.forms.utils import parse_csv
from .widgets import APISelect, APISelectMultiple, ClearableFileInput
__all__ = (
@@ -18,7 +12,6 @@ __all__ = (
'ConfirmationForm',
'CSVModelForm',
'FilterForm',
- 'ImportForm',
'ReturnURLForm',
'TableConfigForm',
)
@@ -157,126 +150,6 @@ class CSVModelForm(forms.ModelForm):
del self.fields[field]
-class ImportForm(BootstrapMixin, forms.Form):
- data = forms.CharField(
- required=False,
- widget=forms.Textarea(attrs={'class': 'font-monospace'}),
- help_text=_("Enter object data in CSV, JSON or YAML format.")
- )
- data_file = forms.FileField(
- label="Data file",
- required=False
- )
- format = forms.ChoiceField(
- choices=ImportFormatChoices,
- initial=ImportFormatChoices.AUTO
- )
-
- data_field = 'data'
-
- def clean(self):
- super().clean()
-
- # Determine whether we're reading from form data or an uploaded file
- if self.cleaned_data['data'] and self.cleaned_data['data_file']:
- raise forms.ValidationError("Form data must be empty when uploading a file.")
- if 'data_file' in self.files:
- self.data_field = 'data_file'
- file = self.files.get('data_file')
- data = file.read().decode('utf-8-sig')
- else:
- data = self.cleaned_data['data']
-
- # Determine the data format
- if self.cleaned_data['format'] == ImportFormatChoices.AUTO:
- format = self._detect_format(data)
- else:
- format = self.cleaned_data['format']
-
- # Process data according to the selected format
- if format == ImportFormatChoices.CSV:
- self.cleaned_data['data'] = self._clean_csv(data)
- elif format == ImportFormatChoices.JSON:
- self.cleaned_data['data'] = self._clean_json(data)
- elif format == ImportFormatChoices.YAML:
- self.cleaned_data['data'] = self._clean_yaml(data)
- else:
- raise forms.ValidationError(f"Unknown data format: {format}")
-
- def _detect_format(self, data):
- """
- Attempt to automatically detect the format (CSV, JSON, or YAML) of the given data, or raise
- a ValidationError.
- """
- try:
- if data[0] in ('{', '['):
- return ImportFormatChoices.JSON
- if data.startswith('---') or data.startswith('- '):
- return ImportFormatChoices.YAML
- if ',' in data.split('\n', 1)[0]:
- return ImportFormatChoices.CSV
- except IndexError:
- pass
- raise forms.ValidationError({
- 'format': _('Unable to detect data format. Please specify.')
- })
-
- def _clean_csv(self, data):
- """
- Clean CSV-formatted data. The first row will be treated as column headers.
- """
- stream = StringIO(data.strip())
- reader = csv.reader(stream)
- headers, records = parse_csv(reader)
-
- # Set CSV headers for reference by the model form
- self._csv_headers = headers
-
- return records
-
- def _clean_json(self, data):
- """
- Clean JSON-formatted data. If only a single object is defined, it will be encapsulated as a list.
- """
- try:
- data = json.loads(data)
- # Accommodate for users entering single objects
- if type(data) is not list:
- data = [data]
- return data
- except json.decoder.JSONDecodeError as err:
- raise forms.ValidationError({
- self.data_field: f"Invalid JSON data: {err}"
- })
-
- def _clean_yaml(self, data):
- """
- Clean YAML-formatted data. Data must be either
- a) A single document comprising a list of dictionaries (each representing an object), or
- b) Multiple documents, separated with the '---' token
- """
- records = []
- try:
- for data in yaml.load_all(data, Loader=yaml.SafeLoader):
- if type(data) == list:
- records.extend(data)
- elif type(data) == dict:
- records.append(data)
- else:
- raise forms.ValidationError({
- self.data_field: _(
- "Invalid YAML data. Data must be in the form of multiple documents, or a single document "
- "comprising a list of dictionaries."
- )
- })
- except yaml.error.YAMLError as err:
- raise forms.ValidationError({
- self.data_field: f"Invalid YAML data: {err}"
- })
-
- return records
-
-
class FilterForm(BootstrapMixin, forms.Form):
"""
Base Form class for FilterSet forms.
diff --git a/netbox/utilities/tests/test_forms.py b/netbox/utilities/tests/test_forms.py
index 0f833d4a9..99c98a406 100644
--- a/netbox/utilities/tests/test_forms.py
+++ b/netbox/utilities/tests/test_forms.py
@@ -3,7 +3,7 @@ from django.test import TestCase
from ipam.forms import IPAddressImportForm
from utilities.choices import ImportFormatChoices
-from utilities.forms import ImportForm
+from utilities.forms.bulk_import import ImportForm
from utilities.forms.fields import CSVDataField
from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipaddress_pattern