mirror of
https://github.com/netbox-community/netbox.git
synced 2025-08-25 08:46:10 -06:00
Consolidate import forms
This commit is contained in:
jeremystretch
parent
d9306d827f
commit
ec053f550b
@ -18,9 +18,7 @@ from extras.models import ExportTemplate
|
||||
from extras.signals import clear_webhooks
|
||||
from utilities.error_handlers import handle_protectederror
|
||||
from utilities.exceptions import AbortRequest, AbortTransaction, PermissionsViolation
|
||||
from utilities.forms import (
|
||||
BulkRenameForm, ConfirmationForm, ImportForm, FileUploadImportForm, restrict_form_fields,
|
||||
)
|
||||
from utilities.forms import BulkRenameForm, ConfirmationForm, ImportForm, restrict_form_fields
|
||||
from utilities.forms.choices import ImportFormatChoices
|
||||
from utilities.htmx import is_htmx
|
||||
from utilities.permissions import get_permission_for_model
|
||||
@ -308,15 +306,6 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
"""
|
||||
return data
|
||||
|
||||
def _get_records(self, form, request):
|
||||
headers = form.cleaned_data['headers']
|
||||
if request.FILES:
|
||||
records = form.cleaned_data['data_file']
|
||||
else:
|
||||
records = form.cleaned_data['data']
|
||||
|
||||
return headers, records
|
||||
|
||||
def _create_object(self, request, model_form):
|
||||
|
||||
# Save the primary object
|
||||
@ -361,7 +350,7 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
|
||||
for i, record in enumerate(form.cleaned_data['data'], start=1):
|
||||
if form.cleaned_data['format'] == ImportFormatChoices.CSV:
|
||||
model_form = self.model_form(record, headers=form.cleaned_data['headers'])
|
||||
model_form = self.model_form(record, headers=form._csv_headers)
|
||||
else:
|
||||
model_form = self.model_form(record)
|
||||
# Assign default values for any fields which were not specified.
|
||||
@ -391,8 +380,10 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
|
||||
return new_objs
|
||||
|
||||
def _update_objects(self, form, request, headers, records):
|
||||
def _update_objects(self, form, request):
|
||||
updated_objs = []
|
||||
records = form.cleaned_data['data']
|
||||
headers = form._csv_headers
|
||||
|
||||
ids = [int(record["id"]) for record in records]
|
||||
qs = self.queryset.model.objects.filter(id__in=ids)
|
||||
@ -435,37 +426,21 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
# Request handlers
|
||||
#
|
||||
|
||||
def get_context(self, request, data_form, file_form, form=None):
|
||||
# small hack - need to return 'form' set to either the file or data form
|
||||
# as the bulk_import base view relies on it for error reporting.
|
||||
return {
|
||||
def get(self, request):
|
||||
form = ImportForm()
|
||||
|
||||
return render(request, self.template_name, {
|
||||
'model': self.model_form._meta.model,
|
||||
'data_form': data_form,
|
||||
'form': form,
|
||||
'file_form': file_form,
|
||||
'fields': self.model_form().fields,
|
||||
'return_url': self.get_return_url(request),
|
||||
**self.get_extra_context(request),
|
||||
}
|
||||
|
||||
def get(self, request):
|
||||
data_form = ImportForm(related=self.related_object_forms)
|
||||
file_form = FileUploadImportForm(related=self.related_object_forms)
|
||||
|
||||
return render(request, self.template_name, self.get_context(request, data_form, file_form))
|
||||
})
|
||||
|
||||
def post(self, request):
|
||||
logger = logging.getLogger('netbox.views.BulkImportView')
|
||||
|
||||
# Instantiate form based on action
|
||||
if 'file_submit' in request.POST:
|
||||
data_form = ImportForm(related=self.related_object_forms)
|
||||
file_form = FileUploadImportForm(request.POST, request.FILES, related=self.related_object_forms)
|
||||
form = file_form
|
||||
else: # data_submit
|
||||
data_form = ImportForm(request.POST, related=self.related_object_forms)
|
||||
file_form = FileUploadImportForm(related=self.related_object_forms)
|
||||
form = data_form
|
||||
form = ImportForm(request.POST, request.FILES)
|
||||
|
||||
if form.is_valid():
|
||||
logger.debug("Import form validation was successful")
|
||||
@ -474,9 +449,8 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
# Iterate through data and bind each record to a new model form instance.
|
||||
with transaction.atomic():
|
||||
if form.cleaned_data['format'] == 'csv':
|
||||
headers, records = self._get_records(form, request)
|
||||
if 'id' in headers:
|
||||
new_objs = self._update_objects(form, request, headers, records)
|
||||
if 'id' in form._csv_headers:
|
||||
new_objs = self._update_objects(form, request)
|
||||
else:
|
||||
new_objs = self._create_objects(form, request)
|
||||
else:
|
||||
@ -510,7 +484,13 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
else:
|
||||
logger.debug("Form validation failed")
|
||||
|
||||
return render(request, self.template_name, self.get_context(request, data_form, file_form, form))
|
||||
return render(request, self.template_name, {
|
||||
'model': self.model_form._meta.model,
|
||||
'form': form,
|
||||
'fields': self.model_form().fields,
|
||||
'return_url': self.get_return_url(request),
|
||||
**self.get_extra_context(request),
|
||||
})
|
||||
|
||||
|
||||
class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
|
||||
|
||||
@ -37,7 +37,8 @@ Context:
|
||||
<div class="col col-md-12 col-lg-10">
|
||||
<form action="" method="post" enctype="multipart/form-data" class="form">
|
||||
{% csrf_token %}
|
||||
{% render_form data_form %}
|
||||
{% render_field form.data %}
|
||||
{% render_field form.format %}
|
||||
<div class="form-group">
|
||||
<div class="col col-md-12 text-end">
|
||||
<button type="submit" name="data_submit" class="btn btn-primary">Submit</button>
|
||||
@ -57,7 +58,8 @@ Context:
|
||||
<div class="col col-md-12 col-lg-10">
|
||||
<form action="" method="post" enctype="multipart/form-data" class="form">
|
||||
{% csrf_token %}
|
||||
{% render_form file_form %}
|
||||
{% render_field form.data_file %}
|
||||
{% render_field form.format %}
|
||||
<div class="form-group">
|
||||
<div class="col col-md-12 text-end">
|
||||
<button type="submit" name="file_submit" class="btn btn-primary">Submit</button>
|
||||
|
||||
@ -15,13 +15,3 @@ class ImportFormatChoices(ChoiceSet):
|
||||
(JSON, 'JSON'),
|
||||
(YAML, 'YAML'),
|
||||
]
|
||||
|
||||
|
||||
class ImportFormatChoicesRelated(ChoiceSet):
|
||||
JSON = 'json'
|
||||
YAML = 'yaml'
|
||||
|
||||
CHOICES = [
|
||||
(JSON, 'JSON'),
|
||||
(YAML, 'YAML'),
|
||||
]
|
||||
|
||||
@ -5,9 +5,9 @@ from io import StringIO
|
||||
|
||||
import yaml
|
||||
from django import forms
|
||||
from utilities.forms.utils import parse_csv, validate_csv
|
||||
from utilities.forms.utils import parse_csv
|
||||
|
||||
from .choices import ImportFormatChoices, ImportFormatChoicesRelated
|
||||
from .choices import ImportFormatChoices
|
||||
from .widgets import APISelect, APISelectMultiple, ClearableFileInput, StaticSelect
|
||||
|
||||
__all__ = (
|
||||
@ -18,7 +18,6 @@ __all__ = (
|
||||
'CSVModelForm',
|
||||
'FilterForm',
|
||||
'ImportForm',
|
||||
'FileUploadImportForm',
|
||||
'ReturnURLForm',
|
||||
'TableConfigForm',
|
||||
)
|
||||
@ -135,9 +134,16 @@ class CSVModelForm(forms.ModelForm):
|
||||
self.fields[field].to_field_name = to_field
|
||||
|
||||
|
||||
class BaseImportForm(BootstrapMixin, forms.Form):
|
||||
data_field = 'data'
|
||||
|
||||
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
|
||||
)
|
||||
# TODO: Enable auto-detection of format
|
||||
format = forms.ChoiceField(
|
||||
choices=ImportFormatChoices,
|
||||
@ -145,74 +151,59 @@ class BaseImportForm(BootstrapMixin, forms.Form):
|
||||
widget=StaticSelect()
|
||||
)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
related = kwargs.pop("related", False)
|
||||
super().__init__(*args, **kwargs)
|
||||
if related:
|
||||
self.fields['format'].choices = ImportFormatChoicesRelated.CHOICES
|
||||
self.fields['format'].initial = ImportFormatChoicesRelated.YAML
|
||||
data_field = 'data'
|
||||
|
||||
def convert_data(self, data):
|
||||
def clean(self):
|
||||
super().clean()
|
||||
format = self.cleaned_data['format']
|
||||
stream = StringIO(data.strip())
|
||||
|
||||
# Process data
|
||||
if format == ImportFormatChoices.CSV:
|
||||
reader = csv.reader(stream)
|
||||
headers, records = parse_csv(reader)
|
||||
self.cleaned_data['data'] = records
|
||||
self.cleaned_data['headers'] = headers
|
||||
elif format == ImportFormatChoices.JSON:
|
||||
try:
|
||||
self.cleaned_data['data'] = json.loads(data)
|
||||
except json.decoder.JSONDecodeError as err:
|
||||
raise forms.ValidationError({
|
||||
self.data_field: f"Invalid JSON data: {err}"
|
||||
})
|
||||
elif format == ImportFormatChoices.YAML:
|
||||
try:
|
||||
self.cleaned_data['data'] = yaml.load_all(data, Loader=yaml.SafeLoader)
|
||||
except yaml.error.YAMLError as err:
|
||||
raise forms.ValidationError({
|
||||
self.data_field: f"Invalid YAML data: {err}"
|
||||
})
|
||||
# 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')
|
||||
else:
|
||||
data = self.cleaned_data['data']
|
||||
|
||||
# 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)
|
||||
|
||||
def _clean_csv(self, data):
|
||||
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):
|
||||
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 file format: {format}"
|
||||
self.data_field: f"Invalid JSON data: {err}"
|
||||
})
|
||||
|
||||
|
||||
class ImportForm(BaseImportForm):
|
||||
"""
|
||||
Generic form for creating an object from CSV/JSON/YAML data
|
||||
"""
|
||||
data = forms.CharField(
|
||||
widget=forms.Textarea(attrs={'class': 'font-monospace'}),
|
||||
help_text="Enter object data in CSV, JSON or YAML format."
|
||||
)
|
||||
|
||||
def clean(self):
|
||||
super().clean()
|
||||
data = self.cleaned_data.get('data')
|
||||
self.convert_data(data)
|
||||
|
||||
|
||||
class FileUploadImportForm(BaseImportForm):
|
||||
"""
|
||||
Generic form for creating an object from JSON/YAML data
|
||||
"""
|
||||
data_file = forms.FileField(
|
||||
label="data file",
|
||||
required=False
|
||||
)
|
||||
|
||||
data_field = 'data_file'
|
||||
|
||||
def clean(self):
|
||||
super().clean()
|
||||
file = self.files.get('data_file')
|
||||
data = file.read().decode('utf-8')
|
||||
self.convert_data(data)
|
||||
def _clean_yaml(self, data):
|
||||
try:
|
||||
return yaml.load_all(data, Loader=yaml.SafeLoader)
|
||||
except yaml.error.YAMLError as err:
|
||||
raise forms.ValidationError({
|
||||
self.data_field: f"Invalid YAML data: {err}"
|
||||
})
|
||||
|
||||
|
||||
class FilterForm(BootstrapMixin, forms.Form):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user