Coverage for netbox/views/generic/bulk_views.py: 12%
724 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import logging
2import re
3from collections import Counter
4from contextlib import contextmanager
5from copy import deepcopy
6from types import SimpleNamespace
8from django.conf import settings
9from django.contrib import messages
10from django.contrib.contenttypes.fields import GenericForeignKey, GenericRel
11from django.core.exceptions import FieldDoesNotExist, ImproperlyConfigured, ObjectDoesNotExist, ValidationError
12from django.db import IntegrityError, router, transaction
13from django.db.models import ManyToManyField, ProtectedError, RestrictedError
14from django.db.models.fields.reverse_related import ManyToManyRel
15from django.forms import ModelMultipleChoiceField, MultipleHiddenInput
16from django.http import HttpResponse
17from django.shortcuts import get_object_or_404, redirect, render
18from django.utils.http import content_disposition_header
19from django.utils.safestring import mark_safe
20from django.utils.translation import gettext as _
22from core.exceptions import JobFailed
23from core.models import ObjectType
24from core.signals import clear_events
25from extras.choices import CustomFieldUIEditableChoices
26from extras.models import CustomField, ExportTemplate
27from netbox.choices import ImportMethodChoices
28from netbox.forms.bulk_rename import NetBoxModelBulkRenameForm
29from netbox.models.features import ChangeLoggingMixin
30from netbox.object_actions import AddObject, BulkDelete, BulkEdit, BulkExport, BulkImport, BulkRename
31from utilities.error_handlers import handle_protectederror
32from utilities.exceptions import AbortRequest, AbortTransaction, PermissionsViolation
33from utilities.export import TableExport, stream_table_csv_response
34from utilities.forms import BulkDeleteForm, BulkRenameForm, restrict_form_fields
35from utilities.forms.bulk_import import BulkImportForm
36from utilities.htmx import htmx_partial
37from utilities.jobs import is_background_request, process_request_as_job
38from utilities.permissions import get_permission_for_model
39from utilities.query import reapply_model_ordering
40from utilities.querydict import normalize_querydict
41from utilities.request import safe_for_redirect
42from utilities.string import title
43from utilities.tables import get_table_configs
44from utilities.views import GetReturnURLMixin, get_action_url
46from .base import BaseMultiObjectView
47from .mixins import ActionsMixin, TableMixin
48from .utils import get_prerequisite_model
50__all__ = (
51 'BulkComponentCreateView',
52 'BulkCreateView',
53 'BulkDeleteView',
54 'BulkEditView',
55 'BulkImportView',
56 'BulkRenameView',
57 'ObjectListView',
58)
61# TODO: Remove in NetBox v5.0.
62# MPTT support is retained only for plugins whose tree models still derive from the
63# deprecated MPTT-backed bases. NetBox core uses netbox.models.ltree.LtreeModel, whose
64# database triggers maintain the tree on every write, so ltree (and non-tree) models
65# need no special bulk handling. These two helpers confine all MPTT-specific bulk
66# bookkeeping so it can be deleted in one place; for non-MPTT models they are no-ops.
67@contextmanager
68def _delay_mptt_updates(model):
69 """
70 Defer tree (lft/rght/tree_id) recomputation until the end of a bulk write for
71 legacy MPTT models. A no-op context manager for ltree and non-tree models.
72 """
73 from mptt.models import MPTTModel
75 if issubclass(model, MPTTModel):
76 with model.objects.delay_mptt_updates():
77 yield
78 else:
79 yield
82def _rebuild_mptt_tree(model):
83 """Rebuild the MPTT tree after a bulk edit for legacy MPTT models; else a no-op."""
84 from mptt.models import MPTTModel
86 if issubclass(model, MPTTModel):
87 model.objects.rebuild()
90class ObjectListView(BaseMultiObjectView, ActionsMixin, TableMixin):
91 """
92 Display multiple objects, all the same type, as a table.
94 Attributes:
95 filterset: A django-filter FilterSet that is applied to the queryset
96 filterset_form: The form class used to render filter options
97 actions: An iterable of ObjectAction subclasses (see ActionsMixin)
98 """
99 template_name = 'generic/object_list.html'
100 filterset = None
101 filterset_form = None
102 actions = (AddObject, BulkImport, BulkExport, BulkEdit, BulkRename, BulkDelete)
104 def get_required_permission(self):
105 return get_permission_for_model(self.queryset.model, 'view')
107 #
108 # Export methods
109 #
111 def export_yaml(self):
112 """
113 Export the queryset of objects as concatenated YAML documents.
114 """
115 yaml_data = [obj.to_yaml() for obj in self.queryset]
117 return '---\n'.join(yaml_data)
119 def export_table(self, table, columns=None, filename=None, delimiter=None):
120 """
121 Export all table data in CSV format.
123 Args:
124 table: The Table instance to export
125 columns: A list of specific columns to include. If None, all columns will be exported.
126 filename: The name of the file attachment sent to the client. If None, will be determined automatically
127 from the queryset model name.
128 delimiter: The character used to separate columns (a comma is used by default)
129 """
130 exclude_columns = {'pk', 'actions'}
131 all_columns = [col_name for col_name, _ in table.selected_columns + table.available_columns]
132 if columns:
133 exclude_columns.update({
134 col for col in all_columns if col not in columns
135 })
137 # Ensure related objects are prefetched for every column that will be exported, not just
138 # those currently visible in the configured table view.
139 table._apply_prefetching(columns=[c for c in all_columns if c not in exclude_columns])
141 filename = filename or f'netbox_{self.queryset.model._meta.verbose_name_plural}.csv'
143 if settings.STREAMING_EXPORTS:
144 return stream_table_csv_response(
145 table=table,
146 exclude_columns=exclude_columns,
147 filename=filename,
148 delimiter=delimiter,
149 )
151 exporter = TableExport(
152 export_format=TableExport.CSV,
153 table=table,
154 exclude_columns=exclude_columns,
155 delimiter=delimiter,
156 )
157 return exporter.response(filename=filename)
159 def export_template(self, template, request):
160 """
161 Render an ExportTemplate using the current queryset.
163 Args:
164 template: ExportTemplate instance
165 request: The current request
166 """
167 try:
168 return template.render_to_response(queryset=self.queryset)
169 except Exception as e:
170 messages.error(
171 request,
172 _("There was an error rendering the selected export template ({template}): {error}").format(
173 template=template.name,
174 error=e
175 )
176 )
177 # Strip the `export` param and redirect user to the filtered objects list
178 query_params = request.GET.copy()
179 query_params.pop('export')
180 redirect_url = f'{request.path}?{query_params.urlencode()}'
181 if safe_for_redirect(redirect_url):
182 return redirect(redirect_url)
183 return redirect(get_action_url(self.queryset.model, action='list'))
185 #
186 # Request handlers
187 #
189 def get_queryset(self, request):
190 qs = super().get_queryset(request)
191 return reapply_model_ordering(qs)
193 def get(self, request):
194 """
195 GET request handler.
197 Args:
198 request: The current request
199 """
200 model = self.queryset.model
201 object_type = ObjectType.objects.get_for_model(model)
203 if self.filterset:
204 self.queryset = self.filterset(request.GET, self.queryset, request=request).qs
206 # Determine the available actions
207 actions = self.get_permitted_actions(request.user)
208 has_table_actions = any(action.multi for action in actions)
210 if 'export' in request.GET:
212 # Export the current table view
213 if request.GET['export'] == 'table':
214 table = self.get_table(self.queryset, request, has_table_actions)
215 columns = [name for name, _ in table.selected_columns]
216 delimiter = request.user.config.get('csv_delimiter') if request.user.is_authenticated else None
217 return self.export_table(table, columns, delimiter=delimiter)
219 # Render an ExportTemplate
220 if request.GET['export']:
221 template = get_object_or_404(
222 ExportTemplate.objects.restrict(request.user, 'view'),
223 object_types=object_type,
224 name=request.GET['export'],
225 )
226 return self.export_template(template, request)
228 # Check for YAML export support on the model
229 if hasattr(model, 'to_yaml'):
230 response = HttpResponse(self.export_yaml(), content_type='text/yaml')
231 filename = 'netbox_{}.yaml'.format(self.queryset.model._meta.verbose_name_plural)
232 response['Content-Disposition'] = content_disposition_header(as_attachment=True, filename=filename)
233 return response
235 # Fall back to default table/YAML export
236 table = self.get_table(self.queryset, request, has_table_actions)
237 delimiter = request.user.config.get('csv_delimiter') if request.user.is_authenticated else None
238 return self.export_table(table, delimiter=delimiter)
240 # Render the objects table
241 table = self.get_table(self.queryset, request, has_table_actions)
243 # If this is an HTMX request, return only the rendered table HTML
244 if htmx_partial(request):
245 if request.GET.get('embedded', False):
246 table.embedded = True
247 # Hide selection checkboxes
248 if 'pk' in table.base_columns:
249 table.columns.hide('pk')
250 return render(request, 'htmx/table.html', {
251 'table': table,
252 'model': model,
253 'actions': actions,
254 })
256 context = {
257 'model': model,
258 'table': table,
259 'table_configs': get_table_configs(table, request.user),
260 'actions': actions,
261 'filter_form': self.filterset_form(request.GET) if self.filterset_form else None,
262 'prerequisite_model': get_prerequisite_model(self.queryset),
263 **self.get_extra_context(request),
264 }
266 return render(request, self.template_name, context)
269class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView):
270 """
271 Create new objects in bulk.
273 form: Form class which provides the `pattern` field
274 model_form: The ModelForm used to create individual objects
275 pattern_target: Name of the field to be evaluated as a pattern (if any)
276 """
277 form = None
278 model_form = None
279 pattern_target = ''
280 pattern_template_fields = ()
281 htmx_template_name = 'htmx/bulk_add_form.html'
283 def get_required_permission(self):
284 return get_permission_for_model(self.queryset.model, 'add')
286 def get_pattern_context(self, value):
287 """
288 Return a context mapping for substituting the generated pattern value into
289 model form fields.
291 By default, the field named by ``pattern_target`` is supported as a
292 placeholder, e.g. ``{vid}``.
293 """
294 if not self.pattern_target:
295 return {}
297 return {
298 self.pattern_target: str(value),
299 }
301 def render_pattern_template(self, template, value):
302 """
303 Replace pattern placeholders in a single form field value.
304 """
305 rendered = str(template)
307 for key, replacement in self.get_pattern_context(value).items():
308 rendered = rendered.replace(f'{{{key}}}', replacement)
310 return rendered
312 def apply_pattern_template_fields(self, data, value):
313 """
314 Apply the generated pattern value to any configured template fields.
315 """
316 for field_name in self.pattern_template_fields:
317 if field_name not in data:
318 continue
320 # QueryDict values may be multi-valued; preserve that behavior.
321 if hasattr(data, 'getlist') and hasattr(data, 'setlist'):
322 data.setlist(field_name, [
323 self.render_pattern_template(field_value, value)
324 for field_value in data.getlist(field_name)
325 ])
326 else:
327 data[field_name] = self.render_pattern_template(data[field_name], value)
329 return data
331 def get_model_form_data(self, form, request, value):
332 """
333 Return the submitted data to use when instantiating the model form for a
334 single generated pattern value.
335 """
336 data = request.POST.copy()
337 data[self.pattern_target] = value
339 return self.apply_pattern_template_fields(data, value)
341 def add_model_form_errors(self, form, model_form, value):
342 """
343 Copy validation errors from the generated object's model form back onto
344 the pattern form for display.
345 """
346 errors = model_form.errors.as_data()
348 if errors.get(self.pattern_target):
349 form.add_error('pattern', errors.pop(self.pattern_target))
351 for field_name, field_errors in errors.items():
352 if field_name == '__all__':
353 field_label = _('General')
354 elif field_name in model_form.fields:
355 field_label = model_form.fields[field_name].label
356 else:
357 field_label = field_name
359 for error in field_errors:
360 for message in error.messages:
361 form.add_error(
362 None,
363 _('{value}: {field}: {error}').format(
364 value=value,
365 field=field_label,
366 error=message,
367 )
368 )
370 def _create_objects(self, form, request):
371 new_objects = []
373 # Create objects from the expanded. Abort the transaction on the first validation error.
374 for value in form.cleaned_data['pattern']:
376 # Reinstantiate the model form each time to avoid overwriting the same instance. Use a mutable
377 # copy of the POST QueryDict so that we can update the target field value.
378 model_form = self.model_form(self.get_model_form_data(form, request, value))
380 # Validate each new object independently.
381 if model_form.is_valid():
382 model_form.instance._changelog_message = model_form.cleaned_data.get('changelog_message', '')
383 obj = model_form.save()
384 new_objects.append(obj)
385 else:
386 self.add_model_form_errors(form, model_form, value)
388 # Abort the transaction and break out of the loop.
389 raise AbortTransaction()
391 return new_objects
393 def _get_context(self, request, form, model_form):
394 model = self.queryset.model
395 return {
396 'object': None,
397 'obj_type': model._meta.verbose_name,
398 'obj_type_plural': model._meta.verbose_name_plural,
399 'form': form,
400 'model_form': model_form,
401 'return_url': self.get_return_url(request),
402 'add_url': get_action_url(model, 'add'),
403 **self.get_extra_context(request),
404 }
406 #
407 # Request handlers
408 #
410 def get(self, request):
411 # Set initial values for visible form fields from query args
412 initial = {}
413 for field in getattr(self.model_form._meta, 'fields', []):
414 if request.GET.get(field):
415 initial[field] = request.GET[field]
417 form = self.form()
418 model_form = self.model_form(initial=initial)
420 # HTMX partial: only re-render the model form fields
421 if htmx_partial(request):
422 return render(request, self.htmx_template_name, {
423 'model_form': model_form,
424 })
426 return render(request, self.template_name, self._get_context(request, form, model_form))
428 def post(self, request):
429 logger = logging.getLogger('netbox.views.BulkCreateView')
430 model = self.queryset.model
431 form = self.form(request.POST)
432 model_form = self.model_form(request.POST)
434 # HTMX partial: only re-render the model form fields
435 if htmx_partial(request):
436 return render(request, self.htmx_template_name, {
437 'model_form': model_form,
438 })
440 if form.is_valid():
441 logger.debug("Form validation was successful")
443 try:
444 with transaction.atomic(using=router.db_for_write(model)):
445 new_objs = self._create_objects(form, request)
447 # Enforce object-level permissions
448 if self.queryset.filter(pk__in=[obj.pk for obj in new_objs]).count() != len(new_objs):
449 raise PermissionsViolation
451 # If we make it to this point, validation has succeeded on all new objects.
452 msg = f"Added {len(new_objs)} {model._meta.verbose_name_plural}"
453 logger.info(msg)
454 messages.success(request, msg)
456 if '_addanother' in request.POST and safe_for_redirect(request.path):
457 return redirect(request.path)
458 return redirect(self.get_return_url(request))
460 except (AbortTransaction, IntegrityError):
461 clear_events.send(sender=self)
463 except (AbortRequest, PermissionsViolation) as e:
464 logger.debug(e.message)
465 form.add_error(None, e.message)
466 clear_events.send(sender=self)
468 else:
469 logger.debug("Form validation failed")
471 return render(request, self.template_name, self._get_context(request, form, model_form))
474class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
475 """
476 Import objects in bulk (CSV/JSON/YAML format).
478 Attributes:
479 model_form: The form used to create each imported object
480 """
481 template_name = 'generic/bulk_import.html'
482 model_form = None
483 related_object_forms = dict()
485 def get_required_permission(self):
486 return get_permission_for_model(self.queryset.model, 'add')
488 def prep_related_object_data(self, parent, data):
489 """
490 Hook to modify the data for related objects before it's passed to the related object form (for example, to
491 assign a parent object).
492 """
493 return data
495 def _get_form_fields(self):
496 form = self.model_form()
497 required_fields = {}
498 optional_fields = {}
500 # Return only visible fields, with required fields listed first
501 for field in form.visible_fields():
502 if field.is_hidden:
503 continue
504 if field.field.required:
505 required_fields[field.name] = field.field
506 else:
507 optional_fields[field.name] = field.field
509 return {**required_fields, **optional_fields}
511 def _compile_form_errors(self, errors, index, prefix=None):
512 error_messages = []
513 for field_name, errors in errors.items():
514 prefix = f'{prefix}.' if prefix else ''
515 if field_name == '__all__':
516 field_name = ''
517 for err in errors:
518 error_messages.append(f"Record {index} {prefix}{field_name}: {err}")
519 return error_messages
521 def _save_object(self, model_form, request, parent_idx):
522 _action = 'Updated' if model_form.instance.pk else 'Created'
524 # Save the primary object. Object-level permissions are enforced in aggregate by
525 # create_and_update_objects() once all records have been processed.
526 obj = self.save_object(model_form, request)
528 # Iterate through the related object forms (if any), validating and saving each instance.
529 for field_name, related_object_form in self.related_object_forms.items():
531 related_objects = model_form.data.get(field_name, list())
532 if not isinstance(related_objects, list):
533 raise ValidationError(
534 self._compile_form_errors(
535 {field_name: [_("Must be a list.")]},
536 index=parent_idx
537 )
538 )
540 related_obj_pks = []
541 for i, rel_obj_data in enumerate(related_objects, start=1):
542 if not isinstance(rel_obj_data, dict):
543 raise ValidationError(
544 self._compile_form_errors(
545 {f'{field_name}[{i}]': [_("Must be a dictionary.")]},
546 index=parent_idx,
547 )
548 )
550 rel_obj_data = self.prep_related_object_data(obj, rel_obj_data)
551 f = related_object_form(rel_obj_data)
553 for subfield_name, field in f.fields.items():
554 if subfield_name not in rel_obj_data and hasattr(field, 'initial'):
555 f.data[subfield_name] = field.initial
557 if f.is_valid():
558 related_obj = f.save()
559 related_obj_pks.append(related_obj.pk)
560 else:
561 # Replicate errors on the related object form to the import form for display and abort
562 raise ValidationError(
563 self._compile_form_errors(f.errors, index=parent_idx, prefix=f'{field_name}[{i}]')
564 )
566 # Enforce object-level permissions on related objects
567 model = related_object_form.Meta.model
568 if model.objects.filter(pk__in=related_obj_pks).count() != len(related_obj_pks):
569 raise ObjectDoesNotExist
571 if is_background_request(request):
572 request.job.logger.info(f'{_action} {obj}')
574 return obj
576 def save_object(self, object_form, request):
577 """
578 Provide a hook to modify the object immediately before saving it (e.g. to encrypt secret data).
580 Args:
581 object_form: The model form instance
582 request: The current request
583 """
584 return object_form.save()
586 def _process_import_records(self, form, request, records, prefetched_objects):
587 """
588 Process CSV import records and save objects.
589 """
590 saved_objects = []
592 for i, record in enumerate(records, start=1):
593 object_id = int(record.pop('id')) if record.get('id') else None
595 # Determine whether this object is being created or updated
596 if object_id:
597 try:
598 instance = prefetched_objects[object_id]
599 except KeyError:
600 raise ValidationError(
601 self._compile_form_errors(
602 {'id': [_("Object with ID {id} does not exist").format(id=object_id)]},
603 index=i
604 )
605 )
607 # Take a snapshot for change logging
608 if instance.pk and hasattr(instance, 'snapshot'):
609 instance.snapshot()
611 else:
612 instance = self.queryset.model()
614 # For newly created objects, apply any default values for custom fields
615 for cf in CustomField.objects.get_for_model(self.queryset.model):
616 if cf.ui_editable != CustomFieldUIEditableChoices.YES:
617 # Skip custom fields which are not editable via the UI
618 continue
619 field_name = f'cf_{cf.name}'
620 if field_name not in record:
621 record[field_name] = cf.default
623 # Record changelog message (if any)
624 instance._changelog_message = form.cleaned_data.get('changelog_message', '')
626 # Instantiate the model form for the object
627 model_form_kwargs = {
628 'data': record,
629 'instance': instance,
630 }
631 if hasattr(form, '_csv_headers'):
632 model_form_kwargs['headers'] = form._csv_headers # Add CSV headers
633 model_form = self.model_form(**model_form_kwargs)
635 # When updating, omit all form fields other than those specified in the record. (No
636 # fields are required when modifying an existing object.)
637 if object_id:
638 unused_fields = [f for f in model_form.fields if f not in record]
639 for field_name in unused_fields:
640 del model_form.fields[field_name]
642 restrict_form_fields(model_form, request.user)
644 if model_form.is_valid():
645 obj = self._save_object(model_form, request, i)
646 saved_objects.append(obj)
647 else:
648 # Raise model form errors
649 raise ValidationError(
650 self._compile_form_errors(model_form.errors, index=i)
651 )
653 return saved_objects
655 def create_and_update_objects(self, form, request):
656 records = list(form.cleaned_data['data'])
658 # Prefetch objects to be updated, if any
659 prefetch_ids = [int(record['id']) for record in records if record.get('id')]
661 # check for duplicate IDs
662 duplicate_pks = [pk for pk, count in Counter(prefetch_ids).items() if count > 1]
663 if duplicate_pks:
664 error_msg = _(
665 "Duplicate objects found: {model} with ID(s) {ids} appears multiple times"
666 ).format(
667 model=title(self.queryset.model._meta.verbose_name),
668 ids=', '.join(str(pk) for pk in sorted(duplicate_pks))
669 )
670 raise ValidationError(error_msg)
672 # A record which references an existing object by ID performs an update rather than a creation. The bulk
673 # import view is gated only on the 'add' permission, but updating an existing object requires 'change' (as
674 # enforced by the REST API). Require the 'change' permission at the model level before permitting any updates,
675 # and restrict the prefetched objects to those the user is permitted to change (object-level enforcement).
676 update_pks = set(prefetch_ids)
677 if prefetch_ids:
678 change_permission = get_permission_for_model(self.queryset.model, 'change')
679 if not request.user.has_perm(change_permission):
680 raise ValidationError(
681 _(
682 "This import includes {count} record(s) that reference an existing object by ID and would "
683 "update it, which requires the {permission} permission. Remove the ID column to create new "
684 "objects instead."
685 ).format(count=len(prefetch_ids), permission=change_permission)
686 )
687 change_queryset = self.queryset.model.objects.restrict(request.user, 'change')
688 else:
689 change_queryset = self.queryset.model.objects
691 prefetched_objects = {
692 obj.pk: obj
693 for obj in change_queryset.filter(id__in=prefetch_ids)
694 } if prefetch_ids else {}
696 # Delay tree updates until all saves are complete (MPTT plugin models only;
697 # no-op for ltree). TODO: Remove the wrapper in v5.0 (see _delay_mptt_updates).
698 with _delay_mptt_updates(self.queryset.model):
699 saved_objects = self._process_import_records(form, request, records, prefetched_objects)
701 # Enforce object-level permissions in aggregate. Newly created objects are constrained by the 'add'
702 # permission (self.queryset is already restricted to 'add'); updated objects by 'change' (reusing the
703 # queryset built above, so no additional per-record work). This runs inside the caller's atomic
704 # transaction, so any violation rolls back the entire import.
705 created_pks = [obj.pk for obj in saved_objects if obj.pk not in update_pks]
706 if self.queryset.filter(pk__in=created_pks).count() != len(created_pks):
707 raise PermissionsViolation()
708 updated_pks = [obj.pk for obj in saved_objects if obj.pk in update_pks]
709 if updated_pks and change_queryset.filter(pk__in=updated_pks).count() != len(updated_pks):
710 raise PermissionsViolation()
712 return saved_objects
714 #
715 # Request handlers
716 #
718 def get(self, request):
719 model = self.model_form._meta.model
720 form = BulkImportForm()
721 if not issubclass(model, ChangeLoggingMixin):
722 form.fields.pop('changelog_message')
724 return render(request, self.template_name, {
725 'model': model,
726 'form': form,
727 'fields': self._get_form_fields(),
728 'import_method': ImportMethodChoices.DIRECT,
729 'return_url': self.get_return_url(request),
730 **self.get_extra_context(request),
731 })
733 def post(self, request):
734 logger = logging.getLogger('netbox.views.BulkImportView')
735 model = self.model_form._meta.model
736 form = BulkImportForm(request.POST, request.FILES)
737 if not issubclass(model, ChangeLoggingMixin):
738 form.fields.pop('changelog_message')
740 if form.is_valid():
741 logger.debug("Import form validation was successful")
742 redirect_url = get_action_url(model, action='list')
744 # If indicated, defer this request to a background job & redirect the user
745 if form.cleaned_data['background_job']:
746 job_name = _('Bulk import {count} {object_type}').format(
747 count=len(form.cleaned_data['data']),
748 object_type=model._meta.verbose_name_plural,
749 )
750 if process_request_as_job(self.__class__, request, name=job_name):
751 return redirect(redirect_url)
753 try:
754 # Iterate through data and bind each record to a new model form instance. Object-level
755 # permissions are enforced within create_and_update_objects().
756 with transaction.atomic(using=router.db_for_write(model)):
757 new_objects = self.create_and_update_objects(form, request)
759 msg = _('Imported {count} {object_type}').format(
760 count=len(new_objects),
761 object_type=model._meta.verbose_name_plural
762 )
763 logger.info(msg)
765 # Handle background job
766 if is_background_request(request):
767 request.job.logger.info(msg)
768 return None
770 messages.success(request, msg)
771 return redirect(f"{redirect_url}?modified_by_request={request.id}")
773 except (AbortRequest, PermissionsViolation, ValidationError) as e:
774 err_messages = e.messages if type(e) is ValidationError else [e.message]
775 for msg in err_messages:
776 logger.debug(msg)
777 form.add_error(None, msg)
778 if is_background_request(request):
779 request.job.logger.error(msg)
780 request.job.logger.warning("Bulk import aborted")
781 clear_events.send(sender=self)
782 if is_background_request(request):
783 raise JobFailed
785 else:
786 logger.debug("Form validation failed")
788 return render(request, self.template_name, {
789 'model': model,
790 'form': form,
791 'fields': self._get_form_fields(),
792 # Return the user to the tab they submitted, so a rejected import shows its own error
793 'import_method': form.cleaned_data.get('import_method') or ImportMethodChoices.DIRECT,
794 'return_url': self.get_return_url(request),
795 **self.get_extra_context(request),
796 })
799class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
800 """
801 Edit objects in bulk.
803 Attributes:
804 filterset: FilterSet to apply when deleting by QuerySet
805 form: The form class used to edit objects in bulk
806 """
807 template_name = 'generic/bulk_edit.html'
808 filterset = None
809 form = None
811 def get_required_permission(self):
812 return get_permission_for_model(self.queryset.model, 'change')
814 def pre_save_operations(self, form, obj):
815 """
816 This method is called for each object in _update_objects immediately before full_clean() and
817 save(). Override to modify the object from form fields that don't map directly to a model field
818 (e.g. add/remove-style deltas), so the change is validated and persisted within the single
819 bulk-edit save. No-op by default.
820 """
821 pass
823 def post_save_operations(self, form, obj):
824 """
825 This method is called for each object in _update_objects. Override to perform additional object-level
826 operations that are specific to a particular ModelForm.
827 """
828 # Add/remove tags
829 if form.cleaned_data.get('add_tags', None):
830 obj.tags.add(*form.cleaned_data['add_tags'])
831 if form.cleaned_data.get('remove_tags', None):
832 obj.tags.remove(*form.cleaned_data['remove_tags'])
834 def _update_objects(self, form, request):
835 custom_fields = getattr(form, 'custom_fields', {})
836 standard_fields = [
837 field for field in form.fields if field not in list(custom_fields) + ['pk']
838 ]
839 nullified_fields = request.POST.getlist('_nullify')
840 updated_objects = []
841 model_fields = {}
842 m2m_fields = {}
844 # Build list of model fields and m2m fields for later iteration
845 for name in standard_fields:
846 try:
847 model_field = self.queryset.model._meta.get_field(name)
848 if isinstance(model_field, (ManyToManyField, ManyToManyRel)):
849 m2m_fields[name] = model_field
850 elif isinstance(model_field, GenericRel):
851 # Ignore generic relations (these may be used for other purposes in the form)
852 continue
853 else:
854 model_fields[name] = model_field
855 except FieldDoesNotExist:
856 # This form field is used to modify a field rather than set its value directly
857 model_fields[name] = None
859 for obj in self.queryset.filter(pk__in=form.cleaned_data['pk']):
861 # Take a snapshot of change-logged models
862 if hasattr(obj, 'snapshot'):
863 obj.snapshot()
865 # Attach the changelog message (if any) to the object
866 obj._changelog_message = form.cleaned_data.get('changelog_message')
868 # Update standard fields. If a field is listed in _nullify, delete its value.
869 for name, model_field in model_fields.items():
870 # Handle nullification
871 if name in form.nullable_fields and name in nullified_fields:
872 if type(model_field) is GenericForeignKey:
873 setattr(obj, name, None)
874 else:
875 setattr(obj, name, None if model_field.null else '')
876 # Normal fields
877 elif name in form.changed_data:
878 setattr(obj, name, form.cleaned_data[name])
880 # Update custom fields
881 for name, customfield in custom_fields.items():
882 if not name.startswith('cf_'):
883 raise ImproperlyConfigured(
884 _("Custom field form field name must begin with 'cf_': {name}").format(name=name)
885 )
886 cf_name = name[3:] # Strip cf_ prefix
887 if name in form.nullable_fields and name in nullified_fields:
888 obj.custom_field_data[cf_name] = None
889 elif name in form.changed_data:
890 obj.custom_field_data[cf_name] = customfield.serialize(form.cleaned_data[name])
892 # Store M2M values for validation
893 obj._m2m_values = {}
894 for field in obj._meta.local_many_to_many:
895 if value := form.cleaned_data.get(field.name):
896 obj._m2m_values[field.name] = list(value)
897 elif field.name in nullified_fields:
898 obj._m2m_values[field.name] = []
900 # Apply any form-driven modifications that don't map directly to a model field (e.g.
901 # add/remove deltas) before validation, so they're part of this single save.
902 self.pre_save_operations(form, obj)
904 obj.full_clean()
905 obj.save()
906 updated_objects.append(obj)
908 # Handle M2M fields after save
909 for name, m2m_field in m2m_fields.items():
910 if name in form.nullable_fields and name in nullified_fields:
911 getattr(obj, name).clear()
912 elif form.cleaned_data[name]:
913 getattr(obj, name).set(form.cleaned_data[name])
915 self.post_save_operations(form, obj)
917 if is_background_request(request):
918 request.job.logger.info(f"Updated {obj}")
920 # Rebuild the tree for MPTT plugin models (no-op for ltree; its triggers keep
921 # the tree current). TODO: Remove in v5.0 (see _rebuild_mptt_tree).
922 _rebuild_mptt_tree(self.queryset.model)
924 return updated_objects
926 #
927 # Request handlers
928 #
930 def get(self, request):
931 return redirect(self.get_return_url(request))
933 def post(self, request, **kwargs):
934 logger = logging.getLogger('netbox.views.BulkEditView')
935 model = self.queryset.model
937 # If we are editing *all* objects in the queryset, replace the PK list with all matched objects.
938 if request.POST.get('_all') and self.filterset is not None:
939 pk_list = self.filterset(request.GET, self.queryset.values_list('pk', flat=True), request=request).qs
940 else:
941 pk_list = request.POST.getlist('pk')
943 # Include the PK list as initial data for the form
944 initial_data = {'pk': pk_list}
946 # Check for other contextual data needed for the form. We avoid passing all of request.GET because the
947 # filter values will conflict with the bulk edit form fields.
948 # TODO: Find a better way to accomplish this
949 if 'device' in request.GET:
950 initial_data['device'] = request.GET.get('device')
951 elif 'device_type' in request.GET:
952 initial_data['device_type'] = request.GET.get('device_type')
953 elif 'virtual_machine' in request.GET:
954 initial_data['virtual_machine'] = request.GET.get('virtual_machine')
956 post_data = request.POST.copy()
957 post_data.setlist('pk', pk_list)
959 # An HTMX request without "_apply" is a dependent-field refresh (e.g. changing a content type), not a
960 # submission. Build the form unbound with the submitted state as initial data so fields reconfigure
961 # without surfacing validation errors before the user clicks Apply.
962 if htmx_partial(request) and '_apply' not in request.POST:
963 initial_data.update(normalize_querydict(post_data))
964 initial_data['pk'] = pk_list
965 form = self.form(initial=initial_data)
966 else:
967 form = self.form(post_data, initial=initial_data)
968 restrict_form_fields(form, request.user)
970 if '_apply' in request.POST:
971 if form.is_valid():
972 logger.debug("Form validation was successful")
974 # If indicated, defer this request to a background job & redirect the user
975 if form.cleaned_data['background_job']:
976 job_name = _('Bulk edit {count} {object_type}').format(
977 count=len(form.cleaned_data['pk']),
978 object_type=model._meta.verbose_name_plural,
979 )
980 if process_request_as_job(self.__class__, request, name=job_name):
981 return redirect(self.get_return_url(request))
983 try:
984 with transaction.atomic(using=router.db_for_write(model)):
985 updated_objects = self._update_objects(form, request)
987 # Enforce object-level permissions
988 object_count = self.queryset.filter(pk__in=[obj.pk for obj in updated_objects]).count()
989 if object_count != len(updated_objects):
990 raise PermissionsViolation
992 msg = _('Updated {count} {object_type}').format(
993 count=len(updated_objects),
994 object_type=model._meta.verbose_name_plural,
995 )
996 logger.info(msg)
998 # Handle background job
999 if is_background_request(request):
1000 request.job.logger.info(msg)
1001 return None
1003 messages.success(self.request, msg)
1004 return redirect(self.get_return_url(request))
1006 except (AbortRequest, PermissionsViolation, ValidationError) as e:
1007 err_messages = e.messages if type(e) is ValidationError else [e.message]
1008 for msg in err_messages:
1009 logger.debug(msg)
1010 form.add_error(None, msg)
1011 if is_background_request(request):
1012 request.job.logger.error(msg)
1013 clear_events.send(sender=self)
1014 if is_background_request(request):
1015 raise JobFailed
1017 else:
1018 logger.debug("Form validation failed")
1020 # Retrieve objects being edited
1021 table = self.table(self.queryset.filter(pk__in=pk_list), orderable=False)
1022 if not table.rows:
1023 messages.warning(
1024 request,
1025 _("No {object_type} were selected.").format(object_type=model._meta.verbose_name_plural)
1026 )
1027 return redirect(self.get_return_url(request))
1029 return render(request, self.template_name, {
1030 'model': model,
1031 'form': form,
1032 'table': table,
1033 'return_url': self.get_return_url(request),
1034 **self.get_extra_context(request),
1035 })
1038class BulkRenameView(GetReturnURLMixin, BaseMultiObjectView):
1039 """
1040 An extendable view for renaming objects in bulk.
1042 Attributes:
1043 field_name: The name of the object attribute to rename (defaults to "name"). Used when
1044 rename_fields is not set; kept for backward compatibility with plugins.
1045 rename_fields: Tuple of field names that can be selected for renaming. When two or more
1046 fields are listed, the form renders a checkbox per field so the user can apply the
1047 find/replace pattern to any combination of them simultaneously.
1048 """
1049 field_name = 'name'
1050 rename_fields = ()
1051 template_name = 'generic/bulk_rename.html'
1052 # Match BulkEditView/BulkDeleteView behavior: allow passing a FilterSet
1053 # so "Select all N matching query" can expand across the full queryset.
1054 filterset = None
1056 def __init__(self, *args, **kwargs):
1057 super().__init__(*args, **kwargs)
1059 # Use the changelog-aware form for models that support change logging
1060 base_form = (
1061 NetBoxModelBulkRenameForm
1062 if issubclass(self.queryset.model, ChangeLoggingMixin)
1063 else BulkRenameForm
1064 )
1066 self.form = type('_Form', (base_form,), {
1067 'pk': ModelMultipleChoiceField(
1068 queryset=self.queryset,
1069 widget=MultipleHiddenInput(),
1070 ),
1071 })
1073 def get_required_permission(self):
1074 return get_permission_for_model(self.queryset.model, 'change')
1076 def _rename_objects(self, form, selected_objects, field_names=None):
1077 if field_names is None:
1078 field_names = [self.field_name]
1080 find = form.cleaned_data['find']
1081 replace = form.cleaned_data['replace']
1082 use_regex = form.cleaned_data['use_regex']
1083 renamed_pks = []
1085 for obj in selected_objects:
1086 # Take a snapshot of change-logged models
1087 if hasattr(obj, 'snapshot'):
1088 obj.snapshot()
1090 new_values = {}
1091 for field in field_names:
1092 current = getattr(obj, field, '') or ''
1093 if use_regex:
1094 try:
1095 new_values[field] = re.sub(find, replace, current)
1096 # Catch regex group reference errors
1097 except re.error:
1098 new_values[field] = current
1099 else:
1100 new_values[field] = current.replace(find, replace)
1102 obj.new_names = SimpleNamespace(**new_values)
1103 obj.has_changes = any(
1104 new_values[f] != (getattr(obj, f, '') or '') for f in field_names
1105 )
1106 renamed_pks.append(obj.pk)
1108 return renamed_pks
1110 def post(self, request):
1111 logger = logging.getLogger('netbox.views.BulkRenameView')
1112 # Default field list: either all rename_fields or the single legacy field_name
1113 field_names = list(self.rename_fields) if self.rename_fields else [self.field_name]
1115 # If we are editing *all* objects in the queryset, replace the PK list with all matched objects.
1116 if request.POST.get('_all') and self.filterset is not None:
1117 pk_list = self.filterset(request.GET, self.queryset.values_list('pk', flat=True), request=request).qs
1118 else:
1119 pk_list = request.POST.getlist('pk')
1121 selected_objects = self.queryset.filter(pk__in=pk_list)
1123 if '_preview' in request.POST or '_apply' in request.POST:
1124 form = self.form(request.POST, initial={'pk': pk_list})
1126 if form.is_valid():
1127 submitted = [
1128 f for f in request.POST.getlist('field_names')
1129 if self.rename_fields and f in self.rename_fields
1130 ]
1131 if self.rename_fields and not submitted:
1132 form.add_error(None, _("Select at least one field to rename."))
1133 elif submitted:
1134 field_names = submitted
1135 if not form.errors:
1136 try:
1137 with transaction.atomic(using=router.db_for_write(self.queryset.model)):
1138 renamed_pks = self._rename_objects(form, selected_objects, field_names)
1140 if '_apply' in request.POST:
1141 # Delay tree updates until all saves are complete (MPTT
1142 # plugin models only; no-op for ltree).
1143 # TODO: Remove the wrapper in v5.0 (see _delay_mptt_updates).
1144 with _delay_mptt_updates(self.queryset.model):
1145 for obj in selected_objects:
1146 for field in field_names:
1147 setattr(obj, field, getattr(obj.new_names, field))
1148 obj._changelog_message = form.cleaned_data.get('changelog_message', '')
1149 obj.save()
1151 # Enforce constrained permissions
1152 if self.queryset.filter(pk__in=renamed_pks).count() != len(selected_objects):
1153 raise PermissionsViolation
1155 messages.success(
1156 request,
1157 _("Renamed {count} {object_type}").format(
1158 count=len(selected_objects),
1159 object_type=self.queryset.model._meta.verbose_name_plural
1160 )
1161 )
1162 return redirect(self.get_return_url(request))
1164 except IntegrityError as e:
1165 messages.error(self.request, ", ".join(e.args))
1166 clear_events.send(sender=self)
1168 except (AbortRequest, PermissionsViolation) as e:
1169 logger.debug(e.message)
1170 form.add_error(None, e.message)
1171 clear_events.send(sender=self)
1173 else:
1174 form = self.form(initial={'pk': pk_list})
1176 return render(request, self.template_name, {
1177 'rename_fields': self.rename_fields,
1178 'selected_field_names': field_names,
1179 'form': form,
1180 'obj_type_plural': self.queryset.model._meta.verbose_name_plural,
1181 'selected_objects': selected_objects,
1182 'return_url': self.get_return_url(request),
1183 })
1186class BulkDeleteView(GetReturnURLMixin, BaseMultiObjectView):
1187 """
1188 Delete objects in bulk.
1190 Attributes:
1191 filterset: FilterSet to apply when deleting by QuerySet
1192 table: The table used to display devices being deleted
1193 """
1194 template_name = 'generic/bulk_delete.html'
1195 filterset = None
1196 table = None
1198 def get_required_permission(self):
1199 return get_permission_for_model(self.queryset.model, 'delete')
1201 #
1202 # Request handlers
1203 #
1205 def get(self, request):
1206 return redirect(self.get_return_url(request))
1208 def post(self, request, **kwargs):
1209 logger = logging.getLogger('netbox.views.BulkDeleteView')
1210 model = self.queryset.model
1212 # Are we deleting *all* objects in the queryset or just a selected subset?
1213 if request.POST.get('_all'):
1214 qs = model.objects.all()
1215 if self.filterset is not None:
1216 qs = self.filterset(request.GET, qs, request=request).qs
1217 pk_list = qs.only('pk').values_list('pk', flat=True)
1218 else:
1219 pk_list = [int(pk) for pk in request.POST.getlist('pk')]
1221 if '_confirm' in request.POST:
1222 form = BulkDeleteForm(model, request.POST)
1223 if form.is_valid():
1224 logger.debug("Form validation was successful")
1226 # If indicated, defer this request to a background job & redirect the user
1227 if form.cleaned_data['background_job']:
1228 job_name = _('Bulk delete {count} {object_type}').format(
1229 count=len(form.cleaned_data['pk']),
1230 object_type=model._meta.verbose_name_plural,
1231 )
1232 if process_request_as_job(self.__class__, request, name=job_name):
1233 return redirect(self.get_return_url(request))
1235 # Delete objects
1236 queryset = self.queryset.filter(pk__in=pk_list)
1237 deleted_count = queryset.count()
1238 try:
1239 with transaction.atomic(using=router.db_for_write(model)):
1240 for obj in queryset:
1242 # Take a snapshot of change-logged models
1243 if hasattr(obj, 'snapshot'):
1244 obj.snapshot()
1246 # Attach the changelog message (if any) to the object
1247 obj._changelog_message = form.cleaned_data.get('changelog_message')
1249 # Delete the object
1250 obj.delete()
1252 if is_background_request(request):
1253 request.job.logger.info(f"Deleted {obj}")
1255 msg = _('Deleted {count} {object_type}').format(
1256 count=deleted_count,
1257 object_type=model._meta.verbose_name_plural
1258 )
1259 logger.info(msg)
1261 # Handle background job
1262 if is_background_request(request):
1263 request.job.logger.info(msg)
1264 return None
1266 messages.success(request, msg)
1268 except (ProtectedError, RestrictedError) as e:
1269 logger.warning(f"Caught {type(e)} while attempting to delete objects")
1270 clear_events.send(sender=self)
1271 if is_background_request(request):
1272 request.job.logger.error(
1273 _("Deletion failed due to the presence of one or more dependent objects.")
1274 )
1275 raise JobFailed
1276 handle_protectederror(queryset, request, e)
1278 except AbortRequest as e:
1279 logger.debug(e.message)
1280 clear_events.send(sender=self)
1281 if is_background_request(request):
1282 request.job.logger.error(e.message)
1283 raise JobFailed
1284 messages.error(request, mark_safe(e.message))
1286 return redirect(self.get_return_url(request))
1288 logger.debug("Form validation failed")
1290 else:
1291 form = BulkDeleteForm(model, initial={
1292 'pk': pk_list,
1293 'return_url': self.get_return_url(request),
1294 })
1296 # Retrieve objects being deleted
1297 table = self.table(self.queryset.filter(pk__in=pk_list), orderable=False)
1298 if not table.rows:
1299 messages.warning(
1300 request,
1301 _("No {object_type} were selected.").format(object_type=model._meta.verbose_name_plural)
1302 )
1303 return redirect(self.get_return_url(request))
1305 return render(request, self.template_name, {
1306 'model': model,
1307 'form': form,
1308 'table': table,
1309 'return_url': self.get_return_url(request),
1310 **self.get_extra_context(request),
1311 })
1314#
1315# Device/VirtualMachine components
1316#
1318class BulkComponentCreateView(GetReturnURLMixin, BaseMultiObjectView):
1319 """
1320 Add one or more components (e.g. interfaces, console ports, etc.) to a set of Devices or VirtualMachines.
1321 """
1322 template_name = 'generic/bulk_add_component.html'
1323 parent_model = None
1324 parent_field = None
1325 form = None
1326 model_form = None
1327 filterset = None
1328 table = None
1330 def get_required_permission(self):
1331 return f'dcim.add_{self.queryset.model._meta.model_name}'
1333 def post(self, request):
1334 logger = logging.getLogger('netbox.views.BulkComponentCreateView')
1335 parent_model_name = self.parent_model._meta.verbose_name_plural
1336 model_name = self.queryset.model._meta.verbose_name_plural
1338 # Are we editing *all* objects in the queryset or just a selected subset?
1339 if request.POST.get('_all') and self.filterset is not None:
1340 queryset = self.filterset(request.GET, self.parent_model.objects.only('pk'), request=request).qs
1341 pk_list = [obj.pk for obj in queryset]
1342 else:
1343 pk_list = [int(pk) for pk in request.POST.getlist('pk')]
1345 selected_objects = self.parent_model.objects.filter(pk__in=pk_list)
1346 if not selected_objects:
1347 messages.warning(
1348 request,
1349 _("No {object_type} were selected.").format(object_type=self.parent_model._meta.verbose_name_plural)
1350 )
1351 return redirect(self.get_return_url(request))
1352 table = self.table(selected_objects, orderable=False)
1354 if '_create' in request.POST:
1355 form = self.form(request.POST)
1357 if form.is_valid():
1358 logger.debug("Form validation was successful")
1360 # If indicated, defer this request to a background job & redirect the user
1361 if form.cleaned_data['background_job']:
1362 job_name = _('Bulk add {count} {object_type}').format(
1363 count=len(form.cleaned_data['pk']),
1364 object_type=self.queryset.model._meta.verbose_name_plural,
1365 )
1366 if process_request_as_job(self.__class__, request, name=job_name):
1367 return redirect(self.get_return_url(request))
1369 new_components = []
1370 data = deepcopy(form.cleaned_data)
1371 changelog_message = data.pop('changelog_message', '')
1372 data.pop('background_job', None)
1373 replication_data = {
1374 field: data.pop(field) for field in form.replication_fields
1375 }
1377 try:
1378 with transaction.atomic(using=router.db_for_write(self.queryset.model)):
1380 for obj in data['pk']:
1382 pattern_count = len(replication_data[form.replication_fields[0]])
1383 for i in range(pattern_count):
1384 component_data = {
1385 self.parent_field: obj.pk
1386 }
1387 component_data.update(data)
1388 for field, values in replication_data.items():
1389 if values:
1390 component_data[field] = values[i]
1392 component_form = self.model_form(component_data)
1393 if component_form.is_valid():
1394 if changelog_message:
1395 component_form.instance._changelog_message = changelog_message
1396 instance = component_form.save()
1397 logger.debug(f"Created {instance} on {instance.parent_object}")
1398 new_components.append(instance)
1399 else:
1400 for field, errors in component_form.errors.as_data().items():
1401 for e in errors:
1402 err_msg = '{}: {}'.format(obj, ', '.join(e))
1403 form.add_error(field, err_msg)
1404 if is_background_request(request):
1405 request.job.logger.error(err_msg)
1407 # Enforce object-level permissions
1408 component_ids = [obj.pk for obj in new_components]
1409 if self.queryset.filter(pk__in=component_ids).count() != len(new_components):
1410 raise PermissionsViolation
1412 except IntegrityError:
1413 clear_events.send(sender=self)
1414 if is_background_request(request):
1415 request.job.logger.error(_("An integrity error occurred while creating components"))
1416 raise JobFailed
1418 except (AbortRequest, PermissionsViolation) as e:
1419 logger.debug(e.message)
1420 form.add_error(None, e.message)
1421 clear_events.send(sender=self)
1422 if is_background_request(request):
1423 request.job.logger.error(e.message)
1424 raise JobFailed
1426 if not form.errors:
1427 msg = _("Added {count} {component} to {parent_count} {parent}.").format(
1428 count=len(new_components),
1429 component=model_name,
1430 parent_count=len(form.cleaned_data['pk']),
1431 parent=parent_model_name,
1432 )
1433 logger.info(msg)
1435 # Handle background job
1436 if is_background_request(request):
1437 request.job.logger.info(msg)
1438 return None
1440 messages.success(request, msg)
1442 return redirect(self.get_return_url(request))
1444 else:
1445 logger.debug("Form validation failed")
1447 else:
1448 form = self.form(initial={'pk': pk_list})
1450 return render(request, self.template_name, {
1451 'form': form,
1452 'parent_model_name': parent_model_name,
1453 'model_name': model_name,
1454 'table': table,
1455 'return_url': self.get_return_url(request),
1456 })