Coverage for netbox_data_import/views.py: 96.22%
2801 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3import difflib
4import hashlib
5import logging
6import time
7import uuid
8from collections import Counter
9from collections.abc import Mapping
10from dataclasses import dataclass, replace
11from datetime import UTC, datetime, timedelta
12from functools import cached_property
13from typing import ClassVar
14from urllib.parse import parse_qs, urlencode, urlsplit
16from core.signals import clear_events
17from django.contrib import messages
18from django.contrib.auth.mixins import PermissionRequiredMixin
19from django.core.exceptions import PermissionDenied, ValidationError
20from django.db import DatabaseError, IntegrityError
21from django.http import Http404, HttpResponse, JsonResponse
22from django.shortcuts import get_object_or_404, redirect, render
23from django.template.defaultfilters import pluralize
24from django.urls import reverse
25from django.utils.http import url_has_allowed_host_and_scheme
26from django.views import View
27from netbox.views import generic
28from redis.exceptions import ConnectionError as RedisConnectionError
29from redis.exceptions import TimeoutError as RedisTimeoutError
30from utilities.permissions import get_permission_for_model
31from utilities.views import ConditionalLoginRequiredMixin
33from . import __version__ as _plugin_version
34from . import adapters, ip_assignment
35from .cable_disclosure import DEVICE_ROW, DISCLOSURE_SOURCE, POLICY_HIDDEN, POLICY_VISIBLE, policy_row_is_disclosed
36from .cable_target import ELIGIBLE_TERMINATION_LIMIT, eligible_terminations
37from .catalog import CANDIDATE_TARGET_PREFIX, CATALOG, POLICY_SECTIONS, OutputKind
38from .contact_resolution import PrimaryContactResolver, contact_identity, suggest_contact_roles
39from .device_field_review import DeviceFieldReviewer, sync_change_preview
40from .adapter_config import preview_port_class_map
41from .field_keys import SELECT_TERMINATION_TASK, parse_termination_field_key, port_class_mapping_errors
42from .filters import ImportProfileFilterSet, InferenceBackendFilterSet
43from .forms import (
44 CableClassMappingForm,
45 CableSegmentOverrideForm,
46 ClassRoleMappingForm,
47 ColumnMappingForm,
48 ColumnTransformRuleForm,
49 DeviceTypeMappingForm,
50 ImportProfileBulkEditForm,
51 ImportProfileFilterForm,
52 ImportProfileForm,
53 ImportProfileImportForm,
54 ImportSetupForm,
55 InferenceBackendFilterForm,
56 InferenceBackendForm,
57 PortClassMappingForm,
58 cable_policy_form_initial,
59)
60from .identity import identity_in, identity_text
61from .import_engine import (
62 ImportEngine,
63 PreconditionFailed,
64 SelectionError,
65 StalePlan,
66 StaleSourceDocument,
67 operator_failure_message,
68)
69from .models import (
70 PreviewState,
71 CableClassMapping,
72 CableSegmentOverride,
73 ClassRoleMapping,
74 ColumnMapping,
75 ColumnTransformRule,
76 DeviceExistingMatch,
77 DeviceTypeMapping,
78 PortClassMapping,
79 IgnoredFieldDifference,
80 ImportExecution,
81 ImportProfile,
82 InferenceBackend,
83 ManufacturerMapping,
84 SourceDocument,
85 SourceResolution,
86 locked_profile_policy,
87 locked_resolution_policy,
88 stored_import_source,
89 validate_adapter_target_module,
90 validate_contact_candidate_resolution,
91 validate_registered_adapter,
92 validate_source_resolution_fields,
93)
94from .netbox_reader import NetBoxReader, PlanningTargetUnavailable
95from .object_permissions import (
96 POLICY_WRITE_REFUSED,
97 ObjectPermissionDenied,
98 assess_permission_scoped_save_option,
99 delete_permission_scoped_objects,
100 save_permission_scoped_object,
101)
102from .plan import Disposition, ImportPlan, PlanError, Severity
103from .preview_coordinator import (
104 SYNC_FINISHED,
105 SYNC_NOT_THIS_PREVIEW,
106 UNREADABLE_PREVIEW,
107 CancelTraceSync,
108 CommandOutcome,
109 DiscardPreview,
110 LockedPreview,
111 PreviewClaim,
112 PreviewCommand,
113 PreviewCommandRefused,
114 QueueImport,
115 RereadPreview,
116 RestorePreview,
117 StalePreview,
118 StartPreview,
119 apply_preview_command,
120 planning_context_of,
121 read_preview,
122 setup_claim,
123 sync_hold_reason,
124 validate_preview_plan,
125 with_preview_port_class,
126)
127from .profile_yaml import (
128 DuplicateYamlKeyError,
129 ProfileDocumentInvalid,
130 apply_profile_document,
131 load_yaml_document,
132 serialize_profile,
133)
134from .review_workspace import (
135 OUTSIDE_CLAIM_MARK,
136 PROFILE_POLICY_MOVED,
137 TERMINATION_UNRESOLVABLE,
138 IneligibleDeviceSelection,
139 IneligibleLocationSelection,
140 SYNC_ALL_NOTHING,
141 SYNC_DEPENDENCY_HELD,
142 SYNC_NOTHING,
143 SYNC_SELECTION_CHANGED,
144 SYNC_SELECTION_INVALID,
145 ProfilePolicyMoved,
146 ReviewWorkspace,
147 UnacceptablePolicyDecision,
148 clear_cable_segment_override_and_replan,
149 clear_port_class_mapping_and_replan,
150 clear_trace_location_resolution_and_replan,
151 held_sync,
152 save_cable_class_mapping_and_replan,
153 save_cable_segment_override_and_replan,
154 save_port_class_mapping_and_replan,
155 save_termination_resolution_and_replan,
156 save_trace_device_resolution_and_replan,
157 save_trace_location_resolution_and_replan,
158 with_blocked_sync,
159)
160from .tables import (
161 CableClassMappingTable,
162 ClassRoleMappingTable,
163 ColumnMappingTable,
164 ColumnTransformRuleTable,
165 DeviceTypeMappingTable,
166 ImportExecutionTable,
167 ImportProfileTable,
168 InferenceBackendTable,
169 PortClassMappingTable,
170)
171from .trace_device_resolution import DeviceEvidence, eligible_trace_devices, source_device_key
172from .trace_location_resolution import (
173 eligible_trace_locations,
174 location_prefix_spellings,
175 present_location_tree,
176 site_locations,
177 source_location_key,
178)
179from .values import (
180 effective_device_name,
181 normalize_for_compare,
182 source_position,
183 source_text,
184 status_map,
185 translation_maps,
186)
189def _safe_next_url(request, fallback: str) -> str:
190 """Return a validated same-host redirect URL from POST or the fallback view name."""
191 url = request.POST.get("next", "")
192 if url and url_has_allowed_host_and_scheme(
193 url, allowed_hosts={request.get_host()}, require_https=request.is_secure()
194 ):
195 return url
196 return reverse(fallback)
199def _navigation_response(request, url):
200 """Send an HTMX caller through a real page load, or redirect a standard browser."""
201 if request.headers.get("HX-Request") == "true":
202 response = HttpResponse(status=204)
203 response["HX-Redirect"] = url
204 return response
205 return redirect(url)
208def _name_resolution_response(request, url):
209 """Return an updated preview for HTMX or redirect a standard browser."""
210 if request.headers.get("HX-Request") == "true":
211 preview_path = reverse("plugins:netbox_data_import:import_preview")
212 if urlsplit(url).path == preview_path:
213 preview_response = ImportPreviewView().render_preview(request, url)
214 if not 300 <= preview_response.status_code < 400:
215 return preview_response
216 url = preview_response.headers["Location"]
217 return _navigation_response(request, url)
220def _row_key(unit) -> str:
221 """Return the key the sync modal looks a row up by.
223 Row numbers repeat across object types, so the number alone lets one row replace another.
224 """
225 return f"{unit.object_type}:{unit.row_number}"
228#: The rack filter's "no rack" option value. Tom Select drops an option whose value is empty.
229NO_RACK_FILTER_VALUE = "__no_rack__"
232def _rack_filter_options(units) -> tuple[list[dict[str, str]], str]:
233 """Return the racks the preview names and the value that stands for no rack.
235 Rows that name no rack are offered as their own option, matching the rack view's group.
236 """
237 named = {unit.rack_name for unit in units if unit.rack_name}
238 options = [{"value": rack, "label": rack} for rack in sorted(named, key=identity_text)]
239 no_rack_value = NO_RACK_FILTER_VALUE
240 # A rack may legally carry the sentinel's name, and two options cannot share one value.
241 while no_rack_value in named:
242 no_rack_value += "_"
243 if any(unit.object_type == "device" and not unit.rack_name for unit in units):
244 options.append({"value": no_rack_value, "label": "(No rack)"})
245 return options, no_rack_value
248def _sync_change_preview_by_row(units, labels):
249 """Return each reviewed row's fields grouped by what a sync would do to them."""
250 previews = {}
251 for unit in units:
252 entries = sync_change_preview(unit.extra_data, labels)
253 if entries:
254 previews[_row_key(unit)] = entries
255 return previews
258def _candidate_values(extra_data):
259 """Return candidate values after validating the serialized display shape."""
260 candidate_values = extra_data.get("candidate_values", {})
261 if not isinstance(candidate_values, Mapping) or any(
262 not isinstance(candidates, Mapping) for candidates in candidate_values.values()
263 ):
264 raise ValidationError("The active Import Plan has invalid candidate values.")
265 return candidate_values
268def _contact_candidate_context(workspace, source_id):
269 """Return Contact candidates and row state for one row of the active preview."""
270 result_rows = [
271 row for row in workspace.units if str(row.source_id) == str(source_id) and row.object_type == "device"
272 ]
273 source_rows = [row for row in workspace.source_rows if str(row.get("source_id")) == str(source_id)]
274 if len(source_rows) != 1 or len(result_rows) != 1:
275 raise ValidationError("The candidate resolution does not identify one active preview row.")
277 candidates = _candidate_values(result_rows[0].extra_data).get("contact", {})
278 if not isinstance(candidates, Mapping) or not candidates:
279 raise ValidationError("The active preview row has no Contact candidate values.")
280 return (
281 {str(source_column): str(value) for source_column, value in candidates.items()},
282 source_rows[0],
283 result_rows[0],
284 )
287def _store_contact_for_unmatched_row(profile, resolved_fields, candidates, user):
288 """Store the Contact a row names while it still has no Device to assign it to.
290 Returns the resolved fields to save, which name the stored Contact, the sentence the operator
291 is told about it, and the stored Contact itself. A decision that names no Contact stores nothing.
292 """
293 selection = PrimaryContactResolver.selection_for_resolution(profile, resolved_fields, candidates)
294 if selection is None:
295 return resolved_fields, "", None
296 contact, created = PrimaryContactResolver.create_contact(profile, selection, user)
297 note = (
298 f" Contact '{contact.name}' was created in NetBox."
299 if created
300 else f" Contact '{contact.name}' already existed in NetBox."
301 )
302 return {**resolved_fields, "contact_id": contact.pk}, note, contact_identity(contact)
305def _planned_device_id(result_row):
306 """Return the Device a row plans to write to, or None when the import refused the row.
308 A refused row still carries the Device it matched, so the identifier alone does not mean the
309 import accepted the match.
310 """
311 if result_row.action != "update":
312 return None
313 return result_row.extra_data.get("netbox_device_id")
316@dataclass(frozen=True)
317class _ContactWrite:
318 """What one decided Contact changed on the Device a row already matched."""
320 assignment_changed: bool
321 contact_created: bool
322 contact: dict
325def _assign_contact_to_matched_device(profile, resolved_fields, source_row, device_id, user) -> _ContactWrite | None:
326 """Apply the decided Contact to the Device this row already matched.
328 Returns what the write changed, or None when the decision names no Contact. `apply` creates the
329 Contact even when the assignment itself is unchanged, so the two are reported separately.
330 """
331 from dcim.models import Device
333 device = Device.objects.restrict(user, "change").filter(pk=device_id).first()
334 if device is None:
335 raise ObjectPermissionDenied("dcim.change_device")
336 resolved_row = dict(source_row)
337 resolved_row.update(resolved_fields)
338 review = PrimaryContactResolver.review(device, resolved_row, profile, user)
339 plan = PrimaryContactResolver.apply(device, profile, review, user)
340 if plan is None:
341 return None
342 return _ContactWrite(
343 assignment_changed=plan["assignment_action"] != "unchanged",
344 contact_created=plan["contact_id"] is None,
345 contact=plan["saved_contact"],
346 )
349@dataclass(frozen=True)
350class _ContactDecision:
351 """One saved Contact decision: the fields to store, and what writing it changed."""
353 resolved_fields: Mapping
354 write: _ContactWrite | None = None
355 note: str = ""
356 contact: dict | None = None
359def _persist_contact_decision(profile, resolved_fields, candidates, contact_context, user) -> _ContactDecision:
360 """Write the Contact a decision names, before the decision itself is stored.
362 A row with a planned Device has the Contact assigned to it. A row without one stores the Contact
363 alone. Either way the returned fields name the Contact that was persisted, so the stored decision
364 links it instead of the null the page posted.
365 """
366 if contact_context is None:
367 return _ContactDecision(resolved_fields)
368 source_row, result_row = contact_context
369 device_id = _planned_device_id(result_row)
370 if not device_id:
371 # No Device to assign to yet, so the Contact itself is stored now.
372 fields, note, contact = _store_contact_for_unmatched_row(profile, resolved_fields, candidates, user)
373 return _ContactDecision(fields, note=note, contact=contact)
374 write = _assign_contact_to_matched_device(profile, resolved_fields, source_row, device_id, user)
375 if write is None:
376 return _ContactDecision(resolved_fields)
377 return _ContactDecision(
378 {**resolved_fields, "contact_id": write.contact["id"]},
379 write=write,
380 contact=write.contact,
381 )
384def _saved_resolution_report(contact_write, contact_note):
385 """Return the sentence and the write detail one saved Contact decision reports.
387 An assignment that did not move is not a Device Contact update, and a Contact this save created
388 is reported whether or not the assignment moved.
389 """
390 if contact_write is None:
391 return "Resolution saved." + contact_note, contact_note.strip()
392 detail = (
393 f"Contact '{contact_write.contact['name']}' was created in NetBox."
394 if contact_write.contact_created
395 else contact_note.strip()
396 )
397 message = (
398 "Resolution saved and the linked Device Contact was updated."
399 if contact_write.assignment_changed
400 else "Resolution saved. The Device Contact already stood as decided."
401 )
402 return (f"{message} {detail}" if detail else message), detail
405def _ensure_field_review_device_match(user, profile, source_id, device, source_asset_tag=""):
406 """Persist the confirmed source-to-device identity for a field review."""
407 existing_match = (
408 DeviceExistingMatch.objects.select_for_update().filter(profile=profile, source_id=source_id).first()
409 )
410 if existing_match is not None and existing_match.netbox_device_id != device.pk:
411 return False, "conflict"
412 conflicting_match = (
413 DeviceExistingMatch.objects.select_for_update()
414 .filter(profile=profile, netbox_device_id=device.pk)
415 .exclude(source_id=source_id)
416 .first()
417 )
418 if conflicting_match is not None:
419 return False, "conflict"
420 if existing_match is not None and existing_match.netbox_device_id == device.pk:
421 return True, ""
422 try:
423 save_permission_scoped_object(
424 user,
425 DeviceExistingMatch,
426 {"profile": profile, "source_id": source_id},
427 {
428 "netbox_device_id": device.pk,
429 "device_name": device.name,
430 "source_asset_tag": source_asset_tag,
431 },
432 )
433 except ObjectPermissionDenied:
434 return False, "permission"
435 return True, ""
438# ---------------------------------------------------------------------------
439# Fuzzy matching: source column name → NetBox target field canonical name
440# ---------------------------------------------------------------------------
442_ALIAS_TO_CANONICAL: dict[str, str] = {
443 # rack_name
444 "rack": "rack_name",
445 "rack_name": "rack_name",
446 "rack name": "rack_name",
447 # device_name
448 "name": "device_name",
449 "device_name": "device_name",
450 "device name": "device_name",
451 "hostname": "device_name",
452 "host": "device_name",
453 # make
454 "make": "make",
455 "manufacturer": "make",
456 "vendor": "make",
457 "brand": "make",
458 # model
459 "model": "model",
460 "device_type": "model",
461 "device type": "model",
462 "product": "model",
463 # serial
464 "serial": "serial",
465 "serial_number": "serial",
466 "serial number": "serial",
467 "sn": "serial",
468 # asset_tag
469 "asset_tag": "asset_tag",
470 "asset tag": "asset_tag",
471 "asset": "asset_tag",
472 "tag": "asset_tag",
473 # source_id
474 "source_id": "source_id",
475 "source id": "source_id",
476 "id": "source_id",
477 "uid": "source_id",
478 # u_position
479 "u_position": "u_position",
480 "u position": "u_position",
481 "position": "u_position",
482 "unit": "u_position",
483 "u": "u_position",
484 # u_height
485 "u_height": "u_height",
486 "u height": "u_height",
487 "height": "u_height",
488 "size": "u_height",
489 # face
490 "face": "face",
491 "side": "face",
492 # airflow
493 "airflow": "airflow",
494 "air_flow": "airflow",
495 # status
496 "status": "status",
497 "state": "status",
498 # device_class
499 "device_class": "device_class",
500 "device class": "device_class",
501 "class": "device_class",
502 "type": "device_class",
503 "role": "device_class",
504}
507def _fuzzy_match_netbox_field(column_name: str) -> str | None:
508 """Return the best-matching canonical target field name for a source column, or None."""
509 normalised = column_name.strip().lower()
510 if normalised in _ALIAS_TO_CANONICAL:
511 return _ALIAS_TO_CANONICAL[normalised]
512 matches = difflib.get_close_matches(normalised, _ALIAS_TO_CANONICAL.keys(), n=1, cutoff=0.6)
513 if matches:
514 return _ALIAS_TO_CANONICAL[matches[0]]
515 return None
518# ---------------------------------------------------------------------------
519# ImportProfile
520# ---------------------------------------------------------------------------
523logger = logging.getLogger(__name__)
525# An Import Plan error can name stored preview data, so a response states one fixed sentence instead.
526UNPLANNABLE_IMPORT = "This import could not be planned. The server log names the reason."
527UPLOAD_NOT_UTF8 = "The uploaded file is not UTF-8 text."
528UPLOAD_UNREADABLE = "Could not read the uploaded file. The server log names the reason."
531class ImportProfileListView(generic.ObjectListView):
532 """List all import profiles with their mapping counts."""
534 queryset = ImportProfile.objects.prefetch_related("column_mappings", "class_role_mappings", "device_type_mappings")
535 table = ImportProfileTable
536 filterset = ImportProfileFilterSet
537 filterset_form = ImportProfileFilterForm
538 template_name = "netbox_data_import/importprofile_list.html"
541class ImportProfileView(generic.ObjectView):
542 """Detail view for a single import profile, with inline mapping tables."""
544 queryset = ImportProfile.objects.prefetch_related(
545 "column_mappings",
546 "class_role_mappings",
547 "device_type_mappings",
548 "cable_class_mappings",
549 "port_class_mappings",
550 )
552 def get_extra_context(self, request, instance):
553 """Inject inline mapping tables into the template context."""
554 column_table = ColumnMappingTable(instance.column_mappings.all())
555 class_role_table = ClassRoleMappingTable(instance.class_role_mappings.all())
556 device_type_table = DeviceTypeMappingTable(instance.device_type_mappings.all())
557 transform_table = ColumnTransformRuleTable(instance.column_transform_rules.all())
558 cable_class_table = CableClassMappingTable(instance.cable_class_mappings.all(), viewer=request.user)
559 port_class_table = PortClassMappingTable(instance.port_class_mappings.all(), viewer=request.user)
560 applicable_policy_sections = frozenset(
561 section.key for section in POLICY_SECTIONS if section.applies_to(instance.output_kinds)
562 )
563 return {
564 "column_table": column_table,
565 "class_role_table": class_role_table,
566 "device_type_table": device_type_table,
567 "transform_table": transform_table,
568 "cable_class_table": cable_class_table,
569 "port_class_table": port_class_table,
570 "applicable_policy_sections": applicable_policy_sections,
571 }
574class ImportProfileEditView(generic.ObjectEditView):
575 """Create or edit an ImportProfile."""
577 queryset = ImportProfile.objects.all()
578 form = ImportProfileForm
579 template_name = "netbox_data_import/importprofile_edit.html"
582class ImportProfileDeleteView(generic.ObjectDeleteView):
583 """Delete an ImportProfile and all its child mappings."""
585 queryset = ImportProfile.objects.all()
588class InferenceBackendListView(generic.ObjectListView):
589 """Every configured backend row. At most one may be enabled, and that one is the active backend."""
591 queryset = InferenceBackend.objects.all()
592 table = InferenceBackendTable
593 filterset = InferenceBackendFilterSet
594 filterset_form = InferenceBackendFilterForm
597_INFERENCE_MODELS_SESSION_KEY = "netbox_data_import.inference_backend_models"
600class InferenceBackendView(generic.ObjectView):
601 """One backend row, as `resolve_active_backend` reads it while this row is the enabled one."""
603 queryset = InferenceBackend.objects.all()
605 def get_extra_context(self, request, instance):
606 """Offer model suggestions once, immediately after a connection test discovered them."""
607 discovery = request.session.pop(_INFERENCE_MODELS_SESSION_KEY, None)
608 if not isinstance(discovery, Mapping) or discovery.get("backend_id") != instance.pk:
609 return {}
610 models = discovery.get("models")
611 if not isinstance(models, list) or not all(isinstance(model, str) for model in models):
612 return {}
613 return {"model_choices": tuple(models)}
616class InferenceBackendEditView(generic.ObjectEditView):
617 """Create or edit one backend row. Model validation applies the api_root trust boundary."""
619 queryset = InferenceBackend.objects.all()
620 form = InferenceBackendForm
623class InferenceBackendDeleteView(generic.ObjectDeleteView):
624 """Delete one backend row. With no enabled row left, the active backend is the plugin setting fallback."""
626 queryset = InferenceBackend.objects.all()
629class InferenceBackendChangeLogView(generic.ObjectChangeLogView):
630 """Display the change log for one InferenceBackend."""
632 queryset = InferenceBackend.objects.all()
635class InferenceBackendConnectionTestView(PermissionRequiredMixin, View):
636 """Run the connection test. Specification 13.1 authorizes it with this one permission."""
638 permission_required = "netbox_data_import.change_inferencebackend"
640 def post(self, request, pk):
641 """Run the test now and show its redacted result where the configuration lives."""
642 from .inference_connection_test import run_connection_test
644 # restrict() applies the ObjectPermission constraints a model-level check would ignore.
645 backend = get_object_or_404(InferenceBackend.objects.restrict(request.user, "change"), pk=pk)
646 result = run_connection_test(backend.pk, backend.backend_key)
647 request.session.pop(_INFERENCE_MODELS_SESSION_KEY, None)
648 if result.models:
649 request.session[_INFERENCE_MODELS_SESSION_KEY] = {
650 "backend_id": backend.pk,
651 "models": list(result.models),
652 }
653 if result.category == "ok":
654 messages.success(request, f"Connection test succeeded. {result.detail}")
655 else:
656 category = result.category.replace("_", " ")
657 messages.error(request, f"Connection test failed ({category}). {result.detail}")
658 return redirect(backend.get_absolute_url())
661class ImportProfileBulkEditView(generic.BulkEditView):
662 """Bulk-edit selected ImportProfiles."""
664 queryset = ImportProfile.objects.all()
665 filterset = ImportProfileFilterSet
666 table = ImportProfileTable
667 form = ImportProfileBulkEditForm
670class ImportProfileBulkDeleteView(generic.BulkDeleteView):
671 """Bulk-delete selected ImportProfiles."""
673 queryset = ImportProfile.objects.all()
674 table = ImportProfileTable
677class ImportProfileChangeLogView(generic.ObjectChangeLogView):
678 """Display the change log for one ImportProfile."""
680 queryset = ImportProfile.objects.all()
683def _validate_model_instance(instance, label):
684 """Call full_clean() and surface ValidationErrors as ValueError so the atomic block rolls back."""
685 from django.core.exceptions import ValidationError as DjangoValidationError
687 try:
688 instance.full_clean(validate_unique=False)
689 except DjangoValidationError as exc:
690 if hasattr(exc, "message_dict"):
691 msg = "; ".join(f"{f}: {', '.join(es)}" for f, es in exc.message_dict.items())
692 else:
693 msg = "; ".join(exc.messages)
694 raise PreviewCommandRefused(f"Validation error in {label}: {msg}") from exc
697def _get_or_init(model_class, **lookup):
698 """Return an existing object for a natural lookup, or one unsaved object."""
699 return model_class.objects.filter(**lookup).first() or model_class(**lookup)
702class ImportProfileBulkImportView(generic.BulkImportView):
703 """Import ImportProfile objects via NetBox's built-in import UI.
705 Supports two formats from the same text area / file upload:
707 * **Hierarchical YAML** - the format produced by the "Export YAML" button
708 (top-level keys: ``profile``, ``column_mappings``, ``class_role_mappings``,
709 ``device_type_mappings``, ``manufacturer_mappings``,
710 ``column_transform_rules``, ``cable_class_mappings``, ``port_class_mappings``). All nested
711 mappings are created/updated.
712 * **Flat CSV/YAML** - one record per profile, plain metadata fields only
713 (name, description, sheet_name, …). Falls back to NetBox's standard
714 bulk-import logic.
715 """
717 queryset = ImportProfile.objects.all()
718 model_form = ImportProfileImportForm
720 def post(self, request):
721 """Detect format and apply hierarchical YAML or delegate to flat bulk import."""
722 import yaml
724 # Read the raw input from the file upload or the text area.
725 upload = request.FILES.get("upload_file")
726 if upload:
727 try:
728 raw = upload.read().decode("utf-8-sig")
729 except UnicodeDecodeError:
730 messages.error(request, UPLOAD_NOT_UTF8)
731 return redirect(reverse("plugins:netbox_data_import:importprofile_bulk_import"))
732 except OSError:
733 logger.warning("ImportProfileBulkImportView: the uploaded file cannot be read.", exc_info=True)
734 messages.error(request, UPLOAD_UNREADABLE)
735 return redirect(reverse("plugins:netbox_data_import:importprofile_bulk_import"))
736 else:
737 raw = request.POST.get("data", "").strip()
739 if not raw:
740 messages.error(request, "No data provided.")
741 return redirect(reverse("plugins:netbox_data_import:importprofile_bulk_import"))
743 try:
744 data = load_yaml_document(raw)
745 except DuplicateYamlKeyError as exc:
746 messages.error(request, f"Failed to parse YAML: {exc}")
747 return redirect(reverse("plugins:netbox_data_import:importprofile_bulk_import"))
748 except yaml.YAMLError:
749 # Input failed YAML parsing — let NetBox's BulkImportView handle it
750 # (covers CSV and flat formats with YAML-invalid characters).
751 if upload:
752 upload.seek(0)
753 return super().post(request)
755 # Hierarchical format: delegate to shared helper.
756 if isinstance(data, dict) and "profile" in data:
757 try:
758 profile, stats = apply_profile_document(data, request.user)
759 except ObjectPermissionDenied as exc:
760 raise PermissionDenied from exc
761 except ProfileDocumentInvalid as exc:
762 messages.error(request, str(exc))
763 return redirect(reverse("plugins:netbox_data_import:importprofile_bulk_import"))
764 summary = ", ".join(f"{v} {k.replace('_', ' ')}" for k, v in stats.items())
765 messages.success(request, f"Profile '{profile.name}' imported/updated. {summary}.")
766 return redirect(profile.get_absolute_url())
768 # Flat format: let NetBox's BulkImportView handle it.
769 # Rewind the file stream so the parent handler receives the full content.
770 if upload:
771 upload.seek(0)
772 return super().post(request)
775# ---------------------------------------------------------------------------
776# Shared base views for ImportProfile child objects
777# ---------------------------------------------------------------------------
780def _profile_pk_for_policy_write(view, url_kwargs):
781 """Return the ImportProfile whose policy this write changes.
783 An edit or delete reads it off the row, through the already-scoped queryset. An add names it in
784 the URL and reads it through the same scope. Either way a target outside the operator's grant
785 raises here, which `post()` reaches before it enters the lock, so it never holds a row the
786 operator cannot see.
787 """
788 if "pk" in url_kwargs:
789 return get_object_or_404(view.queryset, pk=url_kwargs["pk"]).profile_id
790 profile_pk = url_kwargs.get("profile_pk")
791 if profile_pk is None:
792 return None
793 return get_object_or_404(ImportProfile.objects.restrict(view.request.user, "view"), pk=profile_pk).pk
796class _ProfileChildEditView(generic.ObjectEditView):
797 """Base add/edit view for objects that belong to an ImportProfile.
799 Assigns ``profile`` on add from the ``profile_pk`` URL kwarg, and redirects back to the
800 parent profile detail page after a successful save. The forms carry no ``profile`` field,
801 so a posted one is ignored.
803 Object scoping comes from NetBox's ``ObjectPermissionRequiredMixin``. Django's
804 ``PermissionRequiredMixin`` must not sit ahead of it: that shadows the ``restrict()`` call.
806 Override ``get_required_permission`` so that add-URLs (which carry
807 ``profile_pk`` but not ``pk``) are not misidentified as edit-URLs by
808 NetBox's generic ``dispatch`` hook.
809 """
811 def get_required_permission(self):
812 action = "change" if "pk" in self.kwargs else "add"
813 return get_permission_for_model(self.queryset.model, action)
815 def get_object(self, **kwargs):
816 """Filter only by ``pk`` — ignore ``profile_pk`` URL kwarg.
818 NetBox's ``ObjectEditView.get()`` passes all URL kwargs to
819 ``get_object_or_404``. ``profile_pk`` is not a field on child
820 models, so we must strip it before the ORM lookup.
821 """
822 if "pk" in kwargs:
823 return get_object_or_404(self.queryset, pk=kwargs["pk"])
824 return self.queryset.model()
826 def alter_object(self, obj, request, url_args, url_kwargs):
827 if not obj.pk and "profile_pk" in url_kwargs:
828 # The URL names the parent, so the add scope has to cover the profile as well as the row.
829 obj.profile = get_object_or_404(
830 ImportProfile.objects.restrict(request.user, "view"), pk=url_kwargs["profile_pk"]
831 )
832 return obj
834 def get_return_url(self, request, obj=None):
835 if obj is not None and getattr(obj, "profile", None):
836 return obj.profile.get_absolute_url()
837 return super().get_return_url(request, obj)
839 def get_extra_context(self, request, instance):
840 if instance.pk:
841 return {"profile": instance.profile}
842 profile_pk = self.kwargs.get("profile_pk")
843 if profile_pk:
844 return {"profile": get_object_or_404(ImportProfile.objects.restrict(request.user, "view"), pk=profile_pk)}
845 return {}
847 def post(self, request, *args, **kwargs):
848 """Write under the profile policy lock, so a replan cannot commit against stale policy."""
849 try:
850 with locked_profile_policy(_profile_pk_for_policy_write(self, kwargs)):
851 # atomic-exit-safe: locked-policy-write-committed
852 return super().post(request, *args, **kwargs)
853 except ImportProfile.DoesNotExist:
854 # The URL names a profile that is gone, which is the 404 its own fetch would give.
855 raise Http404 from None
858class _ProfileChildDeleteView(generic.ObjectDeleteView):
859 """Base delete view for objects that belong to an ImportProfile.
861 Redirects to the parent profile detail page after successful deletion. Object scoping comes from
862 NetBox's ``ObjectPermissionRequiredMixin``, which Django's must not shadow.
863 """
865 def get_return_url(self, request, obj=None):
866 if obj is not None and getattr(obj, "profile", None):
867 return obj.profile.get_absolute_url()
868 return super().get_return_url(request, obj)
870 def post(self, request, *args, **kwargs):
871 """Delete under the profile policy lock, for the same reason the edit view takes it."""
872 try:
873 with locked_profile_policy(_profile_pk_for_policy_write(self, kwargs)):
874 # atomic-exit-safe: locked-policy-delete-committed
875 return super().post(request, *args, **kwargs)
876 except ImportProfile.DoesNotExist:
877 raise Http404 from None
880# ---------------------------------------------------------------------------
881# ColumnMapping CRUD
882# ---------------------------------------------------------------------------
885class ColumnMappingAddView(_ProfileChildEditView):
886 """Add a column mapping to an existing ImportProfile."""
888 queryset = ColumnMapping.objects.all()
889 form = ColumnMappingForm
890 template_name = "netbox_data_import/columnmapping_edit.html"
893class ColumnMappingEditView(_ProfileChildEditView):
894 """Edit an existing column mapping."""
896 queryset = ColumnMapping.objects.all()
897 form = ColumnMappingForm
898 template_name = "netbox_data_import/columnmapping_edit.html"
901class ColumnMappingDeleteView(_ProfileChildDeleteView):
902 """Delete a column mapping."""
904 queryset = ColumnMapping.objects.all()
907# ---------------------------------------------------------------------------
908# ClassRoleMapping CRUD
909# ---------------------------------------------------------------------------
912class ClassRoleMappingAddView(_ProfileChildEditView):
913 """Add a class→role mapping to an existing ImportProfile."""
915 queryset = ClassRoleMapping.objects.all()
916 form = ClassRoleMappingForm
917 template_name = "netbox_data_import/classrolemapping_edit.html"
920class ClassRoleMappingEditView(_ProfileChildEditView):
921 """Edit an existing class→role mapping."""
923 queryset = ClassRoleMapping.objects.all()
924 form = ClassRoleMappingForm
925 template_name = "netbox_data_import/classrolemapping_edit.html"
928class ClassRoleMappingDeleteView(_ProfileChildDeleteView):
929 """Delete a class→role mapping."""
931 queryset = ClassRoleMapping.objects.all()
934class CableClassMappingAddView(_ProfileChildEditView):
935 """Add a CableClass mapping to an existing ImportProfile."""
937 queryset = CableClassMapping.objects.all()
938 form = CableClassMappingForm
939 template_name = "netbox_data_import/cableclassmapping_edit.html"
940 permission_required = "netbox_data_import.add_cableclassmapping"
943class CableClassMappingEditView(_ProfileChildEditView):
944 """Edit an existing CableClass mapping."""
946 queryset = CableClassMapping.objects.all()
947 form = CableClassMappingForm
948 template_name = "netbox_data_import/cableclassmapping_edit.html"
949 permission_required = "netbox_data_import.change_cableclassmapping"
951 def get_object(self, **kwargs):
952 """Refuse an edit form that would disclose a row through its initial values."""
953 obj = super().get_object(**kwargs)
954 if obj.pk and not self.request.user.has_perm("netbox_data_import.view_cableclassmapping", obj):
955 raise PermissionDenied
956 return obj
959class CableClassMappingDeleteView(_ProfileChildDeleteView):
960 """Delete a CableClass mapping."""
962 queryset = CableClassMapping.objects.all()
963 permission_required = "netbox_data_import.delete_cableclassmapping"
965 def get_object(self, **kwargs):
966 """Refuse a delete page that would disclose the row it names."""
967 obj = super().get_object(**kwargs)
968 if not self.request.user.has_perm("netbox_data_import.view_cableclassmapping", obj):
969 raise PermissionDenied
970 return obj
973class PortClassMappingAddView(_ProfileChildEditView):
974 """Add a PortClass mapping to an existing ImportProfile."""
976 queryset = PortClassMapping.objects.all()
977 form = PortClassMappingForm
978 template_name = "netbox_data_import/portclassmapping_edit.html"
979 permission_required = "netbox_data_import.add_portclassmapping"
982class PortClassMappingEditView(_ProfileChildEditView):
983 """Edit an existing PortClass mapping."""
985 queryset = PortClassMapping.objects.all()
986 form = PortClassMappingForm
987 template_name = "netbox_data_import/portclassmapping_edit.html"
988 permission_required = "netbox_data_import.change_portclassmapping"
990 def get_object(self, **kwargs):
991 """Refuse an edit form that would disclose a row through its initial values."""
992 obj = super().get_object(**kwargs)
993 if obj.pk and not self.request.user.has_perm("netbox_data_import.view_portclassmapping", obj):
994 raise PermissionDenied
995 return obj
998class PortClassMappingDeleteView(_ProfileChildDeleteView):
999 """Delete a PortClass mapping."""
1001 queryset = PortClassMapping.objects.all()
1002 permission_required = "netbox_data_import.delete_portclassmapping"
1004 def get_object(self, **kwargs):
1005 """Refuse a delete page that would disclose the row it names."""
1006 obj = super().get_object(**kwargs)
1007 if not self.request.user.has_perm("netbox_data_import.view_portclassmapping", obj):
1008 raise PermissionDenied
1009 return obj
1012# ---------------------------------------------------------------------------
1013# DeviceTypeMapping CRUD
1014# ---------------------------------------------------------------------------
1017class DeviceTypeMappingAddView(_ProfileChildEditView):
1018 """Add a device type mapping to an existing ImportProfile."""
1020 queryset = DeviceTypeMapping.objects.all()
1021 form = DeviceTypeMappingForm
1022 template_name = "netbox_data_import/devicetypemapping_edit.html"
1025class DeviceTypeMappingEditView(_ProfileChildEditView):
1026 """Edit an existing device type mapping."""
1028 queryset = DeviceTypeMapping.objects.all()
1029 form = DeviceTypeMappingForm
1030 template_name = "netbox_data_import/devicetypemapping_edit.html"
1033class DeviceTypeMappingDeleteView(_ProfileChildDeleteView):
1034 """Delete a device type mapping."""
1036 queryset = DeviceTypeMapping.objects.all()
1039# ---------------------------------------------------------------------------
1040# Import Wizard — Phase 2 (setup + preview)
1041# ---------------------------------------------------------------------------
1044def _review_workspace_url(profile):
1045 """Return the review surface declared for one profile's complete output set."""
1046 if profile.output_kinds == frozenset({OutputKind.SOURCE_TRACE}):
1047 return reverse("plugins:netbox_data_import:trace_workspace")
1048 return reverse("plugins:netbox_data_import:import_preview")
1051# These views intentionally use raw django.views.View rather than a NetBox
1052# generic view base. The wizard is a three-step state machine, kept by the Preview Coordinator
1053# (setup → preview → run → results) that does not correspond to any single
1054# NetBox generic view pattern (ObjectEditView, ObjectListView, etc.). Using a
1055# raw View keeps the control flow explicit and avoids fighting ObjectEditView's
1056# form-save lifecycle, queryset requirements, and redirect conventions.
1059class ImportSetupView(PermissionRequiredMixin, View):
1060 """Step 1: select profile, upload file, choose site/location/tenant."""
1062 permission_required = "netbox_data_import.change_importprofile"
1064 def get(self, request):
1065 """Render the import setup form."""
1066 initial = {}
1067 if profile_pk := request.GET.get("profile"):
1068 initial["profile"] = profile_pk
1069 form = ImportSetupForm(initial=initial, user=request.user)
1070 return render(request, "netbox_data_import/import_setup.html", _import_setup_context(request, form))
1072 def post(self, request):
1073 """Store the uploaded file, plan it, and make it the session's preview."""
1074 form = ImportSetupForm(request.POST, request.FILES, user=request.user)
1075 if not form.is_valid():
1076 return render(request, "netbox_data_import/import_setup.html", _import_setup_context(request, form))
1078 profile = form.cleaned_data["profile"]
1079 excel_file = form.cleaned_data["excel_file"]
1080 command = StartPreview(
1081 profile=profile,
1082 content=excel_file.read(),
1083 filename=excel_file.name,
1084 site=form.cleaned_data["site"],
1085 location=form.cleaned_data.get("location"),
1086 tenant=form.cleaned_data.get("tenant"),
1087 )
1088 try:
1089 apply_preview_command(request, PreviewClaim.posted(request.POST), command)
1090 except StalePreview as exc:
1091 return _stale_response(request, exc, reverse("plugins:netbox_data_import:import_setup"))
1092 except ImportProfile.DoesNotExist:
1093 # The form validated the profile, and the setup lock found it deleted since then.
1094 raise Http404("The import profile is no longer available.") from None
1095 except (adapters.SourceUnreadable, adapters.UnknownSourceAdapter, PlanningTargetUnavailable) as exc:
1096 logger.warning("ImportSetupView: source planning refused", exc_info=True)
1097 messages.error(request, f"Failed to parse file: {operator_failure_message(exc)}")
1098 return render(request, "netbox_data_import/import_setup.html", _import_setup_context(request, form))
1099 except PlanError:
1100 logger.warning("ImportSetupView: planning produced an unreadable Import Plan.", exc_info=True)
1101 messages.error(request, UNPLANNABLE_IMPORT)
1102 return render(request, "netbox_data_import/import_setup.html", _import_setup_context(request, form))
1103 except PreviewCommandRefused as exc:
1104 messages.error(request, exc.operator_message)
1105 return render(request, "netbox_data_import/import_setup.html", _import_setup_context(request, form))
1106 return redirect(_review_workspace_url(profile))
1109_DEVICE_CONFLICT_ROW_LIST_KEYS = (
1110 "duplicate_serial_rows",
1111 "duplicate_asset_tag_rows",
1112)
1115def _other_conflict_row_identities(row, source_object_types_by_number):
1116 """Return the row number and object type named by one preview error."""
1117 identities = [
1118 (row_number, source_object_types_by_number.get(row_number, row.object_type))
1119 for row_number in row.extra_data.get("duplicate_source_id_rows", ())
1120 ]
1121 for key in _DEVICE_CONFLICT_ROW_LIST_KEYS:
1122 identities.extend((row_number, row.object_type) for row_number in row.extra_data.get(key, ()))
1123 conflict_row_number = row.extra_data.get("conflict_row_number")
1124 if conflict_row_number is not None:
1125 identities.append((conflict_row_number, row.object_type))
1126 claimed_by_row = row.extra_data.get("claimed_by_row")
1127 if claimed_by_row is not None:
1128 identities.append((claimed_by_row, row.object_type))
1129 return tuple(dict.fromkeys(identity for identity in identities if identity != (row.row_number, row.object_type)))
1132# A rack position collides between two rows, so either row of the group offers these.
1133_GROUP_CONFLICT_ACTIONS = ("ignore_position", "ignore_row")
1135# The Resolve column of the conflict comparison renders these, and nothing else.
1136_RESOLVE_COLUMN_ACTIONS = ("ignore_serial", "ignore_position", "ignore_row")
1139def _group_offered_actions(row, u_position, group_actions):
1140 """Return the actions one member of a conflict group can run, its own first."""
1141 offered = list(row.extra_data.get("offered_actions", []))
1142 if not row.source_id:
1143 return offered
1144 for action in group_actions:
1145 if action in offered or (action == "ignore_position" and not u_position):
1146 continue
1147 offered.append(action)
1148 return offered
1151def _conflict_comparison_row(row, source_rows_by_number, *, is_current, group_actions=()):
1152 """Return the source facts that the conflict comparison shows for one result row."""
1153 source_row = source_rows_by_number.get(row.row_number, {})
1154 extra_data = row.extra_data
1155 serial = extra_data.get("source_serial", source_row.get("serial", ""))
1156 asset_tag = extra_data.get("asset_tag", source_row.get("asset_tag", ""))
1157 rack_name = row.rack_name or source_row.get("rack_name", "")
1158 if not rack_name and row.object_type == "rack":
1159 rack_name = row.name
1160 u_position = extra_data.get("u_position", source_row.get("u_position"))
1161 offered = _group_offered_actions(row, u_position, group_actions)
1162 return {
1163 "row_number": row.row_number,
1164 "name": row.name,
1165 "source_id": row.source_id,
1166 "serial": serial,
1167 "asset_tag": asset_tag,
1168 "rack_name": rack_name,
1169 "u_position": u_position,
1170 "face": extra_data.get("face", source_row.get("face", "")),
1171 "action": row.action,
1172 "detail": row.detail,
1173 # The row column's own actions, plus the ones any member of this conflict group can run.
1174 "offered_actions": offered,
1175 # One source of truth for whether the Resolve column has anything to draw.
1176 "has_resolve_action": any(action in offered for action in _RESOLVE_COLUMN_ACTIONS),
1177 "duplicate_serial": extra_data.get("duplicate_serial", ""),
1178 "is_current": is_current,
1179 }
1182def _preview_rows_with_conflict_comparisons(workspace, source_rows, profile):
1183 """Copy preview rows and attach comparisons for each within-import row conflict."""
1184 result_rows_by_identity = {(row.row_number, row.object_type): row for row in workspace.units}
1185 source_rows_by_number = {row.get("_row_number"): row for row in source_rows}
1186 object_types_by_class = {
1187 mapping.source_class: "rack" if mapping.creates_rack else "device"
1188 for mapping in profile.class_role_mappings.all()
1189 }
1190 source_object_types_by_number = {}
1191 for source_row in source_rows:
1192 source_class = source_text(source_row.get("device_class"))
1193 if source_class in object_types_by_class:
1194 source_object_types_by_number[source_row.get("_row_number")] = object_types_by_class[source_class]
1195 conflict_rows_by_row = {}
1196 for row in workspace.units:
1197 other_rows = [
1198 result_rows_by_identity.get(identity)
1199 for identity in _other_conflict_row_identities(row, source_object_types_by_number)
1200 ]
1201 other_rows = [other_row for other_row in other_rows if other_row is not None]
1202 if not other_rows:
1203 continue
1204 group_actions = [
1205 action for action in row.extra_data.get("offered_actions", ()) if action in _GROUP_CONFLICT_ACTIONS
1206 ]
1207 conflict_rows_by_row[(row.row_number, row.object_type)] = [
1208 _conflict_comparison_row(row, source_rows_by_number, is_current=True, group_actions=group_actions),
1209 *(
1210 _conflict_comparison_row(
1211 other_row, source_rows_by_number, is_current=False, group_actions=group_actions
1212 )
1213 for other_row in other_rows
1214 ),
1215 ]
1217 preview_rows = []
1218 for row in workspace.units:
1219 preview_rows.append(
1220 replace(
1221 row,
1222 extra_data={
1223 **row.extra_data,
1224 "conflict_rows": conflict_rows_by_row.get((row.row_number, row.object_type), []),
1225 },
1226 )
1227 )
1228 return preview_rows
1231TARGET_GONE = "The saved import target is no longer available. Start a new preview."
1232CONCURRENT_WRITE = "A row this command writes changed while it was processed. Try again."
1235def _unregistered_adapter_reason(profile):
1236 """Return why this release cannot plan for the profile's adapter, or None."""
1237 try:
1238 validate_registered_adapter(profile)
1239 # Planning raises the same error for a registered adapter no Target Module implements.
1240 validate_adapter_target_module(profile.source_adapter)
1241 except ValidationError as exc:
1242 return "; ".join(exc.messages)
1243 return None
1246def _review_snapshot(request, *, profile_action="change"):
1247 """Return the active preview a review page shows, or the response that sends the operator elsewhere.
1249 A page load only reads (ADR 0004), so a preview it cannot show stays until a command replaces it.
1250 """
1251 snapshot = read_preview(request, profile_action=profile_action)
1252 setup = redirect(reverse("plugins:netbox_data_import:import_setup"))
1253 if snapshot.state == PreviewState.EXPIRED or snapshot.expired:
1254 messages.warning(request, "This import preview expired. Start a new import.")
1255 return None, setup
1256 if snapshot.state not in PreviewState.ACTIVE:
1257 messages.warning(request, "No import in progress. Please start a new import.")
1258 return None, setup
1259 if snapshot.profile is None:
1260 messages.warning(request, "Import profile not found.")
1261 return None, setup
1262 if reason := _unregistered_adapter_reason(snapshot.profile):
1263 messages.error(request, reason)
1264 return None, setup
1265 if snapshot.document is None:
1266 messages.warning(request, "The stored source is no longer available. Upload it again.")
1267 return None, setup
1268 if snapshot.state == PreviewState.SUBMITTED:
1269 from core.models import Job
1271 if not Job.objects.filter(pk=snapshot.job_id).exists():
1272 return None, render(
1273 request,
1274 "netbox_data_import/preview_notice.html",
1275 {
1276 "title": "Import Job is no longer available",
1277 "message": "The import Job was removed. Re-read this preview before the next decision.",
1278 "preview_claim": snapshot.claim,
1279 "offer_reread": True,
1280 "next_url": request.get_full_path(),
1281 },
1282 )
1283 return None, redirect(reverse("plugins:netbox_data_import:import_progress", kwargs={"pk": snapshot.job_id}))
1284 return snapshot, None
1287def _retained_sync(snapshot):
1288 """Return the status of the preview's own trace sync and why it holds the preview, for a page that says so."""
1289 if snapshot.state != PreviewState.SYNC_PENDING:
1290 return None, ""
1291 from core.models import Job
1293 from .jobs import IMPORT_TASK_LOST, LOST, import_job_status
1295 job = Job.objects.filter(pk=snapshot.job_id).first()
1296 if job is None:
1297 return None, SYNC_FINISHED
1298 status = import_job_status(job)
1299 if status.state == LOST:
1300 return status, IMPORT_TASK_LOST
1301 return status, sync_hold_reason([job.status]) or SYNC_FINISHED
1304def _sync_status_context(snapshot, identity: str, retained) -> dict:
1305 """Return what the workspace's sync status block renders from `_retained_sync`, on the page and for each poll."""
1306 status, reason = retained
1307 read_url = reverse("plugins:netbox_data_import:trace_sync_status")
1308 return {
1309 "sync_status": status,
1310 "retained_sync_reason": reason,
1311 "sync_active": status is not None and status.active,
1312 "sync_status_url": f"{read_url}?{urlencode({'trace': identity, **snapshot.claim.fields()})}",
1313 "sync_trace": identity,
1314 "preview_claim": snapshot.claim,
1315 "reread_next": _trace_workspace_url(identity),
1316 }
1319def _live_plan(snapshot, actor):
1320 """Return the plan live NetBox states right now for the preview's source, or None once its target is gone."""
1321 try:
1322 return ImportEngine.plan(snapshot.profile, snapshot.document, actor, snapshot.planning_context)
1323 except PlanningTargetUnavailable:
1324 return None
1327def _unreadable_preview_page(request, snapshot):
1328 """Show the recovery a preview whose stored plan this release cannot read still has: re-read it."""
1329 logger.warning("The stored Import Plan of this preview is unreadable.")
1330 return render(
1331 request,
1332 "netbox_data_import/preview_notice.html",
1333 {
1334 "title": "Preview needs a re-read",
1335 "message": "This preview was planned by an earlier release. Re-read it from its stored source.",
1336 "preview_claim": snapshot.claim,
1337 "offer_reread": True,
1338 "next_url": request.get_full_path(),
1339 },
1340 )
1343class ImportPreviewView(PermissionRequiredMixin, View):
1344 """Step 2: show the stored preview, let user confirm or go back."""
1346 permission_required = "netbox_data_import.change_importprofile"
1348 def get(self, request):
1349 """Render the current preview URL."""
1350 return self.render_preview(request, request.get_full_path())
1352 def render_preview(self, request, preview_url):
1353 """Render the session's stored preview and say whether live NetBox has moved under it."""
1354 snapshot, refusal = _review_snapshot(request)
1355 if refusal is not None:
1356 return refusal
1357 profile = snapshot.profile
1358 try:
1359 result = snapshot.workspace(request.user)
1360 except PlanError:
1361 logger.warning("ImportPreviewView: the stored Import Plan is unreadable.", exc_info=True)
1362 return _unreadable_preview_page(request, snapshot)
1363 if retained_reason := _retained_sync(snapshot)[1]:
1364 messages.warning(request, retained_reason)
1365 live = _live_plan(snapshot, request.user)
1366 if live is None:
1367 messages.warning(request, TARGET_GONE)
1368 return redirect(reverse("plugins:netbox_data_import:import_setup"))
1369 rows = result.source_rows
1371 # Build existing resolutions map for the split-name modal preview
1372 import json as _json
1374 from .models import SourceResolution
1376 existing_resolutions = {}
1377 for res in SourceResolution.objects.filter(profile=profile):
1378 existing_resolutions.setdefault(str(res.source_id), {})[res.source_column] = {
1379 "original_value": res.original_value,
1380 "resolved_fields": res.resolved_fields,
1381 }
1383 # Build device matching context for template
1384 device_matches = DeviceExistingMatch.objects.filter(profile=profile)
1385 device_match_source_ids = [m.source_id for m in device_matches]
1386 device_match_info = {}
1388 # Fetch device serial numbers from NetBox Device objects
1389 from dcim.models import Device
1391 netbox_device_ids = [m.netbox_device_id for m in device_matches]
1392 devices_by_id = {
1393 d.id: d for d in Device.objects.restrict(request.user, "view").filter(id__in=netbox_device_ids)
1394 }
1396 for match in device_matches:
1397 device = devices_by_id.get(match.netbox_device_id)
1398 # Bindings to devices outside the user's view scope carry no target metadata.
1399 if device is None:
1400 continue
1401 device_match_info[match.source_id] = {
1402 "device_id": match.netbox_device_id,
1403 "device_name": match.device_name,
1404 "device_serial": device.serial,
1405 }
1407 # The preview serves every adapter, and only the flat one declares a stored view mode.
1408 stored_view_mode = profile.adapter_settings.get("preview_view_mode", "rows")
1409 view_mode = parse_qs(urlsplit(preview_url).query).get("view", [stored_view_mode])[-1]
1411 # Build unused columns list: filter out any that are now mapped
1412 mapped_source_cols = set(profile.column_mappings.values_list("source_column", flat=True))
1413 raw_unused = {column["name"]: column for column in result.unused_columns}
1414 unused_columns = [
1415 {
1416 "name": col,
1417 "count": int(stats.get("count") or 0),
1418 "samples": stats.get("samples") or [],
1419 "suggested_field": _fuzzy_match_netbox_field(col),
1420 }
1421 for col, stats in raw_unused.items()
1422 if isinstance(stats, dict) and col not in mapped_source_cols
1423 ]
1424 unused_columns.sort(key=lambda x: -x["count"])
1425 conflicts_by_row = {
1426 _row_key(r): r.extra_data.get("conflicts", {}) for r in result.units if r.extra_data.get("conflicts")
1427 }
1428 # The modal names a field for the operator; the catalog is where those names live.
1429 target_field_labels = {key: CATALOG.display(key) for key, _label in CATALOG.choices()}
1430 candidate_values_by_row = {}
1431 try:
1432 for row in result.units:
1433 candidate_values = _candidate_values(row.extra_data)
1434 if candidate_values:
1435 candidate_values_by_row[_row_key(row)] = candidate_values
1436 except ValidationError as exc:
1437 messages.error(request, "; ".join(exc.messages))
1438 return redirect(reverse("plugins:netbox_data_import:import_setup"))
1439 contact_suggestions_by_row = {
1440 _row_key(r): r.extra_data["contact_suggestion"]
1441 for r in result.units
1442 if r.extra_data.get("contact_suggestion")
1443 }
1444 contact_role_suggestions_by_row = {
1445 row_number: suggest_contact_roles(candidates["contact"])
1446 for row_number, candidates in candidate_values_by_row.items()
1447 if candidates.get("contact")
1448 }
1449 extra_columns_by_row = {
1450 _row_key(r): r.extra_data.get("extra_columns", {})
1451 for r in result.units
1452 if r.extra_data.get("extra_columns")
1453 }
1454 sync_change_preview_by_row = _sync_change_preview_by_row(result.units, target_field_labels)
1455 rack_filter_options, no_rack_filter_value = _rack_filter_options(result.units)
1456 split_field_values_by_source_id = {
1457 r.source_id: _split_field_values(r) for r in result.units if r.object_type == "device" and r.source_id
1458 }
1459 preview_rows = _preview_rows_with_conflict_comparisons(result, rows, profile)
1461 non_card_error_rows = [
1462 r
1463 for r in result.units
1464 if r.action == "error" and not (r.object_type == "device" or (r.object_type == "rack" and r.name))
1465 ]
1467 return render(
1468 request,
1469 "netbox_data_import/import_preview.html",
1470 {
1471 "result": result,
1472 "preview_rows": preview_rows,
1473 "filename": snapshot.context.get("filename", ""),
1474 "profile_id": profile.pk,
1475 "profile": profile,
1476 "preview_url": preview_url,
1477 "view_mode": view_mode,
1478 # Only a trace preview has a workspace to open, so only it offers the link.
1479 "trace_workspace_available": result.has_traces,
1480 "existing_resolutions_json": _json.dumps(existing_resolutions).translate(
1481 {ord("<"): "\\u003C", ord(">"): "\\u003E", ord("&"): "\\u0026"}
1482 ),
1483 "existing_resolutions": existing_resolutions,
1484 "plugin_version": _plugin_version,
1485 "resolved_contact_source_ids": [
1486 source_id for source_id, columns in existing_resolutions.items() if "candidate:contact" in columns
1487 ],
1488 "configured_source_classes": set(profile.class_role_mappings.values_list("source_class", flat=True)),
1489 "can_create_role": request.user.has_perm("dcim.add_devicerole"),
1490 "unused_columns": unused_columns,
1491 "target_field_choices": CATALOG.choices(output_kinds=profile.output_kinds),
1492 "syncable_fields": SyncDeviceFieldView._ALLOWED_FIELDS,
1493 "reviewable_fields": DeviceFieldReviewer.reviewable_fields(),
1494 "device_match_source_ids": device_match_source_ids,
1495 "device_match_info": device_match_info,
1496 "conflicts_by_row": conflicts_by_row,
1497 "target_field_labels": target_field_labels,
1498 "candidate_values_by_row": candidate_values_by_row,
1499 "contact_suggestions_by_row": contact_suggestions_by_row,
1500 "contact_role_suggestions_by_row": contact_role_suggestions_by_row,
1501 "extra_columns_by_row": extra_columns_by_row,
1502 "sync_change_preview_by_row": sync_change_preview_by_row,
1503 "rack_filter_options": rack_filter_options,
1504 "no_rack_filter_value": no_rack_filter_value,
1505 "split_field_values_by_source_id": split_field_values_by_source_id,
1506 "non_card_error_rows": non_card_error_rows,
1507 "preview_claim": snapshot.claim,
1508 "drift": live.fingerprint != result.plan.fingerprint,
1509 },
1510 )
1513def _user_import_jobs(request):
1514 """Return native data-import Jobs owned by the current user."""
1515 from .jobs import ImportJobRunner
1517 return ImportJobRunner.get_jobs().filter(
1518 user=request.user,
1519 data__job_type=ImportJobRunner.job_type,
1520 )
1523def _import_setup_context(request, form):
1524 """Return the setup form, the claim it posts, and the most relevant resumable import state."""
1525 setup_claim(request)
1526 snapshot = read_preview(request, include_plan=False)
1527 resumable = snapshot.active and snapshot.state in (PreviewState.READY, PreviewState.SYNC_PENDING)
1528 return {
1529 "form": form,
1530 "preview_claim": snapshot.claim,
1531 "resume_job": _resume_import_job(request, snapshot),
1532 "resume_preview_url": _review_workspace_url(snapshot.profile) if resumable else "",
1533 "discardable": snapshot.state in PreviewState.ACTIVE,
1534 }
1537def _resume_import_job(request, snapshot):
1538 """Return the preview's own active Job, or the user's latest active import Job."""
1539 from core.choices import JobStatusChoices
1541 jobs = _user_import_jobs(request).filter(status__in=JobStatusChoices.ENQUEUED_STATE_CHOICES)
1542 if snapshot.job_id is not None and (job := jobs.filter(pk=snapshot.job_id).first()):
1543 return job
1544 return jobs.first()
1547def _import_source_rows_available(job):
1548 """Return whether one Job's stored source is still available."""
1549 source_document_id = (job.data or {}).get("source_document_id")
1550 return bool(source_document_id and SourceDocument.objects.filter(pk=source_document_id).exists())
1553def _import_job_progress(request, job):
1554 """Return current progress from native Job data and RQ metadata, and what the operator can do next.
1556 A failed final import offers its preview back only while the session's preview is still the one
1557 that submitted it; a newer preview is never replaced from here.
1558 """
1559 from core.choices import JobStatusChoices
1560 from .jobs import IMPORT_TASK_LOST, LOST, QUEUED, import_job_status
1562 data = job.data or {}
1563 status = import_job_status(job)
1564 percentage = round(status.processed * 100 / status.total) if status.total else 0
1565 abandoned = status.state == LOST
1566 cancelled = data.get("phase") == "cancelled"
1567 is_failed = abandoned or job.status in (JobStatusChoices.STATUS_FAILED, JobStatusChoices.STATUS_ERRORED)
1568 snapshot = read_preview(request, include_plan=False)
1569 restorable = (
1570 is_failed
1571 and snapshot.state == PreviewState.SUBMITTED
1572 and snapshot.job_id == job.pk
1573 and snapshot.active
1574 and _import_source_rows_available(job)
1575 )
1576 holds_preview = snapshot.active and snapshot.state == PreviewState.SYNC_PENDING and snapshot.job_id == job.pk
1577 execution_id = data.get("import_execution_id")
1578 return {
1579 "job": job,
1580 "status": status,
1581 "processed": status.processed,
1582 "total": status.total,
1583 "percentage": percentage,
1584 "is_active": status.active,
1585 "is_completed": job.status == JobStatusChoices.STATUS_COMPLETED,
1586 "is_failed": is_failed,
1587 "cancelled": cancelled,
1588 "cancel_claim": snapshot.claim if holds_preview and status.state == QUEUED else None,
1589 "workspace_url": _review_workspace_url(snapshot.profile) if cancelled and snapshot.active else "",
1590 "restore_claim": snapshot.claim if restorable else None,
1591 "preview_replaced": (
1592 is_failed
1593 and not restorable
1594 and not cancelled
1595 and isinstance(data.get("context_data"), dict)
1596 and snapshot.job_id != job.pk
1597 ),
1598 "results_url": (
1599 reverse("plugins:netbox_data_import:import_results", kwargs={"pk": execution_id}) if execution_id else ""
1600 ),
1601 "message": IMPORT_TASK_LOST if abandoned else data.get("message") or "",
1602 }
1605class _PreviewCommandMixin:
1606 """Render the refusals a preview command raises, the same way for every command view."""
1608 def dispatch(self, request, *args, **kwargs):
1609 try:
1610 return super().dispatch(request, *args, **kwargs)
1611 except (StalePreview, ProfilePolicyMoved) as exc:
1612 return _stale_response(request, exc, self.refusal_url(request), json=self._answers_json(request))
1613 except PreviewCommandRefused as exc:
1614 return self._refusal(request, exc.operator_message, exc.status)
1615 except StaleSourceDocument as exc:
1616 return self._refusal(request, exc.operator_message, 409, reverse("plugins:netbox_data_import:import_setup"))
1617 except PlanningTargetUnavailable:
1618 return self._refusal(request, TARGET_GONE, 409, reverse("plugins:netbox_data_import:import_setup"))
1619 except ObjectPermissionDenied as exc:
1620 # The permission names an object the caller may not be allowed to know exists.
1621 logger.warning("%s: write refused outside the caller's object scope: %s", type(self).__name__, exc)
1622 return self._refusal(
1623 request,
1624 "Permission denied: this action is outside your NetBox object permissions.",
1625 403,
1626 )
1627 except ImportProfile.DoesNotExist:
1628 # Every policy write locks its profile, which can be deleted after the page was rendered.
1629 return self._refusal(request, "The import profile is no longer available.", 404)
1630 except IntegrityError:
1631 return self._refusal(request, CONCURRENT_WRITE, 409)
1632 except ValidationError as exc:
1633 # A model refused a value the command derived from the plan, so the command wrote nothing.
1634 return self._refusal(request, "; ".join(exc.messages), 400)
1636 def _answers_json(self, request) -> bool:
1637 return getattr(self, "permission_denied_response_format", "redirect") == "json" or _wants_json(request)
1639 def _refusal(self, request, error, status, url=None):
1640 """Render one refused command the way this caller asked for its answer."""
1641 if self._answers_json(request):
1642 return JsonResponse({"ok": False, "error": error}, status=status)
1643 messages.error(request, error)
1644 return _navigation_response(request, url or self.refusal_url(request))
1646 def refusal_url(self, request):
1647 """Return the page a refused form command goes back to."""
1648 return _safe_next_url(request, "plugins:netbox_data_import:import_preview")
1651def _stale_json(exc):
1652 """Refuse a stale preview read or command in the JSON envelope every preview script reads."""
1653 return JsonResponse({"ok": False, "error": exc.operator_message, "code": "preview_stale"}, status=409)
1656def _stale_response(request, exc, url, *, json=False):
1657 """Refuse a command made against a preview that is not the active one, with HTTP 409 in every format."""
1658 message = exc.operator_message
1659 if json or _wants_json(request):
1660 return _stale_json(exc)
1661 if request.headers.get("HX-Request") == "true":
1662 # htmx follows this header before it decides whether a 409 swaps, so the page reloads and says why.
1663 messages.error(request, message)
1664 response = HttpResponse(status=409)
1665 response["HX-Redirect"] = url
1666 return response
1667 return render(
1668 request,
1669 "netbox_data_import/preview_notice.html",
1670 {"title": "This preview changed", "message": message, "next_url": url},
1671 status=409,
1672 )
1675class FinalImport(QueueImport):
1676 """Queue every actionable unit of the reviewed plan, which submits the preview."""
1678 def selection_for(self, preview):
1679 """Refuse a plan with errors or with nothing to apply; otherwise select every actionable unit."""
1680 if preview.workspace.has_errors:
1681 raise PreviewCommandRefused("Resolve every preview error before importing.")
1682 selection = [unit.identity for unit in preview.plan.units if unit.disposition == Disposition.ACTIONABLE]
1683 if not selection:
1684 raise PreviewCommandRefused("The accepted Import Plan has no changes to apply.")
1685 return selection
1688class ImportRunView(_PreviewCommandMixin, PermissionRequiredMixin, View):
1689 """Step 3: queue the accepted Import Plan."""
1691 permission_required = "netbox_data_import.change_importprofile"
1693 def post(self, request):
1694 """Queue the accepted plan and redirect to its progress page."""
1695 result = apply_preview_command(request, PreviewClaim.posted(request.POST), FinalImport())
1696 return redirect(reverse("plugins:netbox_data_import:import_progress", kwargs={"pk": result.outcome.job_id}))
1699SESSION_ENDED = "Your session has ended. Reload the page to log in again."
1702class _SessionEndedRefusal:
1703 """Refuse an ended session in the JSON envelope, so htmx never swaps the login page into the workspace."""
1705 def handle_no_permission(self):
1706 """Answer an anonymous script or htmx caller with 401; a plain form post still goes to the login page."""
1707 if not self.request.user.is_authenticated and (
1708 self.request.headers.get("HX-Request") == "true" or _wants_json(self.request)
1709 ):
1710 return JsonResponse({"ok": False, "error": SESSION_ENDED}, status=401)
1711 return super().handle_no_permission()
1714class PreviewRereadView(_SessionEndedRefusal, _PreviewCommandMixin, PermissionRequiredMixin, View):
1715 """Re-read the preview from its stored source against live NetBox, which also recovers a plan of an older schema."""
1717 permission_required = "netbox_data_import.change_importprofile"
1719 def post(self, request):
1720 """Replan the preview and go back to the page that asked."""
1721 result = apply_preview_command(request, PreviewClaim.posted(request.POST), RereadPreview())
1722 messages.success(request, result.outcome.message)
1723 return redirect(self.refusal_url(request))
1725 def refusal_url(self, request):
1726 """Return the posted page, so a re-read keeps the trace and the view the operator was on."""
1727 return _safe_next_url(request, "plugins:netbox_data_import:import_preview")
1730class PreviewDiscardView(_PreviewCommandMixin, PermissionRequiredMixin, View):
1731 """Discard the session's preview."""
1733 permission_required = "netbox_data_import.change_importprofile"
1735 def post(self, request):
1736 """End the preview and go back to setup."""
1737 result = apply_preview_command(request, PreviewClaim.posted(request.POST), DiscardPreview())
1738 messages.success(request, result.outcome.message)
1739 return redirect(reverse("plugins:netbox_data_import:import_setup"))
1741 def refusal_url(self, request):
1742 """Return the setup page, which shows whatever preview is active now."""
1743 return reverse("plugins:netbox_data_import:import_setup")
1746class ImportRestoreView(_PreviewCommandMixin, PermissionRequiredMixin, View):
1747 """Return to the preview whose final import failed."""
1749 permission_required = "netbox_data_import.change_importprofile"
1751 def post(self, request, pk):
1752 """Re-read the failed import's preview and open it."""
1753 get_object_or_404(_user_import_jobs(request), pk=pk)
1754 result = apply_preview_command(request, PreviewClaim.posted(request.POST), RestorePreview(pk))
1755 messages.success(request, result.outcome.message)
1756 profile = ImportProfile.objects.filter(pk=result.profile_id).first()
1757 return redirect(_review_workspace_url(profile) if profile is not None else self.refusal_url(request))
1759 def refusal_url(self, request):
1760 """Return the progress page the command came from."""
1761 return reverse("plugins:netbox_data_import:import_progress", kwargs=request.resolver_match.kwargs)
1764class ImportProgressView(PermissionRequiredMixin, View):
1765 """Show one resumable background import and its current progress."""
1767 permission_required = "netbox_data_import.change_importprofile"
1769 def get(self, request, pk):
1770 """Render the full progress page."""
1771 job = get_object_or_404(_user_import_jobs(request), pk=pk)
1772 return render(request, "netbox_data_import/import_progress.html", _import_job_progress(request, job))
1775class ImportProgressStatusView(PermissionRequiredMixin, View):
1776 """Render the HTMX progress fragment or redirect a completed import."""
1778 permission_required = "netbox_data_import.change_importprofile"
1780 def get(self, request, pk):
1781 """Return the current Job state."""
1782 job = get_object_or_404(_user_import_jobs(request), pk=pk)
1783 progress = _import_job_progress(request, job)
1784 if progress["is_completed"] and progress["results_url"]:
1785 response = HttpResponse(status=204)
1786 response["HX-Redirect"] = progress["results_url"]
1787 return response
1788 return render(request, "netbox_data_import/_import_progress.html", progress)
1791class ImportResultsView(PermissionRequiredMixin, View):
1792 """Step 4: show one Import Execution audit outcome."""
1794 permission_required = "netbox_data_import.view_importexecution"
1796 def get(self, request, pk):
1797 """Render the results page for one Import Execution this operator ran."""
1798 execution = (
1799 ImportExecution.objects.select_related("profile", "source_document")
1800 .filter(pk=pk, actor=request.user)
1801 .first()
1802 )
1803 if execution is None or not request.user.has_perm(self.permission_required, execution):
1804 return redirect(reverse("plugins:netbox_data_import:import_setup"))
1805 return render(
1806 request,
1807 "netbox_data_import/import_results.html",
1808 {"execution": execution, "job_id": execution.pk},
1809 )
1812class ImportExecutionListView(PermissionRequiredMixin, generic.ObjectListView):
1813 """List every past Import Execution, including the retained legacy rows."""
1815 queryset = ImportExecution.objects.select_related("profile").all()
1816 table = ImportExecutionTable
1817 template_name = "netbox_data_import/importexecution_list.html"
1818 permission_required = "netbox_data_import.view_importexecution"
1820 def get_required_permission(self):
1821 """Answer NetBox's own permission hook, which it checks separately from permission_required."""
1822 return "netbox_data_import.view_importexecution"
1825# ---------------------------------------------------------------------------
1826# ColumnTransformRule CRUD
1827# ---------------------------------------------------------------------------
1830class ColumnTransformRuleAddView(_ProfileChildEditView):
1831 """Add a column transform rule to an existing ImportProfile."""
1833 queryset = ColumnTransformRule.objects.all()
1834 form = ColumnTransformRuleForm
1835 template_name = "netbox_data_import/columntransformrule_edit.html"
1838class ColumnTransformRuleEditView(_ProfileChildEditView):
1839 """Edit an existing column transform rule."""
1841 queryset = ColumnTransformRule.objects.all()
1842 form = ColumnTransformRuleForm
1843 template_name = "netbox_data_import/columntransformrule_edit.html"
1846class ColumnTransformRuleDeleteView(_ProfileChildDeleteView):
1847 """Delete a column transform rule."""
1849 queryset = ColumnTransformRule.objects.all()
1852# ---------------------------------------------------------------------------
1853# Ignore / Unignore device
1854# ---------------------------------------------------------------------------
1855# The action views below (Ignore/Unignore/Sync/Quick*) are lightweight POST
1856# endpoints that return JSON or an immediate redirect. No NetBox generic base
1857# class exists for this pattern; PermissionRequiredMixin + View is intentional.
1858# ---------------------------------------------------------------------------
1861def _wants_json(request) -> bool:
1862 """Return whether one preview action expects a JSON response."""
1863 return "application/json" in request.headers.get("Accept", "")
1866def _command_response(request, result, next_url, *, row_number=None, detail="", **extra):
1867 """Report one committed preview command: JSON for a script, a message and a redirect for a form.
1869 The coordinator replanned and advanced the revision, so the page loads again to show both.
1870 """
1871 message = result.outcome.message
1872 # The page loads again after every command, so the message waits for that load in both formats.
1873 messages.success(request, f"{message} {detail}" if detail and detail not in message else message)
1874 if _wants_json(request):
1875 return JsonResponse(
1876 {
1877 "ok": True,
1878 "row_number": row_number,
1879 "preview_state": "replanned",
1880 "message": message,
1881 "detail": detail,
1882 **extra,
1883 }
1884 )
1885 return redirect(next_url)
1888def _resolved_import_target(planning_context, user):
1889 """Return the engine context the preview names, or None once that target went stale.
1891 A preview outlives a permission change, so each command re-reads the target in the operator's own
1892 scope: a revoked ObjectPermission has to make the target unavailable, not merely unlisted.
1893 """
1894 from dcim.models import Location, Site
1895 from tenancy.models import Tenant
1897 sites = Site.objects.restrict(user, "view")
1898 locations = Location.objects.restrict(user, "view")
1899 tenants = Tenant.objects.restrict(user, "view")
1900 site = sites.filter(pk=planning_context.get("site_id")).first()
1901 location_id, tenant_id = planning_context.get("location_id"), planning_context.get("tenant_id")
1902 location = locations.filter(pk=location_id).first() if location_id else None
1903 tenant = tenants.filter(pk=tenant_id).first() if tenant_id else None
1904 if (
1905 site is None
1906 or (location_id and (location is None or location.site_id != site.pk))
1907 or (tenant_id and tenant is None)
1908 ):
1909 return None
1910 return {"site": site, "location": location, "tenant": tenant}
1913def _source_row(workspace, source_id) -> dict:
1914 """Return the one row of the active preview that carries this source ID, or refuse the command."""
1915 rows = [row for row in workspace.source_rows if source_text(row.get("source_id")) == source_id]
1916 if not source_id or len(rows) != 1:
1917 raise PreviewCommandRefused("The source ID must identify exactly one row in the active import.")
1918 return rows[0]
1921def _posted_row_number(request) -> int:
1922 """Return the posted source row number, or refuse the command."""
1923 try:
1924 return int(request.POST.get("row_number", ""))
1925 except (TypeError, ValueError):
1926 raise PreviewCommandRefused("A valid source row is required.") from None
1929@dataclass(frozen=True)
1930class _IgnoreDevice(PreviewCommand):
1931 """Add one source row of the preview to the profile's ignore list."""
1933 source_id: str
1934 device_name: str
1936 def apply(self, preview):
1937 """Store the ignore row, keeping one that already exists."""
1938 from .models import IgnoredDevice
1940 _source_row(preview.workspace, self.source_id)
1941 ignored = _get_or_init(IgnoredDevice, profile=preview.profile, source_id=self.source_id)
1942 if ignored.pk is None:
1943 ignored.device_name = self.device_name
1944 _validate_model_instance(ignored, f"ignored device '{self.source_id}'")
1945 save_permission_scoped_object(
1946 preview.actor,
1947 IgnoredDevice,
1948 {"profile": preview.profile, "source_id": self.source_id},
1949 {"device_name": self.device_name},
1950 on_existing="keep",
1951 )
1952 return CommandOutcome(message=f"Device '{self.device_name or self.source_id}' added to ignore list.")
1955@dataclass(frozen=True)
1956class _UnignoreDevice(PreviewCommand):
1957 """Remove one source row of the preview from the profile's ignore list."""
1959 source_id: str
1961 def apply(self, preview):
1962 """Delete the ignore row, refusing when there is none to delete."""
1963 from .models import IgnoredDevice
1965 _source_row(preview.workspace, self.source_id)
1966 rows = IgnoredDevice.objects.filter(profile=preview.profile, source_id=self.source_id)
1967 if not delete_permission_scoped_objects(preview.actor, rows):
1968 raise PreviewCommandRefused("Device was not on the ignore list (may be ignored by class mapping).")
1969 return CommandOutcome(message="Device removed from ignore list.")
1972class IgnoreDeviceView(_PreviewCommandMixin, PermissionRequiredMixin, View):
1973 """Mark a specific device (by source_id) as ignored for a profile."""
1975 permission_required = "netbox_data_import.change_importprofile"
1977 def post(self, request):
1978 """Add the specified device to the profile's ignore list."""
1979 command = _IgnoreDevice(
1980 source_id=source_text(request.POST.get("source_id")), device_name=request.POST.get("device_name", "")
1981 )
1982 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
1983 return _command_response(request, result, self.refusal_url(request))
1986class UnignoreDeviceView(_PreviewCommandMixin, PermissionRequiredMixin, View):
1987 """Remove a device from the ignore list."""
1989 permission_required = "netbox_data_import.change_importprofile"
1991 def post(self, request):
1992 """Remove the specified device from the profile's ignore list."""
1993 command = _UnignoreDevice(source_id=source_text(request.POST.get("source_id")))
1994 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
1995 return _command_response(request, result, self.refusal_url(request))
1998def _row_decision_source(workspace, row_number, source_id) -> dict:
1999 """Return the source row a row decision names, refusing one the active preview does not show."""
2000 rows = [
2001 row
2002 for row in workspace.source_rows
2003 if row.get("_row_number") == row_number and source_text(row.get("source_id")) == source_id
2004 ]
2005 if not source_id or len(rows) != 1:
2006 raise PreviewCommandRefused("The source ID and row must identify one active import row.")
2007 return rows[0]
2010def _device_unit(workspace, row_number):
2011 """Return the Device unit the preview shows on one row, for a field or placement command."""
2012 return next(
2013 (
2014 item
2015 for item in workspace.units
2016 if item.row_number == row_number and item.object_type == "device" and item.action in {"update", "error"}
2017 ),
2018 None,
2019 )
2022def _field_review_row(workspace, row_number, target_field):
2023 """Return the row a field-review command names, or refuse one the preview no longer shows."""
2024 row = _device_unit(workspace, row_number)
2025 if row is None or DeviceFieldReviewer.definition(target_field) is None or not source_text(row.source_id):
2026 raise StalePreview("The selected field difference is no longer current. Re-read the preview and try again.")
2027 return row
2030def _offered_difference(row, target_field) -> bool:
2031 """Return whether the preview offered this field for review.
2033 A field the import does not write is reported too, and an operator ignores it to stop
2034 the preview reporting it. Only a synced field has to be a writable difference.
2035 """
2036 return target_field in row.extra_data.get("field_diff", {}) or target_field in row.extra_data.get(
2037 "field_informational", {}
2038 )
2041def _field_baseline_error(row, device, target_field) -> str | None:
2042 """Return why *device* no longer holds the preview baseline of one field, or None when it does."""
2043 snapshots = row.extra_data["field_review_snapshots"][target_field]
2044 current = DeviceFieldReviewer.current_snapshot(device, target_field)
2045 if current is None or current.get("canonical") != snapshots.get("netbox", {}).get("canonical"):
2046 return "The matched NetBox value changed. Re-read the preview and try again."
2047 if target_field in {"u_position", "face"} and not _placement_matches_preview(device, row):
2048 return "The matched NetBox placement changed. Re-read the preview and try again."
2049 return None
2052def _placement_matches_preview(device, row) -> bool:
2053 """Return whether placement fields still match the materialized preview."""
2054 state = row.extra_data.get("_placement_state")
2055 # A baseline that states no location cannot prove the Device stayed in one, so it fails closed.
2056 if not isinstance(state, dict) or "location_id" not in state:
2057 return False
2058 return (
2059 device.location_id == state["location_id"]
2060 and device.rack_id == state.get("rack_id")
2061 and normalize_for_compare(device.position) == state.get("position", "")
2062 and (device.face or "") == state.get("face", "")
2063 )
2066def _locked_device_for_update(actor, device_pk):
2067 """Return the Device row locked and snapshotted for update, or None when it is gone or not permitted."""
2068 from dcim.models import Device
2070 # PostgreSQL refuses FOR UPDATE on a nullable outer join, so `of` locks the Device row alone.
2071 device = (
2072 Device.objects.restrict(actor, "change")
2073 .select_for_update(of=("self",))
2074 .select_related("site", "location", "rack", "device_type")
2075 .filter(pk=device_pk)
2076 .first()
2077 )
2078 if device is not None:
2079 device.snapshot()
2080 return device
2083def _field_review_binding(actor, profile, row, device) -> None:
2084 """Persist the confirmed source-to-device identity a field review rests on, or refuse."""
2085 allowed, error = _ensure_field_review_device_match(
2086 actor, profile, row.source_id, device, source_text(row.extra_data.get("asset_tag"))[:50]
2087 )
2088 if not allowed:
2089 raise PreviewCommandRefused(
2090 "Permission denied: cannot persist the source-to-device field-review match."
2091 if error == "permission"
2092 else "The source row or device is already linked elsewhere.",
2093 409,
2094 )
2097@dataclass(frozen=True)
2098class _IgnoreFieldDifference(PreviewCommand):
2099 """Ignore one exact current field difference of one matched Device."""
2101 row_number: int
2102 target_field: str
2104 def apply(self, preview):
2105 """Save the current snapshots of the difference the preview showed."""
2106 from dcim.models import Device
2108 row = _field_review_row(preview.workspace, self.row_number, self.target_field)
2109 if not _offered_difference(row, self.target_field):
2110 raise StalePreview("The selected field difference is no longer present. Re-read the preview.")
2111 snapshots = row.extra_data.get("field_review_snapshots", {}).get(self.target_field)
2112 device_id = row.extra_data.get("netbox_device_id")
2113 if not isinstance(snapshots, dict) or not device_id:
2114 raise PreviewCommandRefused("The selected field difference has no current matched device.", 409)
2115 device = Device.objects.restrict(preview.actor, "view").filter(pk=device_id).first()
2116 if device is None:
2117 raise PreviewCommandRefused("The matched NetBox device is no longer available.", 409)
2118 current = DeviceFieldReviewer.current_snapshot(device, self.target_field)
2119 if current is None or current.get("canonical") != snapshots.get("netbox", {}).get("canonical"):
2120 raise PreviewCommandRefused("The matched NetBox value changed. Re-read the preview and try again.", 409)
2121 _field_review_binding(preview.actor, preview.profile, row, device)
2122 try:
2123 save_permission_scoped_object(
2124 preview.actor,
2125 IgnoredFieldDifference,
2126 {
2127 "profile": preview.profile,
2128 "source_id": row.source_id,
2129 "netbox_device_id": device.pk,
2130 "target_field": self.target_field,
2131 },
2132 {"file_snapshot": snapshots.get("file", {}), "netbox_snapshot": snapshots.get("netbox", {})},
2133 )
2134 except ObjectPermissionDenied as exc:
2135 raise PreviewCommandRefused("Permission denied: cannot create or change this field review.", 409) from exc
2136 except ValidationError as exc:
2137 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
2138 return CommandOutcome(message=f"Ignored the current {self.target_field} difference.")
2141@dataclass(frozen=True)
2142class _UnignoreFieldDifference(PreviewCommand):
2143 """Remove one exact current field-difference review of one matched Device."""
2145 row_number: int
2146 target_field: str
2148 def apply(self, preview):
2149 """Delete only the review the preview showed."""
2150 from dcim.models import Device
2152 row = _field_review_row(preview.workspace, self.row_number, self.target_field)
2153 if self.target_field not in row.extra_data.get("field_ignored", {}):
2154 raise StalePreview("The selected field review is no longer current. Re-read the preview.")
2155 device = (
2156 Device.objects.restrict(preview.actor, "view").filter(pk=row.extra_data.get("netbox_device_id")).first()
2157 )
2158 if device is None:
2159 raise PreviewCommandRefused("The matched NetBox device is no longer available.", 409)
2160 record = (
2161 IgnoredFieldDifference.objects.select_for_update()
2162 .filter(
2163 profile=preview.profile,
2164 source_id=row.source_id,
2165 netbox_device_id=device.pk,
2166 target_field=self.target_field,
2167 )
2168 .first()
2169 )
2170 if record is None:
2171 raise StalePreview("The selected field review is no longer current. Re-read the preview.")
2172 if not preview.actor.has_perm("netbox_data_import.delete_ignoredfielddifference", record):
2173 raise PreviewCommandRefused("Permission denied: cannot remove this field review.", 409)
2174 _field_review_binding(preview.actor, preview.profile, row, device)
2175 record.delete()
2176 return CommandOutcome(message=f"Showing the {self.target_field} difference again.")
2179class IgnoreFieldDifferenceView(_PreviewCommandMixin, PermissionRequiredMixin, View):
2180 """Ignore one exact current field difference for a matched Device."""
2182 permission_required = "netbox_data_import.add_ignoredfielddifference"
2184 def post(self, request):
2185 """Save current snapshots from the active preview."""
2186 row_number = _posted_row_number(request)
2187 command = _IgnoreFieldDifference(
2188 row_number=row_number, target_field=request.POST.get("target_field", "").strip()
2189 )
2190 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
2191 return _command_response(request, result, self.refusal_url(request), row_number=row_number)
2194class UnignoreFieldDifferenceView(_PreviewCommandMixin, PermissionRequiredMixin, View):
2195 """Remove one exact current field-difference review for a matched Device."""
2197 permission_required = "netbox_data_import.delete_ignoredfielddifference"
2199 def post(self, request):
2200 """Delete only the review represented by the active preview."""
2201 row_number = _posted_row_number(request)
2202 command = _UnignoreFieldDifference(
2203 row_number=row_number, target_field=request.POST.get("target_field", "").strip()
2204 )
2205 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
2206 return _command_response(request, result, self.refusal_url(request), row_number=row_number)
2209class RemoveExtraIpView(PermissionRequiredMixin, View):
2210 """Remove one stored IP from a device's import record."""
2212 permission_required = "dcim.change_device"
2214 def post(self, request):
2215 """Remove an IP field from the device's import record."""
2216 from dcim.models import Device
2218 device_id = request.POST.get("device_id")
2219 ip_field = request.POST.get("ip_field")
2221 def _safe_return(device=None):
2222 url = request.POST.get("next", "")
2223 if url and url_has_allowed_host_and_scheme(
2224 url, allowed_hosts={request.get_host()}, require_https=request.is_secure()
2225 ):
2226 return redirect(url)
2227 if device:
2228 return redirect(device.get_absolute_url())
2229 return redirect("/")
2231 if not device_id or not ip_field:
2232 messages.error(request, "Missing device_id or ip_field.")
2233 return _safe_return()
2235 if ip_field not in ("primary_ip4", "primary_ip6", "oob_ip"):
2236 messages.error(request, f"Invalid ip_field: {ip_field}")
2237 return _safe_return()
2239 device = get_object_or_404(Device.objects.restrict(request.user, "change"), pk=device_id)
2240 import_source = stored_import_source(device)
2241 unassigned_ips = dict(import_source.unassigned_ips) if import_source is not None else {}
2243 if ip_field in unassigned_ips:
2244 del unassigned_ips[ip_field]
2245 import_source.unassigned_ips = unassigned_ips
2246 import_source.save(update_fields=["unassigned_ips"])
2247 messages.success(request, f"Removed {ip_field} from the import record.")
2248 else:
2249 messages.info(request, f"{ip_field} was not in the import record.")
2251 return _safe_return(device)
2254# ---------------------------------------------------------------------------
2255# Sync single device field from import file value
2256# ---------------------------------------------------------------------------
2259class _AjaxPermissionView(ConditionalLoginRequiredMixin, View):
2260 """Base for AJAX/JSON endpoints — inherits NetBox's ``ConditionalLoginRequiredMixin``.
2262 Subclasses set ``permission_required`` (a Django permission string) to gate
2263 access. Unauthenticated requests receive a JSON 401 (never a redirect, since
2264 these endpoints are called via ``fetch``). Authenticated users without the
2265 required permission receive a JSON 403. ``ConditionalLoginRequiredMixin`` is
2266 still inherited so that login redirects work if the request is ever reached
2267 via a browser navigation (e.g. direct URL), but the explicit checks above
2268 fire first for API callers.
2269 """
2271 permission_required: str | tuple[str, ...] | None = None
2272 permission_denied_response_format = "json"
2274 def dispatch(self, request, *args, **kwargs):
2275 from django.http import JsonResponse
2277 if not request.user.is_authenticated:
2278 return JsonResponse({"ok": False, "error": "Authentication required"}, status=401)
2279 if self.permission_required and not request.user.has_perm(self.permission_required):
2280 return JsonResponse({"ok": False, "error": "Permission denied"}, status=403)
2281 return super().dispatch(request, *args, **kwargs)
2284class ContactLookupView(_AjaxPermissionView):
2285 """Search visible NetBox Contacts for the contact-resolution picker."""
2287 permission_required = "tenancy.view_contact"
2289 def get(self, request):
2290 """Return a small real-shape Contact result set."""
2291 from django.db.models import Q
2292 from django.http import JsonResponse
2293 from tenancy.models import Contact
2295 query = request.GET.get("q", "").strip()
2296 if len(query) < 2:
2297 return JsonResponse({"results": []})
2298 contacts = (
2299 Contact.objects.restrict(request.user, "view")
2300 .filter(Q(name__icontains=query) | Q(email__icontains=query) | Q(phone__icontains=query))
2301 .order_by("name", "email", "pk")[:20]
2302 )
2303 return JsonResponse(
2304 {
2305 "results": [
2306 {
2307 "id": contact.pk,
2308 "name": contact.name,
2309 "email": contact.email,
2310 "phone": contact.phone,
2311 }
2312 for contact in contacts
2313 ]
2314 }
2315 )
2318class ContactSuggestionView(_AjaxPermissionView):
2319 """Return the Contact one preview row's candidate values identify, as it stands now."""
2321 permission_required = "tenancy.view_contact"
2323 def get(self, request):
2324 """Recompute one row's suggestion, so a Contact created on another row is offered here."""
2325 try:
2326 snapshot = read_preview(request, expected=PreviewClaim.posted(request.GET), profile_action="view")
2327 except StalePreview as exc:
2328 return _stale_json(exc)
2329 if not snapshot.active:
2330 return _stale_json(StalePreview("No current import preview matches this request."))
2331 # The open picker outlives an upgrade, so the stored profile can name a retired adapter.
2332 if reason := _unregistered_adapter_reason(snapshot.profile):
2333 return JsonResponse({"error": reason}, status=400)
2334 try:
2335 candidates, _source_row, _result_row = _contact_candidate_context(
2336 snapshot.workspace(request.user), request.GET.get("source_id", "")
2337 )
2338 except PlanError:
2339 return JsonResponse({"error": UNREADABLE_PREVIEW}, status=409)
2340 except ValidationError as exc:
2341 return JsonResponse({"error": "; ".join(exc.messages)}, status=400)
2342 return JsonResponse({"suggestion": PrimaryContactResolver.suggest(candidates, snapshot.profile, request.user)})
2345@dataclass(frozen=True)
2346class _SyncDeviceField(PreviewCommand):
2347 """Apply one previewed field value of one row to its matched Device."""
2349 row_number: int
2350 field: str
2352 def apply(self, preview):
2353 """Write the file value the preview showed onto the locked Device, after rechecking its baseline."""
2354 row = _device_unit(preview.workspace, self.row_number)
2355 device_id = row.extra_data.get("netbox_device_id") if row is not None else None
2356 if not device_id:
2357 raise StalePreview("The active preview row is no longer available.")
2358 if self.field not in row.extra_data.get("field_diff", {}):
2359 raise StalePreview("The selected field difference is no longer present.")
2360 snapshots = row.extra_data.get("field_review_snapshots", {}).get(self.field)
2361 if not isinstance(snapshots, dict):
2362 raise PreviewCommandRefused("The selected field has no authoritative preview value.", 409)
2363 device = _locked_device_for_update(preview.actor, device_id)
2364 if device is None:
2365 raise PreviewCommandRefused("Device not found", 409)
2366 if error := _field_baseline_error(row, device, self.field):
2367 raise PreviewCommandRefused(error, 409)
2368 value = snapshots.get("file", {}).get("canonical", "")
2369 try:
2370 display = SyncDeviceFieldView._apply_field(device, self.field, value, status_map(), preview.actor)
2371 except PreviewCommandRefused:
2372 raise
2373 except Exception as exc:
2374 logger.exception("SyncDeviceFieldView failed for device_id=%s field=%s", device.pk, self.field)
2375 raise PreviewCommandRefused("An internal error occurred.", 500) from exc
2376 return CommandOutcome(message=f"Updated {self.field} to {display}.")
2379class SyncDeviceFieldView(_PreviewCommandMixin, _AjaxPermissionView):
2380 """Apply a single field value from the import file to an existing NetBox device."""
2382 permission_required = "dcim.change_device"
2384 _IP_FIELDS = ("primary_ip4", "primary_ip6", "oob_ip")
2385 _ALLOWED_FIELDS = {"u_position", "status", "serial", "asset_tag", "face", "airflow", *_IP_FIELDS}
2387 def post(self, request):
2388 """Apply one previewed field value to its matched Device."""
2389 field = request.POST.get("field", "")
2390 if not field or field not in self._ALLOWED_FIELDS:
2391 return JsonResponse({"ok": False, "error": f"Field '{field}' is not syncable"}, status=400)
2392 row_number = _posted_row_number(request)
2393 command = _SyncDeviceField(row_number=row_number, field=field)
2394 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
2395 return _command_response(request, result, self.refusal_url(request), row_number=row_number)
2397 @staticmethod
2398 def _writer_safe_text(device, label, model_field, value):
2399 """Reject a value the writer would otherwise truncate away from what the preview showed."""
2400 text = str(value)
2401 limit = type(device)._meta.get_field(model_field).max_length
2402 if len(text) > limit:
2403 raise PreviewCommandRefused(f"The {label} is {len(text)} characters; NetBox allows {limit}.")
2404 return text
2406 @classmethod
2407 def _apply_field(cls, device, field, value, status_map, user):
2408 """Write one previewed value onto the locked and snapshotted device, through that field's own writer."""
2409 if field in cls._IP_FIELDS:
2410 return cls._apply_ip_field(device, field, value, user)
2411 writer = {
2412 "airflow": lambda: cls._apply_airflow(device, value),
2413 "u_position": lambda: cls._apply_u_position(device, value),
2414 "status": lambda: cls._apply_status(device, value, status_map),
2415 "serial": lambda: cls._apply_serial(device, value),
2416 "asset_tag": lambda: cls._apply_asset_tag(device, value),
2417 "face": lambda: cls._apply_face(device, value),
2418 }.get(field)
2419 if writer is None:
2420 raise PreviewCommandRefused(f"Field '{field}' is not syncable")
2421 return writer()
2423 @classmethod
2424 def _apply_u_position(cls, device, value):
2425 pos = source_position(value)
2426 if pos is None:
2427 raise PreviewCommandRefused(f"Cannot parse '{value}' as a finite number for u_position")
2428 zero_u_type = _zero_u_device_type(device)
2429 if zero_u_type:
2430 raise PreviewCommandRefused(f"Cannot set a rack position: the device type '{zero_u_type}' is 0U.")
2431 device.position = pos
2432 cls._reject_invalid_placement(device)
2433 device.save(update_fields=["position"])
2434 return f"U{device.position}"
2436 @staticmethod
2437 def _apply_status(device, value, status_map):
2438 text = str(value).strip().lower()
2439 # A NetBox status slug is accepted directly too (for example "active", "offline").
2440 mapped = status_map.get(text) or (text if text in set(status_map.values()) else None)
2441 if mapped is None:
2442 raise PreviewCommandRefused(f"Unknown status value '{value}'")
2443 device.status = mapped
2444 device.save(update_fields=["status"])
2445 return device.status
2447 @classmethod
2448 def _apply_serial(cls, device, value):
2449 device.serial = cls._writer_safe_text(device, "serial", "serial", value)
2450 device.save(update_fields=["serial"])
2451 return device.serial
2453 @classmethod
2454 def _apply_asset_tag(cls, device, value):
2455 device.asset_tag = cls._writer_safe_text(device, "asset tag", "asset_tag", value) if value else None
2456 device.save(update_fields=["asset_tag"])
2457 return device.asset_tag
2459 @classmethod
2460 def _apply_face(cls, device, value):
2461 if device.rack_id is None:
2462 raise PreviewCommandRefused(
2463 "Cannot set face: device has no rack assigned. Sync rack first, or use Sync Placement."
2464 )
2465 zero_u_type = _zero_u_device_type(device)
2466 if zero_u_type:
2467 raise PreviewCommandRefused(f"Cannot set a rack face: the device type '{zero_u_type}' is 0U.")
2468 mapped = _FACE_MAP.get(str(value).strip().lower())
2469 if mapped is None:
2470 raise PreviewCommandRefused(f"Unknown face value '{value}' — expected 'front' or 'rear'")
2471 device.face = mapped
2472 cls._reject_invalid_placement(device)
2473 device.save(update_fields=["face"])
2474 return device.face
2476 @staticmethod
2477 def _apply_airflow(device, value):
2478 """Write the airflow the source row states, in the wording the importer already reads."""
2479 _side, airflow_map, _status = translation_maps()
2480 text = str(value).strip().lower()
2481 mapped = airflow_map.get(text)
2482 # The stored value is also accepted, so a row already carrying one syncs as it stands.
2483 if mapped is None and text in set(airflow_map.values()):
2484 mapped = text
2485 if mapped is None:
2486 raise PreviewCommandRefused(f"Unknown airflow value '{value}'")
2487 device.airflow = mapped
2488 device.save(update_fields=["airflow"])
2489 return device.airflow
2491 @classmethod
2492 def _apply_ip_field(cls, device, field, value, user):
2493 """Point one of the device's IP fields at the address the source row carries."""
2494 try:
2495 target = ip_assignment.resolve(device, field, value)
2496 except ip_assignment.IPAssignmentError as exc:
2497 raise PreviewCommandRefused(exc.operator_message) from exc
2499 if target.already_held:
2500 # The device carries it already, so only the field moves. No IPAM row is written.
2501 held = target.held
2502 if getattr(device, f"{field}_id", None) != held.pk:
2503 setattr(device, field, held)
2504 device.save(update_fields=[field])
2505 return target.summary
2507 try:
2508 address = ip_assignment.apply(target, user)
2509 except ValidationError as exc:
2510 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
2511 except ObjectPermissionDenied as exc:
2512 raise PreviewCommandRefused("Permission denied: cannot assign this IP address.") from exc
2513 setattr(device, field, address)
2514 device.save(update_fields=[field])
2515 return f"{address.address} on {target.interface.name}"
2517 @staticmethod
2518 def _reject_invalid_placement(device) -> None:
2519 """Reject a placement value NetBox would refuse, before it reaches an unvalidated save."""
2520 try:
2521 _validate_device_placement(device)
2522 except ValidationError as exc:
2523 raise PreviewCommandRefused(_placement_error_text(exc)) from exc
2526def _lookup_rack_for_device(actor, device, value):
2527 """Look up a Rack by name within ``device.site``, honoring ``device.location``.
2529 If the device has a location set, the rack must be in the same location. If the
2530 device has no location, the rack must also have no location (the implicit
2531 "default location" semantic).
2533 Returns ``(rack, None)`` on success or ``(None, error_message)`` on failure.
2534 Error messages are static, controlled strings — no exception text is exposed,
2535 so the result is safe to return directly to the client.
2536 """
2537 from dcim.models import Rack
2539 name = (str(value) if value is not None else "").strip()
2540 if not name:
2541 return None, "Rack name is empty"
2542 if device.site_id is None:
2543 return None, "Device has no site; cannot resolve rack"
2544 qs = Rack.objects.restrict(actor, "view").filter(identity_in("name", [identity_text(name)]), site=device.site)
2545 if device.location_id is not None:
2546 qs = qs.filter(location=device.location)
2547 loc_str = f" / location '{device.location}'"
2548 else:
2549 qs = qs.filter(location__isnull=True)
2550 loc_str = ""
2551 racks = list(qs[:2])
2552 if not racks:
2553 return None, f"Rack '{name}' not found in site '{device.site}'{loc_str}"
2554 if len(racks) > 1:
2555 return None, f"Multiple racks named '{name}' found; cannot disambiguate"
2556 return racks[0], None
2559def _validate_device_placement(device) -> None:
2560 """Run NetBox validation and reject only errors caused by placement fields."""
2561 try:
2562 device.full_clean()
2563 except ValidationError as exc:
2564 if not hasattr(exc, "message_dict"):
2565 raise
2566 placement_fields = {"rack", "location", "position", "face", "device_type", "__all__"}
2567 errors = {field: messages for field, messages in exc.message_dict.items() if field in placement_fields}
2568 if errors:
2569 raise ValidationError(errors) from exc
2572_FACE_MAP = {"front": "front", "rear": "rear", "0": "front", "1": "rear"}
2575def _placement_error_text(exc) -> str:
2576 """Return one readable line for a placement ValidationError."""
2577 if hasattr(exc, "message_dict"):
2578 return "; ".join(f"{name}: {', '.join(messages)}" for name, messages in exc.message_dict.items())
2579 return "; ".join(exc.messages)
2582def _zero_u_device_type(device) -> str:
2583 """Return the device type label when it is zero-U, which takes no position or face."""
2584 device_type = device.device_type
2585 if device_type is not None and device_type.u_height == 0:
2586 return str(device_type)
2587 return ""
2590def _set_rack_placement(device, u_position, face, zero_u_type):
2591 """Set the rack position and face on *device*.
2593 Returns the written field names, the field names a zero-U device type cannot take,
2594 and one error message for a value the writer cannot accept.
2595 """
2596 update_fields = []
2597 skipped = []
2599 if zero_u_type:
2600 # Clear a stored position the way the import writer does, so the device stays valid.
2601 if device.position is not None:
2602 device.position = None
2603 update_fields.append("position")
2604 if device.face:
2605 device.face = None
2606 update_fields.append("face")
2607 if u_position not in ("", None):
2608 skipped.append("position")
2609 if face not in ("", None):
2610 skipped.append("face")
2611 return update_fields, skipped, None
2613 if u_position not in ("", None):
2614 position = source_position(u_position)
2615 if position is None:
2616 return update_fields, skipped, f"Cannot parse '{u_position}' as a finite number for u_position"
2617 device.position = position
2618 update_fields.append("position")
2620 if face not in ("", None):
2621 mapped = _FACE_MAP.get(str(face).strip().lower())
2622 if mapped is None:
2623 return update_fields, skipped, f"Unknown face value '{face}' — expected 'front' or 'rear'"
2624 device.face = mapped
2625 update_fields.append("face")
2627 return update_fields, skipped, None
2630@dataclass(frozen=True)
2631class _SyncPlacement(PreviewCommand):
2632 """Apply one row's previewed rack, position and face to its matched Device, all or nothing."""
2634 row_number: int
2636 def apply(self, preview):
2637 """Recheck the placement baseline under the Device lock, then write the previewed placement."""
2638 row = _device_unit(preview.workspace, self.row_number)
2639 device_id = row.extra_data.get("netbox_device_id") if row is not None else None
2640 if not device_id:
2641 raise StalePreview("The active preview row is no longer available.")
2642 device = _locked_device_for_update(preview.actor, device_id)
2643 if device is None:
2644 raise PreviewCommandRefused("Device not found", 409)
2645 if not _placement_matches_preview(device, row):
2646 raise PreviewCommandRefused("The matched NetBox placement changed. Re-read the preview and try again.", 409)
2647 rack, error = _lookup_rack_for_device(preview.actor, device, row.rack_name)
2648 if error:
2649 raise PreviewCommandRefused(error)
2650 device.rack = rack
2651 # NetBox rejects a rack position on a zero-U device type, so sync the rack alone.
2652 zero_u_type = _zero_u_device_type(device)
2653 placement_fields, skipped, error = _set_rack_placement(
2654 device, row.extra_data.get("u_position", ""), row.extra_data.get("face", ""), zero_u_type
2655 )
2656 if error:
2657 raise PreviewCommandRefused(error)
2658 update_fields = ["rack", *placement_fields]
2659 try:
2660 _validate_device_placement(device)
2661 except ValidationError as exc:
2662 raise PreviewCommandRefused(f"Validation failed: {_placement_error_text(exc)}") from exc
2663 try:
2664 device.save(update_fields=update_fields)
2665 except Exception as exc:
2666 logger.exception("SyncPlacementView save failed for device_id=%s", device.pk)
2667 raise PreviewCommandRefused("An internal error occurred.", 500) from exc
2668 parts = [f"rack={rack.name}"]
2669 if "position" in update_fields and device.position is not None:
2670 parts.append(f"U{device.position}")
2671 if "face" in update_fields and device.face:
2672 parts.append(device.face)
2673 display = ", ".join(parts)
2674 if skipped:
2675 display += f" (0U device type {zero_u_type} takes no {' or '.join(skipped)})"
2676 return CommandOutcome(message=f"Updated placement to {display}.")
2679class SyncPlacementView(_PreviewCommandMixin, _AjaxPermissionView):
2680 """Atomically sync rack + (optional) u_position + (optional) face for a device.
2682 All-or-nothing: if the rack lookup fails, nothing is saved.
2683 """
2685 permission_required = "dcim.change_device"
2687 def post(self, request):
2688 """Apply the previewed placement to its matched Device."""
2689 row_number = _posted_row_number(request)
2690 result = apply_preview_command(request, PreviewClaim.posted(request.POST), _SyncPlacement(row_number))
2691 return _command_response(request, result, self.refusal_url(request), row_number=row_number)
2694# ---------------------------------------------------------------------------
2695# Save resolution (rerere)
2696# ---------------------------------------------------------------------------
2699def _device_name_already_claimed(effective_rows, row_number, new_name, target):
2700 """Return why this device name is unavailable at the import target, or None when it is free."""
2701 from dcim.models import Device
2703 other_names = {
2704 identity_text(device_name)
2705 for row in effective_rows
2706 if row.get("_row_number") != row_number and (device_name := effective_device_name(row))
2707 }
2708 if identity_text(new_name) in other_names:
2709 return f"Device name '{new_name}' is already used by another source row."
2710 tenant = target["tenant"]
2711 tenant_filter = {"tenant": tenant} if tenant is not None else {"tenant__isnull": True}
2712 if (
2713 Device.objects.filter(site=target["site"], **tenant_filter)
2714 .filter(identity_in("name", [identity_text(new_name)]))
2715 .exists()
2716 ):
2717 return f"Device name '{new_name}' already exists at the active import site."
2718 return None
2721class _RowDecisionView(_PreviewCommandMixin, PermissionRequiredMixin, View):
2722 """A preview row decision: a refused one renders the preview in place for HTMX, as a saved one does."""
2724 permission_required = "netbox_data_import.change_importprofile"
2726 def _refusal(self, request, error, status, url=None):
2727 if self._answers_json(request):
2728 return JsonResponse({"ok": False, "error": error}, status=status)
2729 messages.error(request, error)
2730 return _name_resolution_response(request, url or self.refusal_url(request))
2732 def decide(self, request, command):
2733 """Run one row decision and answer with the preview it leaves behind."""
2734 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
2735 messages.success(request, result.outcome.message)
2736 return _name_resolution_response(request, self.refusal_url(request))
2739@dataclass(frozen=True)
2740class _ResolveDuplicateName(PreviewCommand):
2741 """Save a unique device name for one duplicate source row."""
2743 row_number: int
2744 source_id: str
2745 new_name: str
2747 def apply(self, preview):
2748 """Check the name against the other rows and the import target, then save it."""
2749 source_row = _row_decision_source(preview.workspace, self.row_number, self.source_id)
2750 target = _resolved_import_target(preview.planning_context, preview.actor)
2751 if target is None:
2752 raise PreviewCommandRefused(TARGET_GONE, 409)
2753 # The claims are read under the locks, so two rows cannot both resolve to one device name.
2754 refusal = _device_name_already_claimed(preview.workspace.source_rows, self.row_number, self.new_name, target)
2755 if refusal is not None:
2756 raise PreviewCommandRefused(refusal)
2757 try:
2758 save_permission_scoped_object(
2759 preview.actor,
2760 SourceResolution,
2761 {"profile": preview.profile, "source_id": self.source_id, "source_column": "device_name"},
2762 {
2763 "original_value": source_text(source_row.get("device_name")),
2764 "resolved_fields": {"device_name": self.new_name},
2765 },
2766 )
2767 except ObjectPermissionDenied as exc:
2768 raise PreviewCommandRefused("Permission denied: cannot create or change this saved name.", 403) from exc
2769 except ValidationError as exc:
2770 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
2771 return CommandOutcome(message=f"Source '{self.source_id}' will use device name '{self.new_name}'.")
2774class ResolveDuplicateNameView(_RowDecisionView):
2775 """Save a unique device name for one duplicate source row."""
2777 def post(self, request):
2778 """Validate and persist the replacement device name."""
2779 new_name = request.POST.get("new_name", "").strip()
2780 if not new_name or len(new_name) > 64:
2781 raise PreviewCommandRefused("The device name must contain 1 to 64 characters.")
2782 command = _ResolveDuplicateName(
2783 row_number=_posted_row_number(request),
2784 source_id=source_text(request.POST.get("source_id")),
2785 new_name=new_name,
2786 )
2787 return self.decide(request, command)
2790def _duplicate_serial_shown(preview_rows, row_number) -> str:
2791 """Return the serial the engine calls a duplicate on this source row, or an empty string."""
2792 for item in preview_rows:
2793 if (
2794 item.row_number == row_number
2795 and item.object_type == "device"
2796 and item.extra_data.get("identity_conflict") == "duplicate_serial"
2797 ):
2798 return source_text(item.extra_data.get("duplicate_serial"))
2799 return ""
2802@dataclass(frozen=True)
2803class _IgnoreDuplicateSerial(PreviewCommand):
2804 """Drop the serial from one source row so the rows sharing it stop colliding."""
2806 row_number: int
2807 source_id: str
2809 def apply(self, preview):
2810 """Settle the collision the operator was shown, and only while live NetBox still has it."""
2811 source_row = _row_decision_source(preview.workspace, self.row_number, self.source_id)
2812 shown_serial = _duplicate_serial_shown(preview.workspace.units, self.row_number)
2813 if not shown_serial:
2814 raise PreviewCommandRefused("This row shows no duplicate serial in the current preview.")
2815 original_serial = source_text(source_row.get("serial"))
2816 if not original_serial:
2817 raise PreviewCommandRefused("This row carries no serial to give up.")
2818 held_since = time.monotonic()
2819 current = ReviewWorkspace(
2820 ImportEngine.plan(preview.profile, preview.document, preview.actor, preview.planning_context),
2821 preview.actor,
2822 )
2823 # The dry run costs more as the file grows, and the import worker waits behind it.
2824 logger.info(
2825 "IgnoreDuplicateSerialView: held the profile policy lock for %.2fs over %d source rows.",
2826 time.monotonic() - held_since,
2827 len(preview.workspace.source_rows),
2828 )
2829 if _duplicate_serial_shown(current.units, self.row_number) != shown_serial:
2830 raise PreviewCommandRefused(f"No other row this import creates still claims serial '{shown_serial}'.")
2831 try:
2832 save_permission_scoped_object(
2833 preview.actor,
2834 SourceResolution,
2835 {"profile": preview.profile, "source_id": self.source_id, "source_column": "serial"},
2836 {"original_value": original_serial, "resolved_fields": {"serial": ""}},
2837 )
2838 except ObjectPermissionDenied as exc:
2839 raise PreviewCommandRefused("Permission denied: cannot create or change this saved serial.", 403) from exc
2840 except ValidationError as exc:
2841 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
2842 return CommandOutcome(message=f"Source '{self.source_id}' will import without serial '{shown_serial}'.")
2845class IgnoreDuplicateSerialView(_RowDecisionView):
2846 """Drop the serial from one source row so the rows sharing it stop colliding."""
2848 def post(self, request):
2849 """Persist an empty serial for the row the operator gives it up on."""
2850 command = _IgnoreDuplicateSerial(
2851 row_number=_posted_row_number(request), source_id=source_text(request.POST.get("source_id"))
2852 )
2853 return self.decide(request, command)
2856def _position_conflict_rows(units) -> set[int]:
2857 """Return every row number a rack-position collision names in the current preview."""
2858 involved: set[int] = set()
2859 for unit in units:
2860 if "rack_position_occupied" not in (unit.extra_data.get("identity_conflicts") or ()):
2861 continue
2862 involved.add(unit.row_number)
2863 for key in ("claimed_by_row", "conflict_row_number"):
2864 if (other := unit.extra_data.get(key)) is not None:
2865 involved.add(other)
2866 return involved
2869@dataclass(frozen=True)
2870class _IgnorePosition(PreviewCommand):
2871 """Drop the rack position from one source row so the rows sharing the unit stop colliding."""
2873 row_number: int
2874 source_id: str
2876 def apply(self, preview):
2877 """Save an empty rack position for a row of a collision the preview shows."""
2878 source_row = _row_decision_source(preview.workspace, self.row_number, self.source_id)
2879 # Either row of the collision may give its position up, so the whole group is eligible.
2880 if self.row_number not in _position_conflict_rows(preview.workspace.units):
2881 raise PreviewCommandRefused("This row is not part of a rack position conflict in the current preview.")
2882 original_position = source_text(source_row.get("u_position"))
2883 if not original_position:
2884 raise PreviewCommandRefused("This row carries no rack position to give up.")
2885 try:
2886 save_permission_scoped_object(
2887 preview.actor,
2888 SourceResolution,
2889 {"profile": preview.profile, "source_id": self.source_id, "source_column": "u_position"},
2890 {"original_value": original_position, "resolved_fields": {"u_position": None}},
2891 )
2892 except ObjectPermissionDenied as exc:
2893 raise PreviewCommandRefused(
2894 "Permission denied: cannot create or change this saved rack position.", 403
2895 ) from exc
2896 except ValidationError as exc:
2897 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
2898 return CommandOutcome(
2899 message=f"Source '{self.source_id}' will import into its rack without position U{original_position}."
2900 )
2903class IgnorePositionView(_RowDecisionView):
2904 """Drop the rack position from one source row so the rows sharing the unit stop colliding."""
2906 def post(self, request):
2907 """Persist an empty rack position for the row the operator unplaces."""
2908 command = _IgnorePosition(
2909 row_number=_posted_row_number(request), source_id=source_text(request.POST.get("source_id"))
2910 )
2911 return self.decide(request, command)
2914# The split modal sends each part to one of these Target Fields, and a part can replace the row's value.
2915SPLIT_TARGET_FIELDS = ("device_name", "asset_tag", "serial", "make", "model", "rack_name")
2916# A serial is compared exactly; every other split target compares by name identity.
2917EXACT_SPLIT_FIELDS = frozenset({"serial"})
2918ACKNOWLEDGEMENT_INVALID = "Acknowledged fields must be a JSON list of field names."
2921def _split_field_values(unit) -> dict[str, str]:
2922 """Return the value one preview row carries for each split target, which a split part can replace."""
2923 return {
2924 "device_name": unit.name or "",
2925 "asset_tag": unit.extra_data.get("asset_tag", ""),
2926 "serial": unit.extra_data.get("source_serial", ""),
2927 "make": unit.extra_data.get("source_make", ""),
2928 "model": unit.extra_data.get("source_model", ""),
2929 "rack_name": unit.rack_name or "",
2930 "source_id": unit.source_id,
2931 }
2934def _unacknowledged_replacement(workspace, source_id, source_column, resolved_fields, acknowledged) -> str:
2935 """Return why a split replaces a value the current preview row carries without the operator's acknowledgement.
2937 The row is read from the plan at the claimed revision, so a second save compares against the first.
2938 """
2939 if source_column not in SPLIT_TARGET_FIELDS:
2940 return ""
2941 rows = [unit for unit in workspace.units if unit.object_type == "device" and str(unit.source_id) == str(source_id)]
2942 carried = _split_field_values(rows[0]) if len(rows) == 1 else {}
2943 for field, value in resolved_fields.items():
2944 if field == source_column or field not in SPLIT_TARGET_FIELDS or field in acknowledged:
2945 continue
2946 existing, replacement = source_text(carried.get(field)), source_text(value)
2947 if not existing:
2948 continue
2949 same = (
2950 existing == replacement
2951 if field in EXACT_SPLIT_FIELDS
2952 else identity_text(existing) == identity_text(replacement)
2953 )
2954 if not same:
2955 return (
2956 f"The split replaces the {CATALOG.display(field)} '{existing}' with '{replacement}'. "
2957 "Acknowledge the replacement to save it."
2958 )
2959 return ""
2962def _posted_acknowledgement(request) -> tuple[str, ...]:
2963 """Return the field names the operator acknowledged a split replaces, or refuse a malformed list."""
2964 import json
2966 try:
2967 acknowledged = json.loads(request.POST.get("acknowledged_fields") or "[]")
2968 except json.JSONDecodeError:
2969 raise PreviewCommandRefused(ACKNOWLEDGEMENT_INVALID) from None
2970 if not isinstance(acknowledged, list) or not all(isinstance(field, str) for field in acknowledged):
2971 raise PreviewCommandRefused(ACKNOWLEDGEMENT_INVALID)
2972 return tuple(acknowledged)
2975@dataclass(frozen=True)
2976class _SaveResolution(PreviewCommand):
2977 """Save one manual field resolution of one preview row for rerere replay."""
2979 source_id: str
2980 source_column: str
2981 original_value: str
2982 resolved_fields: Mapping
2983 acknowledged: tuple[str, ...]
2985 def apply(self, preview):
2986 """Validate the decision against the current row, write the Contact it names, then save it."""
2987 import json
2989 workspace = preview.workspace
2990 _source_row(workspace, self.source_id)
2991 refusal = _unacknowledged_replacement(
2992 workspace, self.source_id, self.source_column, self.resolved_fields, self.acknowledged
2993 )
2994 if refusal:
2995 raise PreviewCommandRefused(refusal)
2996 contact_context = None
2997 candidates = {}
2998 original_value = self.original_value
2999 try:
3000 if self.source_column == "candidate:contact":
3001 candidates, source_row, result_row = _contact_candidate_context(workspace, self.source_id)
3002 validate_contact_candidate_resolution(
3003 self.resolved_fields, preview.profile.adapter_settings.primary_contact_lookup_field, candidates
3004 )
3005 original_value = json.dumps(candidates, sort_keys=True)
3006 contact_context = (source_row, result_row)
3007 validate_source_resolution_fields(preview.profile, self.source_column, self.resolved_fields)
3008 decision = _persist_contact_decision(
3009 preview.profile, self.resolved_fields, candidates, contact_context, preview.actor
3010 )
3011 save_permission_scoped_object(
3012 preview.actor,
3013 SourceResolution,
3014 {"profile": preview.profile, "source_id": self.source_id, "source_column": self.source_column},
3015 {"original_value": original_value or "", "resolved_fields": decision.resolved_fields},
3016 )
3017 except ValidationError as exc:
3018 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
3019 message, detail = _saved_resolution_report(decision.write, decision.note)
3020 return CommandOutcome(
3021 message=message,
3022 payload={
3023 "row_number": contact_context[1].row_number if contact_context else None,
3024 "detail": detail,
3025 # The page keeps its own copy of the decision, so it is given the saved one.
3026 "resolution": {
3027 "original_value": original_value or "",
3028 "resolved_fields": decision.resolved_fields,
3029 "contact": decision.contact,
3030 },
3031 },
3032 )
3035class SaveResolutionView(_PreviewCommandMixin, _AjaxPermissionView):
3036 """Save a manual field resolution for rerere replay."""
3038 permission_required = "netbox_data_import.change_importprofile"
3040 def post(self, request):
3041 """Persist a manual field resolution of the active preview."""
3042 import json
3044 source_id = source_text(request.POST.get("source_id"))
3045 source_column = request.POST.get("source_column")
3046 try:
3047 resolved_fields = json.loads(request.POST.get("resolved_fields", "{}"))
3048 except (json.JSONDecodeError, TypeError):
3049 resolved_fields = {}
3050 if not isinstance(resolved_fields, Mapping):
3051 raise PreviewCommandRefused("Resolved fields must be a JSON object.")
3052 if not source_id or not source_column:
3053 raise PreviewCommandRefused("A source row and column are required.")
3054 command = _SaveResolution(
3055 source_id=source_id,
3056 source_column=source_column,
3057 original_value=request.POST.get("original_value") or "",
3058 resolved_fields=resolved_fields,
3059 acknowledged=_posted_acknowledgement(request),
3060 )
3061 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
3062 payload = result.outcome.payload
3063 return _command_response(
3064 request,
3065 result,
3066 self.refusal_url(request),
3067 row_number=payload["row_number"],
3068 detail=payload["detail"],
3069 resolution=payload["resolution"],
3070 )
3073# ---------------------------------------------------------------------------
3074# Device type analysis view
3075# ---------------------------------------------------------------------------
3078class DeviceTypeAnalysisView(PermissionRequiredMixin, View):
3079 """Show all unique (make, model) pairs across import jobs and profiles.
3081 Highlights which ones have explicit DeviceTypeMapping vs auto-slugified.
3082 """
3084 permission_required = "netbox_data_import.view_importprofile"
3086 def get(self, request, profile_pk=None):
3087 """Render the device type analysis page for the given profile."""
3088 profile = get_object_or_404(ImportProfile, pk=profile_pk) if profile_pk else None
3089 profiles = ImportProfile.objects.all()
3091 # Build analysis from DeviceTypeMapping + auto-slugify check
3092 if profile:
3093 dt_mappings = DeviceTypeMapping.objects.filter(profile=profile)
3094 else:
3095 dt_mappings = DeviceTypeMapping.objects.select_related("profile").all()
3097 # Collect entries: explicit mappings
3098 entries = []
3099 for dtm in dt_mappings:
3100 entries.append(
3101 {
3102 "profile": dtm.profile,
3103 "source_make": dtm.source_make,
3104 "source_model": dtm.source_model,
3105 "manufacturer_slug": dtm.netbox_manufacturer_slug,
3106 "device_type_slug": dtm.netbox_device_type_slug,
3107 "mapping_type": "explicit",
3108 "mapping_pk": dtm.pk,
3109 }
3110 )
3112 # Check which mapped device types exist in NetBox
3113 from dcim.models import DeviceType
3115 for entry in entries:
3116 entry["exists_in_netbox"] = DeviceType.objects.filter(
3117 manufacturer__slug=entry["manufacturer_slug"],
3118 slug=entry["device_type_slug"],
3119 ).exists()
3121 return render(
3122 request,
3123 "netbox_data_import/analysis.html",
3124 {
3125 "profile": profile,
3126 "profiles": profiles,
3127 "entries": entries,
3128 },
3129 )
3132# ---------------------------------------------------------------------------
3133# Bulk YAML import for mappings
3134# ---------------------------------------------------------------------------
3137def _bulk_row_refusal(item, required: tuple[str, ...]) -> str:
3138 """Return why one bulk YAML mapping row cannot be imported, or an empty string when it can."""
3139 if not isinstance(item, dict):
3140 return "A row is not a mapping."
3141 missing = [key for key in required if key not in item]
3142 return f"A row is missing the key(s): {', '.join(missing)}." if missing else ""
3145class BulkYamlImportView(PermissionRequiredMixin, View):
3146 """Accept a YAML file and bulk-create ClassRoleMappings or DeviceTypeMappings for a profile.
3148 Useful for bootstrapping from contrib/ definition files.
3149 """
3151 permission_required = "netbox_data_import.change_importprofile"
3153 def get(self, request, profile_pk):
3154 """Render the bulk YAML import form."""
3155 profile = get_object_or_404(ImportProfile, pk=profile_pk)
3156 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})
3158 def _import_class_role_rows(self, data, profile, errors):
3159 """Import a list of class-role mapping items; return (created, skipped)."""
3160 created = skipped = 0
3161 for item in data:
3162 if refusal := _bulk_row_refusal(item, ("source_class",)):
3163 errors.append(refusal)
3164 continue
3165 try:
3166 rack_type = None
3167 rack_type_present = "rack_type" in item
3168 rack_type_slug = item.get("rack_type") if rack_type_present else None
3169 if rack_type_slug:
3170 from dcim.models import RackType
3172 try:
3173 rack_type = RackType.objects.get(slug=rack_type_slug)
3174 except RackType.DoesNotExist:
3175 errors.append(
3176 f"RackType with slug '{rack_type_slug}' not found for source_class '{item.get('source_class')}'"
3177 )
3178 continue
3180 defaults = {
3181 "creates_rack": item.get("creates_rack", False),
3182 "role_slug": item.get("role_slug", ""),
3183 "ignore": item.get("ignore", False),
3184 }
3185 if rack_type_present:
3186 defaults["rack_type"] = rack_type
3188 obj, was_created = ClassRoleMapping.objects.get_or_create(
3189 profile=profile,
3190 source_class=item["source_class"],
3191 defaults=defaults,
3192 )
3193 if not was_created and rack_type_present:
3194 obj.rack_type = rack_type
3195 obj.save(update_fields=["rack_type"])
3196 if was_created:
3197 created += 1
3198 else:
3199 skipped += 1
3200 except Exception:
3201 logger.exception("BulkYamlImportView class_role row failed for profile_id=%s", profile.pk)
3202 errors.append("A row failed due to an unexpected error — see server logs.")
3203 return created, skipped
3205 def _import_device_type_rows(self, data, profile, errors):
3206 """Import a list of device-type mapping items; return (created, skipped)."""
3207 created = skipped = 0
3208 required = ("source_make", "source_model", "netbox_manufacturer_slug", "netbox_device_type_slug")
3209 for item in data:
3210 if refusal := _bulk_row_refusal(item, required):
3211 errors.append(refusal)
3212 continue
3213 try:
3214 _, was_created = DeviceTypeMapping.objects.get_or_create(
3215 profile=profile,
3216 source_make=item["source_make"],
3217 source_model=item["source_model"],
3218 defaults={
3219 "netbox_manufacturer_slug": item["netbox_manufacturer_slug"],
3220 "netbox_device_type_slug": item["netbox_device_type_slug"],
3221 },
3222 )
3223 if was_created:
3224 created += 1
3225 else:
3226 skipped += 1
3227 except Exception:
3228 logger.exception("BulkYamlImportView device_type row failed for profile_id=%s", profile.pk)
3229 errors.append("A row failed due to an unexpected error — see server logs.")
3230 return created, skipped
3232 def post(self, request, profile_pk):
3233 """Parse the uploaded YAML file and create mappings in bulk."""
3234 profile = get_object_or_404(ImportProfile, pk=profile_pk)
3235 yaml_file = request.FILES.get("yaml_file")
3236 mapping_type = request.POST.get("mapping_type", "class_role")
3238 if not yaml_file:
3239 messages.error(request, "No YAML file uploaded.")
3240 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})
3242 try:
3243 import yaml
3245 data = load_yaml_document(yaml_file.read())
3246 except yaml.YAMLError as exc:
3247 messages.error(request, f"Failed to parse YAML: {exc}")
3248 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})
3249 except Exception:
3250 logger.exception("BulkYamlImportView: failed to read uploaded file for profile_id=%s", profile_pk)
3251 messages.error(request, "Could not read the uploaded file.")
3252 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})
3254 if not isinstance(data, list):
3255 messages.error(request, "YAML must be a list of mapping objects.")
3256 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})
3258 errors = []
3259 if mapping_type == "class_role":
3260 created, skipped = self._import_class_role_rows(data, profile, errors)
3261 elif mapping_type == "device_type":
3262 created, skipped = self._import_device_type_rows(data, profile, errors)
3263 else:
3264 messages.error(request, f"Unknown mapping type '{mapping_type}'.")
3265 return redirect(profile.get_absolute_url())
3267 if errors:
3268 messages.warning(
3269 request, f"Created {created}, skipped {skipped}, {len(errors)} errors: {'; '.join(errors[:3])}"
3270 )
3271 else:
3272 messages.success(request, f"Bulk import complete: {created} created, {skipped} already existed.")
3273 return redirect(profile.get_absolute_url())
3276# ---------------------------------------------------------------------------
3277# Profile YAML export / full-profile YAML import
3278# ---------------------------------------------------------------------------
3281class ExportProfileYamlView(PermissionRequiredMixin, View):
3282 """Download all profile configuration as a single YAML file."""
3284 permission_required = "netbox_data_import.view_importprofile"
3286 def get(self, request, pk):
3287 """Serialize the profile and all its mappings to YAML and return as a file download."""
3288 import yaml
3289 from django.http import HttpResponse
3291 profile = get_object_or_404(ImportProfile.objects.restrict(request.user, "view"), pk=pk)
3293 try:
3294 data = serialize_profile(profile, request.user)
3295 except ObjectPermissionDenied as exc:
3296 raise PermissionDenied from exc
3298 yaml_str = yaml.dump(data, allow_unicode=True, default_flow_style=False, sort_keys=False)
3299 safe_name = profile.name.lower().replace(" ", "_").replace("/", "-")
3300 filename = f"profile_{safe_name}.yaml"
3301 return HttpResponse(
3302 yaml_str,
3303 content_type="application/x-yaml",
3304 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
3305 )
3308class ImportProfileYamlView(PermissionRequiredMixin, View):
3309 """Import a full profile YAML (as exported by ExportProfileYamlView).
3311 If the profile already exists (by name), merges/updates its mappings.
3312 """
3314 permission_required = "netbox_data_import.change_importprofile"
3316 def has_permission(self):
3317 """Allow the page when the actor can create or update at least one profile."""
3318 return any(
3319 self.request.user.has_perm(permission)
3320 for permission in (
3321 "netbox_data_import.add_importprofile",
3322 "netbox_data_import.change_importprofile",
3323 )
3324 )
3326 def get(self, request):
3327 """Render the profile YAML import form."""
3328 return render(request, "netbox_data_import/import_profile_yaml.html")
3330 def post(self, request):
3331 """Parse the uploaded YAML and create or update the profile and its mappings."""
3332 import yaml
3334 yaml_file = request.FILES.get("yaml_file")
3335 if not yaml_file:
3336 messages.error(request, "No YAML file uploaded.")
3337 return render(request, "netbox_data_import/import_profile_yaml.html")
3339 try:
3340 data = load_yaml_document(yaml_file.read())
3341 except yaml.YAMLError as exc:
3342 messages.error(request, f"Failed to parse YAML: {exc}")
3343 return render(request, "netbox_data_import/import_profile_yaml.html")
3344 except OSError:
3345 logger.warning("ImportProfileYamlView: the uploaded file cannot be read.", exc_info=True)
3346 messages.error(request, UPLOAD_UNREADABLE)
3347 return render(request, "netbox_data_import/import_profile_yaml.html")
3349 try:
3350 profile, stats = apply_profile_document(data, request.user)
3351 except ObjectPermissionDenied as exc:
3352 raise PermissionDenied from exc
3353 except ProfileDocumentInvalid as exc:
3354 messages.error(request, str(exc))
3355 return render(request, "netbox_data_import/import_profile_yaml.html")
3357 summary = ", ".join(f"{v} {k.replace('_', ' ')}" for k, v in stats.items())
3358 messages.success(request, f"Profile '{profile.name}' imported/updated. {summary}.")
3359 return redirect(profile.get_absolute_url())
3362# ---------------------------------------------------------------------------
3365class CheckDeviceNameView(PermissionRequiredMixin, View):
3366 """AJAX endpoint: check if a Device the user can view has the name identity of the given name.
3368 Returns JSON: {"exists": bool, "url": str|null, "id": int|null}, and "count" when several Devices match.
3369 """
3371 permission_required = "netbox_data_import.view_importprofile"
3373 def get(self, request):
3374 """Return JSON indicating whether a device with the given name exists."""
3375 from dcim.models import Device
3376 from django.http import JsonResponse
3378 if not request.user.has_perm("dcim.view_device"): # pragma: no cover
3379 from django.http import HttpResponseForbidden
3381 return HttpResponseForbidden()
3383 key = identity_text(request.GET.get("name", ""))
3384 if not key:
3385 return JsonResponse({"exists": False, "url": None, "id": None})
3387 devices = Device.objects.restrict(request.user, "view").filter(identity_in("name", [key])).order_by("pk")
3388 found = list(devices[:2])
3389 if not found:
3390 return JsonResponse({"exists": False, "url": None, "id": None})
3391 answer = {
3392 "exists": True,
3393 "url": request.build_absolute_uri(found[0].get_absolute_url()),
3394 "id": found[0].pk,
3395 }
3396 if len(found) > 1:
3397 answer["count"] = devices.count()
3398 return JsonResponse(answer)
3401# ---------------------------------------------------------------------------
3402# Source Resolutions list view (per profile)
3403# ---------------------------------------------------------------------------
3406class SourceResolutionListView(PermissionRequiredMixin, View):
3407 """List all saved name-split resolutions for a profile."""
3409 permission_required = "netbox_data_import.view_importprofile"
3411 def get(self, request, profile_pk):
3412 """Render the list of saved source resolutions for the given profile."""
3413 profile = get_object_or_404(ImportProfile, pk=profile_pk)
3414 resolutions = SourceResolution.objects.filter(profile=profile).order_by("source_id")
3415 return render(
3416 request,
3417 "netbox_data_import/source_resolution_list.html",
3418 {
3419 "profile": profile,
3420 "resolutions": resolutions,
3421 },
3422 )
3425class SourceResolutionDeleteView(_ProfileChildDeleteView):
3426 """Delete a saved source resolution."""
3428 queryset = SourceResolution.objects.all()
3430 def post(self, request, *args, **kwargs):
3431 """Serialize against an executing import, which holds the same profile row."""
3432 resolution = self.get_object(**kwargs)
3433 try:
3434 with locked_resolution_policy(resolution.pk):
3435 # atomic-exit-safe: locked-delete-committed
3436 return super().post(request, *args, **kwargs)
3437 except (SourceResolution.DoesNotExist, ImportProfile.DoesNotExist):
3438 # The row went away between the fetch and the lock, which is the 404 the fetch would give.
3439 raise Http404 from None
3442# ---------------------------------------------------------------------------
3443# Quick-resolve views (inline fixes from preview page)
3444# ---------------------------------------------------------------------------
3447def _trace_sync_block_reason(reviewed_plan: ImportPlan, live_plan: ImportPlan) -> str:
3448 """Return why live NetBox prevents synchronization of the reviewed plan."""
3449 if live_plan.fingerprint != reviewed_plan.fingerprint:
3450 return "NetBox has changed. Re-read the preview before synchronizing."
3451 return ""
3454def _with_device_resolution_permissions(profile, actor, questions):
3455 """Add the permission state for each Device resolution action."""
3456 from .models import TraceDeviceResolution, index_digest
3458 results = []
3459 for question in questions:
3460 key = question["key"]
3461 assessment = assess_permission_scoped_save_option(
3462 actor,
3463 TraceDeviceResolution,
3464 {
3465 "profile": profile,
3466 "source_device_key": key,
3467 "source_device_key_digest": index_digest(key),
3468 },
3469 {
3470 "source_device_label": question["labels"][0] if question["labels"] else "",
3471 "selected_device_id": 1,
3472 "selected_display_name": "Pending Device selection",
3473 },
3474 unknown_fields={"selected_device_id", "selected_display_name"},
3475 )
3476 results.append(
3477 {
3478 **question,
3479 "action_allowed": assessment.allowed,
3480 "action_reason": (
3481 "" if assessment.allowed else "You do not have permission to save a Device resolution."
3482 ),
3483 }
3484 )
3485 return results
3488def _disable_policy_form(form) -> None:
3489 for field in form.fields.values():
3490 field.disabled = True
3493def _visible_row_ids(model, viewer, rows) -> set[int]:
3494 """Return the supplied row IDs this live viewer may read."""
3495 row_ids = [row.pk for row in rows]
3496 return set(model.objects.restrict(viewer, "view").filter(pk__in=row_ids).values_list("pk", flat=True))
3499POLICY_SAVE_PERMISSION_REFUSED = "You do not have permission to save this CableClass policy."
3502def _cable_policy_forms(profile, trace, viewer) -> list:
3503 """Return one policy form per CableClass in the selected trace's `cable_policies` display."""
3504 if trace is None:
3505 return []
3506 planned = {policy["cable_class"]: policy for policy in trace.cable_policies}
3507 stated = list(planned)
3508 stored_rows = list(CableClassMapping.objects.filter(profile=profile, cable_class__in=stated))
3509 rows = {row.cable_class: row for row in stored_rows}
3510 visible_ids = _visible_row_ids(CableClassMapping, viewer, stored_rows)
3511 forms = []
3512 for index, cable_class in enumerate(stated):
3513 row = rows.get(cable_class)
3514 decision = planned[cable_class]
3515 visible = row is None or row.pk in visible_ids
3516 disclosed = row is not None and visible and policy_row_is_disclosed(decision, row)
3517 hidden = decision.get(POLICY_VISIBLE) is False or not visible
3518 moved = row is not None and visible and not disclosed
3519 assessment = assess_permission_scoped_save_option(
3520 viewer,
3521 CableClassMapping,
3522 {"profile": profile, "cable_class": cable_class},
3523 {
3524 "cable_type_resolved": False,
3525 "cable_type": None,
3526 "cable_profile_resolved": False,
3527 "cable_profile": None,
3528 },
3529 unknown_fields={"cable_type_resolved", "cable_type", "cable_profile_resolved", "cable_profile"},
3530 )
3531 form = CableClassMappingForm(
3532 instance=row if disclosed else CableClassMapping(profile=profile, cable_class=cable_class),
3533 auto_id=f"id_%s_{index}",
3534 )
3535 if hidden or moved or not assessment.allowed:
3536 _disable_policy_form(form)
3537 forms.append(
3538 {
3539 "cable_class": cable_class,
3540 "cable_type": POLICY_HIDDEN if hidden else decision["cable_type"],
3541 "cable_profile": POLICY_HIDDEN if hidden else decision["cable_profile"],
3542 "breakout_profile": decision["breakout_profile"],
3543 "resolved": False if hidden else bool(decision["policy"]),
3544 "form": form,
3545 "reason": (
3546 POLICY_WRITE_REFUSED
3547 if hidden
3548 else PROFILE_POLICY_MOVED
3549 if moved
3550 else POLICY_SAVE_PERMISSION_REFUSED
3551 if not assessment.allowed
3552 else ""
3553 ),
3554 }
3555 )
3556 return forms
3559PORT_CLASS_FOR_PREVIEW = "preview"
3560PORT_CLASS_FOR_PROFILE = "profile"
3561PORT_CLASS_SAVE_PERMISSION_REFUSED = "You do not have permission to save this PortClass mapping."
3562PORT_CLASS_CLEAR_PERMISSION_REFUSED = "You do not have permission to clear this PortClass mapping."
3565def _port_class_rows(profile, workspace, planning_context, viewer) -> list:
3566 """Return one row per PortClass label outside the fixed vocabulary, with its mapping and its controls."""
3567 labels = workspace.port_class_labels
3568 if not labels:
3569 return []
3570 preview_map = preview_port_class_map(planning_context)
3571 stored = {
3572 row.source_label: row for row in PortClassMapping.objects.filter(profile=profile, source_label__in=list(labels))
3573 }
3574 visible_ids = _visible_row_ids(PortClassMapping, viewer, list(stored.values()))
3575 counts = Counter(item["label"] for trace in workspace.traces for item in trace.port_classes)
3576 delete_permission = get_permission_for_model(PortClassMapping, "delete")
3577 rows = []
3578 for index, (label, port_class) in enumerate(labels.items()):
3579 row = stored.get(label)
3580 source = PORT_CLASS_FOR_PREVIEW if label in preview_map else PORT_CLASS_FOR_PROFILE if port_class else ""
3581 assessment = assess_permission_scoped_save_option(
3582 viewer,
3583 PortClassMapping,
3584 {"profile": profile, "source_label": label},
3585 {"port_class": None},
3586 unknown_fields={"port_class"},
3587 )
3588 hidden = row is not None and row.pk not in visible_ids
3589 # A preview mapping is this preview's own decision, so only a profile mapping can be hidden.
3590 redacted = hidden and source == PORT_CLASS_FOR_PROFILE
3591 save_reason = (
3592 POLICY_WRITE_REFUSED if hidden else "" if assessment.allowed else PORT_CLASS_SAVE_PERMISSION_REFUSED
3593 )
3594 clear_reason = ""
3595 if source == PORT_CLASS_FOR_PROFILE:
3596 clear_reason = (
3597 POLICY_WRITE_REFUSED
3598 if row is None or hidden
3599 else ""
3600 if viewer.has_perm(delete_permission, row)
3601 else PORT_CLASS_CLEAR_PERMISSION_REFUSED
3602 )
3603 rows.append(
3604 {
3605 "label": label,
3606 "port_class": POLICY_HIDDEN if redacted else port_class,
3607 "source": source,
3608 "trace_count": counts[label],
3609 "save_reason": save_reason,
3610 "clear_reason": clear_reason,
3611 "form": PortClassMappingForm(
3612 initial={} if redacted else {"port_class": port_class},
3613 auto_id=f"id_%s_port_class_{index}",
3614 ),
3615 }
3616 )
3617 return rows
3620RETAINED_SEGMENT_REASON = (
3621 "This plan keeps the Cable that already proves this segment, and an override cannot change it. "
3622 "Correct that Cable in NetBox, then re-read."
3623)
3624SEGMENT_FORCE_PERMISSION_REFUSED = "You do not have permission to force this segment policy."
3625SEGMENT_OUT_OF_VIEW = (
3626 "A port this segment reaches is outside your view, so this preview cannot change its Cable policy."
3627)
3628SEGMENT_CLEAR_PERMISSION_REFUSED = "You do not have permission to clear this segment policy."
3631def _workspace_segment(workspace, unit_identity: str, position: str):
3632 """Return the reviewed trace and the segment it states at *position*, or None for neither."""
3633 try:
3634 index = int(position)
3635 except (TypeError, ValueError):
3636 return None, None
3637 trace = next((item for item in workspace.traces if item.identity == unit_identity), None)
3638 if trace is None:
3639 return None, None
3640 return trace, next((segment for segment in trace.segments if segment["index"] == index), None)
3643def _segment_policy_forms(profile, trace, viewer) -> list:
3644 """Return each segment of the selected trace with the control that forces its Cable policy."""
3645 if trace is None:
3646 return []
3647 keys = [segment["segment_key"] for segment in trace.segments if segment["segment_key"]]
3648 stored_rows = list(CableSegmentOverride.objects.filter(profile=profile, segment_key__in=keys))
3649 rows = {row.segment_key: row for row in stored_rows}
3650 mappings = list(
3651 CableClassMapping.objects.filter(
3652 profile=profile,
3653 cable_class__in={segment["cable_class"] for segment in trace.segments},
3654 )
3655 )
3656 mappings_by_class = {row.cable_class: row for row in mappings}
3657 visible_overrides = _visible_row_ids(CableSegmentOverride, viewer, stored_rows)
3658 visible_mappings = _visible_row_ids(CableClassMapping, viewer, mappings)
3659 forms = []
3660 for segment in trace.segments:
3661 stored = rows.get(segment["segment_key"])
3662 # An unplanned segment carries no policy disclosure, so no row can have moved under it.
3663 deciding = (stored or mappings_by_class.get(segment["cable_class"])) if segment["segment_key"] else None
3664 visible_ids = visible_overrides if stored is not None else visible_mappings
3665 visible = deciding is None or deciding.pk in visible_ids
3666 disclosed = deciding is not None and visible and policy_row_is_disclosed(segment, deciding)
3667 hidden = segment.get(POLICY_VISIBLE) is False or not visible
3668 moved = deciding is not None and visible and not disclosed
3669 force_assessment = assess_permission_scoped_save_option(
3670 viewer,
3671 CableSegmentOverride,
3672 {"profile": profile, "segment_key": segment["segment_key"]},
3673 {
3674 "cable_type": None,
3675 "cable_profile": None,
3676 "source_trace_identity": trace.trace_identity,
3677 "segment_index": segment["index"],
3678 },
3679 unknown_fields={"cable_type", "cable_profile"},
3680 )
3681 form = CableSegmentOverrideForm(
3682 instance=(
3683 stored
3684 if stored is not None and disclosed
3685 else CableSegmentOverride(profile=profile, segment_key=segment["segment_key"])
3686 ),
3687 initial={} if hidden or moved else cable_policy_form_initial(segment["policy"]),
3688 auto_id=f"id_%s_segment_{segment['index']}",
3689 breakout_leg=bool(segment["breakout_profile"]),
3690 )
3691 shared_reason = POLICY_WRITE_REFUSED if hidden else PROFILE_POLICY_MOVED if moved else ""
3692 force_reason = (
3693 shared_reason
3694 or _segment_override_reason(segment)
3695 or ("" if force_assessment.allowed else SEGMENT_FORCE_PERMISSION_REFUSED)
3696 )
3697 delete_permission = get_permission_for_model(CableSegmentOverride, "delete")
3698 clear_reason = shared_reason or (
3699 SEGMENT_CLEAR_PERMISSION_REFUSED
3700 if stored is not None and viewer is not None and not viewer.has_perm(delete_permission, stored)
3701 else ""
3702 )
3703 if force_reason:
3704 _disable_policy_form(form)
3705 forms.append(
3706 {
3707 **segment,
3708 "cable_type": POLICY_HIDDEN if hidden else segment["cable_type"],
3709 "cable_profile": POLICY_HIDDEN if hidden else segment["cable_profile"],
3710 "position": segment["index"] + 1,
3711 "reason": force_reason,
3712 "clear_reason": clear_reason,
3713 "form": form,
3714 }
3715 )
3716 return forms
3719def _topology_summary(segments) -> str:
3720 """Count the proposed segments by status, for the line the collapsed topology group shows."""
3721 if not segments:
3722 return "No segment planned"
3723 counts = Counter(segment["status"] or "not planned" for segment in segments)
3724 statuses = ", ".join(f"{count} {status}" for status, count in counts.items())
3725 return f"{len(segments)} segment{pluralize(len(segments))}: {statuses}"
3728def _segment_override_reason(segment) -> str:
3729 """Return why one segment cannot take an override now, or an empty string when it can."""
3730 if not segment["segment_key"]:
3731 return "This trace has not resolved both ends of this segment."
3732 if segment["retained"]:
3733 return RETAINED_SEGMENT_REASON
3734 return ""
3737def _planned_segment(plan, unit_identity: str, index: int):
3738 """Return one segment as the plan states it, before presentation hides a viewer's rows; a write reads this."""
3739 segments = plan.unit(unit_identity).display["trace"]["segments"]
3740 return next(segment for segment in segments if segment["index"] == index)
3743def _workspace_cable_classes(workspace) -> set:
3744 """Return every CableClass value the reviewed preview's traces actually state."""
3745 return {
3746 segment["cable_class"] for trace in workspace.traces for segment in trace.segments if segment["cable_class"]
3747 }
3750def _workspace_device_questions(workspace) -> dict[str, dict]:
3751 """Return the active plan's Device questions, keyed by canonical source label."""
3752 questions: dict[str, dict] = {}
3753 for trace in workspace.traces:
3754 for item in trace.devices:
3755 key = source_device_key(item.get("key", ""))
3756 if key:
3757 questions.setdefault(key, item)
3758 return questions
3761def _workspace_location_paths(workspace) -> dict[str, str]:
3762 """Return each source Location path the active plan carries, by canonical key, in key order."""
3763 paths: dict[str, str] = {}
3764 for trace in workspace.traces:
3765 for item in trace.devices:
3766 for path in item.get("locations", ()):
3767 if key := source_location_key(path):
3768 paths.setdefault(key, path)
3769 return dict(sorted(paths.items()))
3772def _workspace_location_prefixes(workspace) -> dict[str, str]:
3773 """Return each prefix key of the active plan's source Location paths, with its representative spelling."""
3774 return location_prefix_spellings(_workspace_location_paths(workspace))
3777_SEVERITY_RANK = {Severity.INFO: 0, Severity.WARNING: 1, Severity.ERROR: 2}
3780def _deduplicate_findings(findings: list[dict[str, str]]) -> list[dict[str, str]]:
3781 """Return findings once per message in first-seen order, each with its most severe copy, and every lost override."""
3782 unique: list[dict[str, str]] = []
3783 placed: dict[str, int] = {}
3784 for finding in findings:
3785 message = finding["message"]
3786 if finding["code"] == "cable.segment_override_lost" or message not in placed:
3787 placed.setdefault(message, len(unique))
3788 unique.append(finding)
3789 elif _SEVERITY_RANK[finding["severity"]] > _SEVERITY_RANK[unique[placed[message]]["severity"]]:
3790 # One line stands for every copy, so it must not read as less severe than the worst of them.
3791 unique[placed[message]] = finding
3792 return unique
3795class _TraceWorkspaceMixin(_SessionEndedRefusal, _PreviewCommandMixin):
3796 """A trace workspace command: a refusal goes back to the trace it came from."""
3798 def refusal_url(self, request):
3799 """Return the trace a refused workspace command came from, so the selection survives."""
3800 return _trace_workspace_url(request.POST.get("trace", ""))
3803def _claimed_snapshot(request, *, profile_action="change", claim=None):
3804 """Return the preview a read answers for, which must be exactly the one its page displays."""
3805 expected = PreviewClaim.posted(request.GET) if claim is None else claim
3806 snapshot = read_preview(request, expected=expected, profile_action=profile_action)
3807 if not snapshot.active:
3808 raise StalePreview("No current import preview matches this request.")
3809 return snapshot
3812def _trace_reader(actor, profile, planning_context):
3813 """Return the operator's scoped reader for the import target of one trace preview."""
3814 return NetBoxReader.for_actor(actor).for_planning_context(planning_context, output_kinds=profile.output_kinds)
3817def _trace_workspace_url(identity: str) -> str:
3818 """Return the workspace URL that reopens one trace."""
3819 url = reverse("plugins:netbox_data_import:trace_workspace")
3820 # An identity the replan dropped selects nothing, and the page falls back to its first trace.
3821 return f"{url}?{urlencode({'trace': identity.strip()})}" if identity.strip() else url
3824def _proposal_entries(workspace, terminations, *, profile, viewer, reader):
3825 """Return the proposal display and each termination with its proposal read, as the workspace shows them."""
3826 from .proposal_presentation import ProposalPresentation
3828 display = ProposalPresentation(profile=profile, actor=viewer, reader=reader)
3829 reads = display.fields(
3830 [
3831 {**field, "source_ambiguous": workspace.termination_sources.get(field["field_key"]) is None}
3832 for field in terminations
3833 ]
3834 )
3835 return display, [
3836 {**field, "proposal": reads[field["field_key"]]["presentation"], "proposal_read": reads[field["field_key"]]}
3837 for field in terminations
3838 ]
3841def _active_proposal_count(profile, workspace, display) -> tuple[int | str, int]:
3842 """Return what the summary strip shows for the active proposals, and when the database counted them.
3844 The time is the counting statement's own start, in microseconds, so a later count saw every commit an
3845 earlier one saw, and the page can refuse an answer that arrives after a newer one.
3846 """
3847 from django.db.models import Count, DateTimeField, Func, Max
3848 from django.db.models.functions import Coalesce
3850 from .models import ProposalStatus, ResolutionProposal
3852 counted_at = Func(function="statement_timestamp", output_field=DateTimeField())
3853 result = ResolutionProposal.objects.filter(
3854 profile=profile,
3855 task_type=SELECT_TERMINATION_TASK,
3856 field_key__in=list(workspace.termination_sources),
3857 status__in=ProposalStatus.ACTIVE,
3858 ).aggregate(count=Count("pk"), at=Coalesce(Max(counted_at), counted_at))
3859 stamp = (result["at"] - datetime.fromtimestamp(0, tz=UTC)) // timedelta(microseconds=1)
3860 return ("Not permitted" if display.view_reason else result["count"]), stamp
3863def _device_url(question) -> str:
3864 """Return the NetBox page of a Device question's resolved Device, while the viewer may still view it."""
3865 # The live disclosure check drops this source from a question that names a hidden Device.
3866 source = question.get(DISCLOSURE_SOURCE)
3867 return reverse("dcim:device", kwargs={"pk": source["pk"]}) if source and source["kind"] == DEVICE_ROW else ""
3870def _termination_cards(terminations, trace, claim):
3871 """Add what each termination card renders beyond its proposal: the resolved Device, an id, and its read."""
3872 resolved = {source_device_key(device["key"]): device for device in trace.devices if device.get("selected")}
3873 read_url = reverse("plugins:netbox_data_import:trace_proposal")
3874 return [
3875 {
3876 **termination,
3877 "resolved_device": device["selected"] if device else "",
3878 "resolved_device_url": _device_url(device) if device else "",
3879 "card_id": "proposalCard" + hashlib.sha256(termination["field_key"].encode()).hexdigest()[:16],
3880 "read_url": f"{read_url}?"
3881 + urlencode({"field_key": termination["field_key"], "trace": trace.identity, **claim.fields()}),
3882 }
3883 for termination in terminations
3884 for device in (resolved.get(parse_termination_field_key(termination["field_key"])["device"]),)
3885 ]
3888def _proposal_card(request, snapshot, field_key: str, identity: str):
3889 """Render one termination card of the preview the page displays, for a poll or after a proposal command."""
3890 try:
3891 workspace = snapshot.workspace(request.user)
3892 except PlanError:
3893 raise StalePreview(UNREADABLE_PREVIEW) from None
3894 traces = [trace for trace in workspace.traces if any(item["field_key"] == field_key for item in trace.terminations)]
3895 trace = next((trace for trace in traces if trace.identity == identity), traces[0] if traces else None)
3896 if trace is None:
3897 raise InvalidProposalTarget("This preview asked no question about that termination.")
3898 field = next(item for item in trace.terminations if item["field_key"] == field_key)
3899 try:
3900 reader = _trace_reader(request.user, snapshot.profile, snapshot.planning_context)
3901 except PlanningTargetUnavailable:
3902 reader = None
3903 display, entries = _proposal_entries(
3904 workspace, [field], profile=snapshot.profile, viewer=request.user, reader=reader
3905 )
3906 active_proposals, counted_at = _active_proposal_count(snapshot.profile, workspace, display)
3907 return render(
3908 request,
3909 "netbox_data_import/_proposal_card_answer.html",
3910 {
3911 "termination": _termination_cards(entries, trace, snapshot.claim)[0],
3912 "selected_trace": trace,
3913 "preview_claim": snapshot.claim,
3914 "active_proposals": active_proposals,
3915 "active_proposals_counted_at": counted_at,
3916 },
3917 )
3920def _proposal_card_after(request, result, field_key: str):
3921 """Render the card a proposal command changed, from the preview the command left behind."""
3922 snapshot = _claimed_snapshot(request, profile_action="view", claim=result.claim)
3923 return _proposal_card(request, snapshot, field_key, request.POST.get("trace", ""))
3926class TraceReviewWorkspaceView(PermissionRequiredMixin, View):
3927 """Section 10.2: one review workspace page per preview, for the traces it planned."""
3929 permission_required = "netbox_data_import.change_importprofile"
3931 def get(self, request):
3932 """Render the reviewed traces and say whether live NetBox has moved under them."""
3933 snapshot, refusal = _review_snapshot(request)
3934 if refusal is not None:
3935 return refusal
3936 profile = snapshot.profile
3937 try:
3938 workspace = snapshot.workspace(request.user)
3939 except PlanError:
3940 return _unreadable_preview_page(request, snapshot)
3941 live = _live_plan(snapshot, request.user)
3942 if live is None:
3943 messages.warning(request, TARGET_GONE)
3944 return redirect(reverse("plugins:netbox_data_import:import_setup"))
3945 # Section 10.2: compared on each full load and on the re-read action, never polled.
3946 sync_block_reason = _trace_sync_block_reason(workspace.plan, live)
3947 drift = bool(sync_block_reason)
3948 retained = _retained_sync(snapshot)
3949 retained_reason = retained[1]
3950 block_reason = retained_reason or sync_block_reason
3951 traces = (
3952 [with_blocked_sync(trace, block_reason) for trace in workspace.traces] if block_reason else workspace.traces
3953 )
3954 wanted = request.GET.get("trace", "")
3955 selected = next((trace for trace in traces if trace.identity == wanted), traces[0] if traces else None)
3956 summary = dict(workspace.trace_summary)
3957 from .models import TerminationResolution, TraceDeviceResolution, TraceLocationResolution
3959 summary["saved_decisions"] = sum(
3960 model.objects.restrict(request.user, "view").filter(profile=profile).count()
3961 for model in (TerminationResolution, TraceDeviceResolution, TraceLocationResolution)
3962 )
3963 summary["preview_state"] = "changed in NetBox" if drift else "current"
3964 from .proposal_presentation import group_terminations
3966 try:
3967 reader = _trace_reader(request.user, profile, snapshot.planning_context)
3968 except PlanningTargetUnavailable:
3969 messages.warning(request, TARGET_GONE)
3970 return redirect(reverse("plugins:netbox_data_import:import_setup"))
3971 # The picker reads Locations a page at a time, so the page only asks whether any is visible.
3972 has_locations = site_locations(reader).exists()
3973 location_tree = present_location_tree(
3974 profile=profile,
3975 viewer=request.user,
3976 reader=reader,
3977 paths=_workspace_location_paths(workspace),
3978 has_locations=has_locations,
3979 )
3980 proposal_display, terminations = _proposal_entries(
3981 workspace, selected.terminations if selected else [], profile=profile, viewer=request.user, reader=reader
3982 )
3983 if selected is not None:
3984 selected = replace(
3985 selected,
3986 findings=(
3987 _deduplicate_findings(selected.findings) if selected.disposition != "invalid" else selected.findings
3988 ),
3989 terminations=terminations,
3990 )
3991 cable_policy_forms = _cable_policy_forms(profile, selected, request.user)
3992 port_class_rows = _port_class_rows(profile, workspace, snapshot.planning_context, request.user)
3993 segment_policy_forms = _segment_policy_forms(profile, selected, request.user)
3994 attention, settled = group_terminations(selected.terminations if selected else [])
3995 selected_devices = _with_device_resolution_permissions(
3996 profile,
3997 request.user,
3998 selected.devices if selected else [],
3999 )
4000 attention_devices = [device for device in selected_devices if device.get("selectable")]
4001 manual_devices = [device for device in selected_devices if device.get("state_style") == "manual"]
4002 settled_devices = [
4003 device
4004 for device in selected_devices
4005 if not device.get("selectable") and device.get("state_style") != "manual"
4006 ]
4007 attention = _termination_cards(attention, selected, snapshot.claim) if selected else []
4008 summary["active_proposals"], summary["active_proposals_counted_at"] = _active_proposal_count(
4009 profile, workspace, proposal_display
4010 )
4011 ask_all_reason = (
4012 proposal_display.request_block_reason
4013 or retained_reason
4014 or ("" if summary["unresolved_terminations"] else "Every termination in this preview is resolved.")
4015 )
4016 return render(
4017 request,
4018 "netbox_data_import/trace_workspace.html",
4019 {
4020 "profile": profile,
4021 "traces": traces,
4022 "trace_dispositions": workspace.trace_dispositions,
4023 "selected_trace": selected,
4024 "attention_devices": attention_devices,
4025 "manual_devices": manual_devices,
4026 "settled_devices": settled_devices,
4027 "attention_terminations": attention,
4028 "outside_claim_mark": OUTSIDE_CLAIM_MARK,
4029 "settled_terminations": settled,
4030 "cable_policy_forms": cable_policy_forms,
4031 "port_class_rows": port_class_rows,
4032 "segment_policy_forms": segment_policy_forms,
4033 "topology_summary": _topology_summary(segment_policy_forms),
4034 "summary": summary,
4035 "ask_all_reason": ask_all_reason,
4036 "sync_all": held_sync(workspace.sync_all.action, block_reason),
4037 "sync_all_note": workspace.sync_all.unsynced_note,
4038 "location_tree": location_tree,
4039 "has_locations": has_locations,
4040 "import_location_unavailable": reader.location_unavailable,
4041 "drift": drift,
4042 **_sync_status_context(snapshot, selected.identity if selected else "", retained),
4043 "plugin_version": _plugin_version,
4044 },
4045 )
4048def _refuse_drifted_sync(preview) -> None:
4049 """Refuse a trace sync that live NetBox has moved under."""
4050 live = ImportEngine.plan(preview.profile, preview.document, preview.actor, preview.planning_context)
4051 if reason := _trace_sync_block_reason(preview.plan, live):
4052 raise PreviewCommandRefused(reason, 409)
4055def _posted_sync_selection(data) -> tuple[str, ...]:
4056 """Return the selection a trace sync form posts, its trace first; refuse an empty or repeated one."""
4057 selection = tuple(data.getlist("selection"))
4058 if not selection or "" in selection or len(set(selection)) != len(selection):
4059 raise PreviewCommandRefused(SYNC_SELECTION_INVALID)
4060 return selection
4063@dataclass(frozen=True)
4064class _TraceSync(QueueImport):
4065 """Queue the selection one trace's page showed, keeping the preview pending on it."""
4067 selection: tuple[str, ...]
4068 keeps_preview = True
4070 def selection_for(self, preview):
4071 """Refuse an unknown unit, a held dependency, a selection the page did not show, drift, or nothing to sync."""
4072 if any(preview.plan.unit(identity) is None for identity in self.selection):
4073 raise PreviewCommandRefused(SYNC_SELECTION_INVALID)
4074 selection = preview.workspace.sync_selection(self.selection[0])
4075 if selection and preview.workspace.cannot_sync(selection):
4076 raise PreviewCommandRefused(SYNC_DEPENDENCY_HELD)
4077 if selection and selection != self.selection:
4078 raise StalePreview(SYNC_SELECTION_CHANGED)
4079 _refuse_drifted_sync(preview)
4080 if not selection:
4081 raise PreviewCommandRefused(SYNC_NOTHING)
4082 return list(selection)
4085class TraceSyncView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4086 """Synchronize one Source Trace together with the units its changes depend on."""
4088 permission_required = "netbox_data_import.change_importprofile"
4090 def post(self, request):
4091 """Queue the reviewed plan for the selection one trace's page showed."""
4092 command = _TraceSync(selection=_posted_sync_selection(request.POST))
4093 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4094 return redirect(reverse("plugins:netbox_data_import:import_progress", kwargs={"pk": result.outcome.job_id}))
4096 def refusal_url(self, request):
4097 """Return the trace the sync was asked for, which the form posts first."""
4098 return _trace_workspace_url(next(iter(request.POST.getlist("selection")), ""))
4101class _TraceSyncAll(QueueImport):
4102 """Queue every actionable Source Trace that can sync, with the units their changes depend on, as one Job."""
4104 keeps_preview = True
4106 def selection_for(self, preview):
4107 """Refuse a sync live NetBox has moved under, or a preview with no trace that can sync."""
4108 _refuse_drifted_sync(preview)
4109 selection = preview.workspace.sync_all.units
4110 if not selection:
4111 raise PreviewCommandRefused(SYNC_ALL_NOTHING)
4112 return list(selection)
4115class TraceSyncAllView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4116 """Synchronize every actionable Source Trace and keep the workspace for the traces that remain."""
4118 permission_required = "netbox_data_import.change_importprofile"
4120 def post(self, request):
4121 """Queue the reviewed plan for every trace that can sync, then show the Job's progress."""
4122 result = apply_preview_command(request, PreviewClaim.posted(request.POST), _TraceSyncAll())
4123 return redirect(reverse("plugins:netbox_data_import:import_progress", kwargs={"pk": result.outcome.job_id}))
4126class TraceSyncStatusView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4127 """Answer the workspace's poll of its trace sync, and reload the page once the sync ends."""
4129 permission_required = "netbox_data_import.view_importprofile"
4131 def get(self, request):
4132 """Return the status block while the Job waits or runs; after that the whole page changes."""
4133 snapshot = _claimed_snapshot(request, profile_action="view")
4134 context = _sync_status_context(snapshot, request.GET.get("trace", ""), _retained_sync(snapshot))
4135 if not context["sync_active"]:
4136 # The trace actions and the re-read button all change, so the page loads again.
4137 response = HttpResponse(status=204)
4138 response["HX-Refresh"] = "true"
4139 return response
4140 return render(request, "netbox_data_import/_trace_sync_status.html", context)
4142 def refusal_url(self, request):
4143 """Return the trace the polling page shows."""
4144 return _trace_workspace_url(request.GET.get("trace", ""))
4147class TraceSyncCancelView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4148 """Cancel the preview's own trace sync before a worker starts it."""
4150 permission_required = "netbox_data_import.change_importprofile"
4152 def post(self, request):
4153 """Cancel the sync, then show the workspace with the preview re-read."""
4154 job_id = request.POST.get("job_id", "")
4155 if not job_id.isascii() or not job_id.isdigit():
4156 raise StalePreview(SYNC_NOT_THIS_PREVIEW)
4157 result = apply_preview_command(request, PreviewClaim.posted(request.POST), CancelTraceSync(int(job_id)))
4158 messages.success(request, result.outcome.message)
4159 return redirect(_trace_workspace_url(request.POST.get("trace", "")))
4162CANDIDATE_LIMIT_INVALID = f"Candidate limit must be an integer from 1 to {ELIGIBLE_TERMINATION_LIMIT}."
4163# PostgreSQL reads OFFSET and LIMIT as bigint, so the last row of a page must stay inside that range.
4164CANDIDATE_OFFSET_MAX = 2**63 - 1 - ELIGIBLE_TERMINATION_LIMIT
4165CANDIDATE_OFFSET_INVALID = f"Candidate offset must be an integer from 0 to {CANDIDATE_OFFSET_MAX}."
4168@dataclass(frozen=True)
4169class CandidatePage:
4170 """One picker page, or the fixed sentence that refuses it."""
4172 limit: int = 0
4173 offset: int = 0
4174 error: str = ""
4177def _candidate_page(params) -> CandidatePage:
4178 """Return the page limit and offset of one picker read, or of the write that rechecks its offer."""
4179 raw_limit, raw_offset = params.get("limit"), params.get("offset")
4180 try:
4181 limit = ELIGIBLE_TERMINATION_LIMIT if raw_limit is None else int(raw_limit)
4182 except (TypeError, ValueError):
4183 return CandidatePage(error=CANDIDATE_LIMIT_INVALID)
4184 if not 1 <= limit <= ELIGIBLE_TERMINATION_LIMIT:
4185 return CandidatePage(error=CANDIDATE_LIMIT_INVALID)
4186 try:
4187 offset = 0 if raw_offset in (None, "") else int(raw_offset)
4188 except (TypeError, ValueError):
4189 return CandidatePage(error=CANDIDATE_OFFSET_INVALID)
4190 if not 0 <= offset <= CANDIDATE_OFFSET_MAX:
4191 return CandidatePage(error=CANDIDATE_OFFSET_INVALID)
4192 return CandidatePage(limit=limit, offset=offset)
4195class TraceTerminationCandidatesView(PermissionRequiredMixin, View):
4196 """Serve one page of eligible terminations for the workspace picker."""
4198 permission_required = "netbox_data_import.change_importprofile"
4200 def get(self, request):
4201 """Return the eligible candidates and the uncapped total the count states."""
4202 try:
4203 snapshot = _claimed_snapshot(request)
4204 workspace = snapshot.workspace(request.user)
4205 except StalePreview as exc:
4206 return _stale_json(exc)
4207 except PlanError:
4208 return JsonResponse({"ok": False, "error": UNREADABLE_PREVIEW}, status=409)
4209 field_key = request.GET.get("field_key", "").strip()
4210 # A review read answers a question this preview asked, never one the caller invented.
4211 if field_key not in workspace.termination_sources:
4212 return JsonResponse(
4213 {"ok": False, "error": "This preview asked no question about that termination."}, status=400
4214 )
4215 if workspace.termination_sources[field_key] is None:
4216 return JsonResponse({"ok": False, "error": TERMINATION_UNRESOLVABLE}, status=400)
4217 page = _candidate_page(request.GET)
4218 if page.error:
4219 return JsonResponse({"ok": False, "error": page.error}, status=400)
4220 limit, offset = page.limit, page.offset
4221 try:
4222 reader = _trace_reader(request.user, snapshot.profile, snapshot.planning_context)
4223 found = eligible_terminations(
4224 field_key,
4225 reader,
4226 profile=snapshot.profile,
4227 search=request.GET.get("search", ""),
4228 limit=limit,
4229 offset=offset,
4230 )
4231 except (PlanningTargetUnavailable, ValueError):
4232 return JsonResponse({"ok": False, "error": TERMINATION_UNRESOLVABLE}, status=400)
4233 return JsonResponse(
4234 {
4235 "ok": True,
4236 # A candidate is its model and its id together, because two models can share one id.
4237 "candidates": [
4238 {
4239 "id": candidate.pk,
4240 "object_type": candidate._meta.label_lower,
4241 "model": str(candidate._meta.verbose_name),
4242 "name": candidate.name,
4243 "display": str(candidate),
4244 "outside_claim": found.outside_claim(candidate),
4245 }
4246 for candidate in found.candidates
4247 ],
4248 "shown": len(found.candidates),
4249 "total": found.total,
4250 "offset": offset,
4251 "limit": limit,
4252 }
4253 )
4256class TraceDeviceCandidatesView(PermissionRequiredMixin, View):
4257 """Serve one bounded page of visible Device candidates for a plan-authored question."""
4259 permission_required = "netbox_data_import.change_importprofile"
4261 def get(self, request):
4262 """Return permission-scoped candidates and their source-evidence explanations."""
4263 try:
4264 snapshot = _claimed_snapshot(request)
4265 workspace = snapshot.workspace(request.user)
4266 except StalePreview as exc:
4267 return _stale_json(exc)
4268 except PlanError:
4269 return JsonResponse({"ok": False, "error": UNREADABLE_PREVIEW}, status=409)
4270 device_key = source_device_key(request.GET.get("device_key", ""))
4271 question = _workspace_device_questions(workspace).get(device_key)
4272 if question is None:
4273 return JsonResponse({"ok": False, "error": "This preview asked no question about that Device."}, status=400)
4274 search = request.GET.get("search", "")
4275 if len(search) > 200:
4276 return JsonResponse({"ok": False, "error": "Device search must be 200 characters or fewer."}, status=400)
4277 page = _candidate_page(request.GET)
4278 if page.error:
4279 return JsonResponse({"ok": False, "error": page.error}, status=400)
4280 limit, offset = page.limit, page.offset
4281 try:
4282 evidence = DeviceEvidence.from_dict(question)
4283 except (TypeError, ValueError):
4284 return JsonResponse({"ok": False, "error": "That Device cannot be resolved here."}, status=400)
4285 try:
4286 reader = _trace_reader(request.user, snapshot.profile, snapshot.planning_context)
4287 found = eligible_trace_devices(
4288 profile=snapshot.profile, reader=reader, evidence=evidence, search=search, limit=limit, offset=offset
4289 )
4290 except PlanningTargetUnavailable:
4291 return JsonResponse({"ok": False, "error": "That Device cannot be resolved here."}, status=400)
4292 return JsonResponse(
4293 {
4294 "ok": True,
4295 "candidates": [
4296 {
4297 "id": candidate.device.pk,
4298 "name": candidate.device.name,
4299 "display": str(candidate.device),
4300 "matched_facts": [fact.to_dict() for fact in candidate.matched],
4301 "conflicting_facts": [fact.to_dict() for fact in candidate.conflicting],
4302 "import_location": (
4303 candidate.import_location.to_dict() if candidate.import_location is not None else None
4304 ),
4305 }
4306 for candidate in found.candidates
4307 ],
4308 "shown": len(found.candidates),
4309 "total": found.total,
4310 "offset": offset,
4311 "limit": limit,
4312 }
4313 )
4316@dataclass(frozen=True)
4317class _ResolveTraceDevice(PreviewCommand):
4318 """Save one plan-authored source Device decision; the coordinator stores the replan it returns."""
4320 device_key: str
4321 device_id: int
4322 search: str
4323 limit: int
4324 offset: int
4326 def apply(self, preview):
4327 """Recheck the offered Device, save it under the profile policy lock, and replan."""
4328 question = _workspace_device_questions(preview.workspace).get(self.device_key)
4329 if question is None:
4330 raise PreviewCommandRefused("This preview asked no question about that Device.")
4331 try:
4332 evidence = DeviceEvidence.from_dict(question)
4333 except (TypeError, ValueError):
4334 raise PreviewCommandRefused("That Device is not one of the eligible candidates.") from None
4335 try:
4336 plan, chosen = save_trace_device_resolution_and_replan(
4337 profile=preview.profile,
4338 source_document=preview.document,
4339 actor=preview.actor,
4340 planning_context=preview.planning_context,
4341 evidence=evidence,
4342 selected_device_id=self.device_id,
4343 search=self.search,
4344 limit=self.limit,
4345 offset=self.offset,
4346 reviewed_fingerprint=preview.reviewed_fingerprint,
4347 )
4348 except IneligibleDeviceSelection:
4349 raise PreviewCommandRefused("That Device is not one of the eligible candidates.") from None
4350 return CommandOutcome(message=f"Source Device resolved to '{chosen}'.", plan=plan)
4353class TraceResolveDeviceView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4354 """Save one plan-authored source Device decision and replan the workspace."""
4356 permission_required = "netbox_data_import.change_importprofile"
4358 def post(self, request):
4359 """Run the decision as one coordinated command."""
4360 search = request.POST.get("search", "")
4361 if len(search) > 200:
4362 raise PreviewCommandRefused("Device search must be 200 characters or fewer.")
4363 try:
4364 device_id = int(request.POST.get("device_id", ""))
4365 except (TypeError, ValueError):
4366 raise PreviewCommandRefused("That Device is not one of the eligible candidates.") from None
4367 # The write rechecks the page that made the offer.
4368 page = _candidate_page(request.POST)
4369 if page.error:
4370 raise PreviewCommandRefused(page.error)
4371 command = _ResolveTraceDevice(
4372 device_key=source_device_key(request.POST.get("device_key", "")),
4373 device_id=device_id,
4374 search=search,
4375 limit=page.limit,
4376 offset=page.offset,
4377 )
4378 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4379 return _command_response(request, result, self.refusal_url(request))
4382LOCATION_CHOICE_REFUSED = "Choose a visible Location in the selected Site."
4385class TraceLocationCandidatesView(PermissionRequiredMixin, View):
4386 """Serve one bounded page of the selected Site's visible Locations for one source Location prefix."""
4388 permission_required = "netbox_data_import.change_importprofile"
4390 def get(self, request):
4391 """Return the searched page and the uncapped total the picker counts."""
4392 try:
4393 snapshot = _claimed_snapshot(request)
4394 workspace = snapshot.workspace(request.user)
4395 except StalePreview as exc:
4396 return _stale_json(exc)
4397 except PlanError:
4398 return JsonResponse({"ok": False, "error": UNREADABLE_PREVIEW}, status=409)
4399 # A review read answers a question this preview asked, never one the caller invented.
4400 if source_location_key(request.GET.get("location_key", "")) not in _workspace_location_prefixes(workspace):
4401 return JsonResponse(
4402 {"ok": False, "error": "This preview carries no such source Location path."}, status=400
4403 )
4404 search = request.GET.get("search", "")
4405 if len(search) > 200:
4406 return JsonResponse({"ok": False, "error": "Location search must be 200 characters or fewer."}, status=400)
4407 page = _candidate_page(request.GET)
4408 if page.error:
4409 return JsonResponse({"ok": False, "error": page.error}, status=400)
4410 limit, offset = page.limit, page.offset
4411 try:
4412 found = eligible_trace_locations(
4413 _trace_reader(request.user, snapshot.profile, snapshot.planning_context),
4414 search=search,
4415 limit=limit,
4416 offset=offset,
4417 )
4418 except PlanningTargetUnavailable:
4419 return JsonResponse({"ok": False, "error": "The import target is no longer available."}, status=400)
4420 return JsonResponse(
4421 {
4422 "ok": True,
4423 "candidates": [
4424 {"id": candidate.location.pk, "name": candidate.location.name, "parent": candidate.parent}
4425 for candidate in found.candidates
4426 ],
4427 "shown": len(found.candidates),
4428 "total": found.total,
4429 "offset": offset,
4430 "limit": limit,
4431 }
4432 )
4435@dataclass(frozen=True)
4436class _MapTraceLocation(PreviewCommand):
4437 """Map one source Location prefix the reviewed preview carries, or clear its mapping."""
4439 location_key: str
4440 location_id: int | None
4442 def apply(self, preview):
4443 """Write the decision under the profile lock; the coordinator stores the replan it returns."""
4444 paths = _workspace_location_prefixes(preview.workspace)
4445 # A review command answers a question this preview asked, never one the caller invented.
4446 if self.location_key not in paths:
4447 raise PreviewCommandRefused("This preview carries no such source Location path.")
4448 try:
4449 if self.location_id is None:
4450 plan = clear_trace_location_resolution_and_replan(
4451 profile=preview.profile,
4452 source_document=preview.document,
4453 actor=preview.actor,
4454 planning_context=preview.planning_context,
4455 source_location_key=self.location_key,
4456 reviewed_fingerprint=preview.reviewed_fingerprint,
4457 )
4458 else:
4459 plan = save_trace_location_resolution_and_replan(
4460 profile=preview.profile,
4461 source_document=preview.document,
4462 actor=preview.actor,
4463 planning_context=preview.planning_context,
4464 source_location_path=paths[self.location_key],
4465 selected_location_id=self.location_id,
4466 reviewed_fingerprint=preview.reviewed_fingerprint,
4467 )
4468 except UnacceptablePolicyDecision as exc:
4469 raise PreviewCommandRefused("; ".join(exc.errors)) from exc
4470 except IneligibleLocationSelection:
4471 raise PreviewCommandRefused(LOCATION_CHOICE_REFUSED) from None
4472 settled = "saved for" if self.location_id is not None else "cleared for"
4473 return CommandOutcome(message=f"Location mapping {settled} '{paths[self.location_key]}'.", plan=plan)
4476class TraceLocationMappingView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4477 """Map one source Location prefix the reviewed preview carries, or clear its mapping, then replan."""
4479 permission_required = "netbox_data_import.change_importprofile"
4481 def post(self, request):
4482 """Run the save or the clear as one coordinated command."""
4483 location_id = None
4484 if not request.POST.get("clear"):
4485 try:
4486 location_id = int(request.POST.get("location_id", ""))
4487 except (TypeError, ValueError):
4488 raise PreviewCommandRefused(LOCATION_CHOICE_REFUSED) from None
4489 command = _MapTraceLocation(
4490 location_key=source_location_key(request.POST.get("location_key", "")), location_id=location_id
4491 )
4492 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4493 return _command_response(request, result, self.refusal_url(request))
4496@dataclass(frozen=True)
4497class _SetCablePolicy(PreviewCommand):
4498 """Set the Cable policy for one CableClass the reviewed preview states."""
4500 cable_class: str
4501 data: Mapping
4503 def apply(self, preview):
4504 """Save the policy under the profile lock; the coordinator stores the replan it returns."""
4505 # A review command answers a question this preview asked, never one the caller invented.
4506 if self.cable_class not in _workspace_cable_classes(preview.workspace):
4507 raise PreviewCommandRefused("This preview states no segment with that CableClass.")
4508 try:
4509 plan = save_cable_class_mapping_and_replan(
4510 profile=preview.profile,
4511 source_document=preview.document,
4512 actor=preview.actor,
4513 planning_context=preview.planning_context,
4514 cable_class=self.cable_class,
4515 data=dict(self.data),
4516 reviewed_fingerprint=preview.reviewed_fingerprint,
4517 )
4518 except UnacceptablePolicyDecision as exc:
4519 raise PreviewCommandRefused("; ".join(exc.errors)) from exc
4520 return CommandOutcome(message=f"Cable policy saved for CableClass '{self.cable_class}'.", plan=plan)
4523class TraceCablePolicyView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4524 """Set the Cable policy for one CableClass the reviewed preview states, then replan."""
4526 permission_required = "netbox_data_import.change_importprofile"
4528 def post(self, request):
4529 """Run the policy save as one coordinated command."""
4530 command = _SetCablePolicy(
4531 cable_class=request.POST.get("cable_class", "").strip(),
4532 data={
4533 "cable_type": request.POST.get("cable_type", ""),
4534 "cable_profile": request.POST.get("cable_profile", ""),
4535 },
4536 )
4537 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4538 return _command_response(request, result, self.refusal_url(request))
4541def _replanned_trace(workspace, identity: str, plan) -> str:
4542 """Return the unit that states the reviewed trace's source block in a replan, or "" when none does."""
4543 trace = next((item for item in workspace.traces if item.identity == identity), None)
4544 if trace is None:
4545 return ""
4546 position = {"sheet": trace.sheet, "row_number": trace.row_number}
4547 for unit in plan.units:
4548 display = unit.display
4549 # Duplicate collapse keeps every occurrence, and the reviewed one need not be the first.
4550 if display.get("trace") is not None and position in display["occurrences"]:
4551 return unit.identity
4552 return ""
4555@dataclass(frozen=True)
4556class _MapPortClass(PreviewCommand):
4557 """Map one PortClass label the reviewed preview states, on the profile or for this preview alone."""
4559 label: str
4560 port_class: str
4561 scope: str
4562 clearing: bool
4563 trace: str
4565 def apply(self, preview):
4566 """Refuse a label or PortClass this preview cannot map, then write the decision and replan."""
4567 self._refuse_unmappable(preview)
4568 context = None
4569 if self.scope == PORT_CLASS_FOR_PREVIEW:
4570 if self.clearing and self.label not in preview_port_class_map(preview.planning_context):
4571 raise PreviewCommandRefused(f"This preview maps no PortClass label '{self.label}'.")
4572 context = with_preview_port_class(preview.context, self.label, None if self.clearing else self.port_class)
4573 # The coordinator stores this context with the plan, and it refuses a plan made for another one.
4574 plan = ImportEngine.plan(preview.profile, preview.document, preview.actor, planning_context_of(context))
4575 else:
4576 plan = self._profile_decision(preview)
4577 settled = "cleared for" if self.clearing else f"set to '{self.port_class}' for"
4578 where = "this preview" if self.scope == PORT_CLASS_FOR_PREVIEW else "the profile"
4579 return CommandOutcome(
4580 message=f"PortClass {settled} label '{self.label}' in {where}.",
4581 plan=plan,
4582 context=context,
4583 payload={"trace": _replanned_trace(preview.workspace, self.trace, plan)},
4584 )
4586 def _refuse_unmappable(self, preview):
4587 """Refuse a fixed PortClass label, a label this preview does not state, or a target outside the vocabulary."""
4588 errors = port_class_mapping_errors(self.label, self.port_class)
4589 if "source_label" in errors:
4590 raise PreviewCommandRefused(errors["source_label"])
4591 # A review command answers a question this preview asked, never one the caller invented.
4592 if self.label not in preview.workspace.port_class_labels:
4593 raise PreviewCommandRefused("This preview states no such PortClass label.")
4594 if not self.clearing and "port_class" in errors:
4595 raise PreviewCommandRefused(errors["port_class"])
4597 def _profile_decision(self, preview):
4598 """Save or clear the profile mapping under the profile lock, and return its replan."""
4599 arguments = {
4600 "profile": preview.profile,
4601 "source_document": preview.document,
4602 "actor": preview.actor,
4603 "planning_context": preview.planning_context,
4604 "source_label": self.label,
4605 "reviewed_fingerprint": preview.reviewed_fingerprint,
4606 }
4607 try:
4608 if self.clearing:
4609 return clear_port_class_mapping_and_replan(**arguments)
4610 return save_port_class_mapping_and_replan(**arguments, port_class=self.port_class)
4611 except UnacceptablePolicyDecision as exc:
4612 raise PreviewCommandRefused("; ".join(exc.errors)) from exc
4615class TracePortClassView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4616 """Map one PortClass label the reviewed preview states, or clear its mapping, then replan."""
4618 permission_required = "netbox_data_import.change_importprofile"
4620 def post(self, request):
4621 """Run the save or the clear as one coordinated command, then reopen the same source block."""
4622 scope = request.POST.get("scope", "")
4623 if scope not in (PORT_CLASS_FOR_PREVIEW, PORT_CLASS_FOR_PROFILE):
4624 raise PreviewCommandRefused("Choose whether the PortClass mapping is for the profile or for this preview.")
4625 clearing = bool(request.POST.get("clear"))
4626 command = _MapPortClass(
4627 label=request.POST.get("port_class_label", "").strip(),
4628 port_class="" if clearing else request.POST.get("port_class", "").strip(),
4629 scope=scope,
4630 clearing=clearing,
4631 trace=request.POST.get("trace", ""),
4632 )
4633 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4634 return _command_response(request, result, _trace_workspace_url(result.outcome.payload["trace"]))
4637@dataclass(frozen=True)
4638class _SetSegmentPolicy(PreviewCommand):
4639 """Force one planned segment to its own Cable policy, or clear the override again."""
4641 trace: str
4642 segment: str
4643 clearing: bool
4644 data: Mapping
4646 def apply(self, preview):
4647 """Write the decision under the profile lock; the coordinator stores the replan it returns."""
4648 trace, segment = _workspace_segment(preview.workspace, self.trace, self.segment)
4649 # A review command answers a question this preview asked, never one the caller invented.
4650 if segment is None or not segment["segment_key"]:
4651 raise PreviewCommandRefused("This preview resolved no segment there.")
4652 planned = _planned_segment(preview.plan, trace.identity, segment["index"])
4653 # An override cannot decide a Cable the plan keeps, but the operator can still remove it.
4654 if planned["retained"] and not self.clearing:
4655 # A viewer who cannot read that the Cable is kept learns only that the segment is out of reach.
4656 raise PreviewCommandRefused(RETAINED_SEGMENT_REASON if segment["retained"] else SEGMENT_OUT_OF_VIEW)
4657 try:
4658 if self.clearing:
4659 plan = clear_cable_segment_override_and_replan(
4660 profile=preview.profile,
4661 source_document=preview.document,
4662 actor=preview.actor,
4663 planning_context=preview.planning_context,
4664 segment_key=segment["segment_key"],
4665 reviewed_fingerprint=preview.reviewed_fingerprint,
4666 )
4667 else:
4668 plan = save_cable_segment_override_and_replan(
4669 profile=preview.profile,
4670 source_document=preview.document,
4671 actor=preview.actor,
4672 planning_context=preview.planning_context,
4673 segment_key=segment["segment_key"],
4674 trace_identity=trace.trace_identity,
4675 segment_index=segment["index"],
4676 cable_class=segment["cable_class"],
4677 data=dict(self.data),
4678 reviewed_fingerprint=preview.reviewed_fingerprint,
4679 breakout_leg=bool(planned["breakout_leg"]),
4680 )
4681 except UnacceptablePolicyDecision as exc:
4682 raise PreviewCommandRefused("; ".join(exc.errors)) from exc
4683 settled = "cleared on" if self.clearing else "forced on"
4684 return CommandOutcome(message=f"Cable policy {settled} segment {segment['index'] + 1}.", plan=plan)
4687class TraceCableSegmentPolicyView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4688 """Force one planned segment to its own Cable policy, or clear the override again."""
4690 permission_required = "netbox_data_import.change_importprofile"
4692 def post(self, request):
4693 """Run the override save or clear as one coordinated command."""
4694 command = _SetSegmentPolicy(
4695 trace=request.POST.get("trace", ""),
4696 segment=request.POST.get("segment", ""),
4697 clearing=bool(request.POST.get("clear")),
4698 data={
4699 "cable_type": request.POST.get("cable_type", ""),
4700 "cable_profile": request.POST.get("cable_profile", ""),
4701 },
4702 )
4703 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4704 return _command_response(request, result, self.refusal_url(request))
4707@dataclass(frozen=True)
4708class _ResolveTermination(PreviewCommand):
4709 """Record one operator termination decision the picker offered."""
4711 field_key: str
4712 object_type: str
4713 object_id: int
4714 search: str
4715 limit: int
4716 offset: int
4718 def apply(self, preview):
4719 """Recheck the offer, save the selection, and return the replan the coordinator stores."""
4720 from core.models import ObjectType
4722 workspace = preview.workspace
4723 # A review command answers a question this preview asked, never one the caller invented.
4724 if self.field_key not in workspace.termination_sources:
4725 raise PreviewCommandRefused("This preview asked no question about that termination.")
4726 if workspace.termination_sources[self.field_key] is None:
4727 raise PreviewCommandRefused(TERMINATION_UNRESOLVABLE)
4728 try:
4729 reader = _trace_reader(preview.actor, preview.profile, preview.planning_context)
4730 # The recheck repeats the query that made the offer, so a searched or paged candidate still counts.
4731 found = eligible_terminations(
4732 self.field_key,
4733 reader,
4734 profile=preview.profile,
4735 search=self.search,
4736 limit=self.limit,
4737 offset=self.offset,
4738 )
4739 except (PlanningTargetUnavailable, ValueError):
4740 raise PreviewCommandRefused(TERMINATION_UNRESOLVABLE) from None
4741 # The picker is the only legal source of a choice, so the write rechecks the offer.
4742 chosen = next(
4743 (
4744 candidate
4745 for candidate in found.candidates
4746 if candidate.pk == self.object_id and candidate._meta.label_lower == self.object_type
4747 ),
4748 None,
4749 )
4750 if chosen is None:
4751 raise PreviewCommandRefused("That termination is not one of the eligible candidates.")
4752 plan = save_termination_resolution_and_replan(
4753 profile=preview.profile,
4754 source_document=preview.document,
4755 actor=preview.actor,
4756 planning_context=preview.planning_context,
4757 task_type=SELECT_TERMINATION_TASK,
4758 field_key=self.field_key,
4759 source=workspace.termination_sources[self.field_key],
4760 selected_object_type=ObjectType.objects.get_for_model(type(chosen)),
4761 selected_object_id=chosen.pk,
4762 selected_display_name=str(chosen),
4763 device=chosen.device,
4764 reviewed_fingerprint=preview.reviewed_fingerprint,
4765 )
4766 return CommandOutcome(message=f"Termination resolved to '{chosen}'.", plan=plan)
4769class TraceResolveTerminationView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
4770 """Record one operator termination decision and replan the preview against it."""
4772 permission_required = "netbox_data_import.change_importprofile"
4774 def post(self, request):
4775 """Run the decision as one coordinated command."""
4776 try:
4777 object_id = int(request.POST.get("object_id", ""))
4778 except (TypeError, ValueError):
4779 raise PreviewCommandRefused("A termination selection names one object.") from None
4780 page = _candidate_page(request.POST)
4781 if page.error:
4782 raise PreviewCommandRefused(page.error)
4783 command = _ResolveTermination(
4784 field_key=request.POST.get("field_key", "").strip(),
4785 object_type=request.POST.get("object_type", "").strip(),
4786 object_id=object_id,
4787 search=request.POST.get("search", ""),
4788 limit=page.limit,
4789 offset=page.offset,
4790 )
4791 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
4792 return _command_response(request, result, self.refusal_url(request))
4795class InvalidProposalId(ValueError):
4796 """A proposal action received no integer id."""
4799class InvalidProposalTarget(Exception):
4800 """A proposal action does not identify a termination in this preview."""
4803INVALID_PROPOSAL_ID_ERROR = "Enter a valid proposal_id integer."
4806class _TraceProposalMixin(_TraceWorkspaceMixin):
4807 """Answer every proposal command and read in the workspace JSON envelope."""
4809 permission_denied_response_format = "json"
4811 def dispatch(self, request, *args, **kwargs):
4812 """Translate domain refusals into the workspace JSON envelope."""
4813 from .models import ResolutionProposal
4814 from .proposal_tasks import UnusableCandidateSet
4815 from .resolution_proposals import ActiveProposalExists
4816 from .termination_proposal import InvalidProposalCandidate, UnsupportedProposalRole
4818 try:
4819 return super().dispatch(request, *args, **kwargs)
4820 except InvalidProposalId:
4821 return JsonResponse({"ok": False, "error": INVALID_PROPOSAL_ID_ERROR}, status=400)
4822 except (Http404, ResolutionProposal.DoesNotExist):
4823 return JsonResponse({"ok": False, "error": "That proposal is no longer available."}, status=404)
4824 except ActiveProposalExists as exc:
4825 return JsonResponse({"ok": False, "error": exc.operator_message}, status=409)
4826 except UnusableCandidateSet as exc:
4827 return JsonResponse({"ok": False, "error": exc.operator_message, "reason": exc.reason}, status=400)
4828 except (InvalidProposalTarget, InvalidProposalCandidate, UnsupportedProposalRole) as exc:
4829 logger.warning("%s: termination refused: %s", type(self).__name__, exc)
4830 return JsonResponse({"ok": False, "error": TERMINATION_UNRESOLVABLE}, status=400)
4831 except ValidationError as exc:
4832 return JsonResponse({"ok": False, "error": "; ".join(exc.messages)}, status=400)
4835PROPOSAL_QUEUE_UNAVAILABLE = "The proposal queue is unavailable. Try again later."
4838def _refuse_without_backend() -> None:
4839 """Refuse a proposal request that no Inference Backend can answer, before any attempt row exists."""
4840 from .proposal_presentation import backend_unavailable_reason
4842 if reason := backend_unavailable_reason():
4843 raise PreviewCommandRefused(reason, 409)
4846class _ProposalRequests:
4847 """The live NetBox reads that every proposal request of one command shares."""
4849 def __init__(self, preview: LockedPreview):
4850 self.preview = preview
4852 @cached_property
4853 def reader(self):
4854 """Return the operator's scoped reader for the preview's import target."""
4855 return _trace_reader(self.preview.actor, self.preview.profile, self.preview.planning_context)
4857 @cached_property
4858 def inventories(self):
4859 """Return the inventory cache that every request of this command shares."""
4860 from .proposal_presentation import TerminationInventories
4862 return TerminationInventories(profile=self.preview.profile, reader=self.reader)
4864 @cached_property
4865 def live_terminations(self) -> dict:
4866 """Return each termination of a plan made against live NetBox now, by field key."""
4867 preview = self.preview
4868 live = ImportEngine.plan(preview.profile, preview.document, preview.actor, preview.planning_context)
4869 fields: dict = {}
4870 for trace in ReviewWorkspace(live, preview.actor).traces:
4871 for item in trace.terminations:
4872 fields.setdefault(item["field_key"], item)
4873 return fields
4875 def queue(self, field_key: str) -> None:
4876 """Recheck one field against live NetBox and create its attempt row, which waits for a request slot."""
4877 from core.models import ObjectType
4879 from .cable_target import UNRESOLVED
4880 from .inference_backend import proposal_candidate_limit
4881 from .proposal_jobs import PROMPT_VERSION
4882 from .proposal_response import RESPONSE_SCHEMA_VERSION
4883 from .resolution_proposals import (
4884 ActiveProposalExists,
4885 active_proposal_exists,
4886 next_page_offset,
4887 request_proposal,
4888 )
4890 profile, actor = self.preview.profile, self.preview.actor
4891 sources = self.preview.workspace.termination_sources
4892 if field_key not in sources:
4893 raise InvalidProposalTarget("This preview asked no question about that termination.")
4894 if sources[field_key] is None:
4895 raise InvalidProposalTarget(TERMINATION_UNRESOLVABLE)
4896 field = self.live_terminations.get(field_key)
4897 if field is None:
4898 raise InvalidProposalTarget("This field is no longer in the preview.")
4899 if field["state"] != UNRESOLVED:
4900 raise PreviewCommandRefused("This termination is already resolved.", 409)
4901 # Refuse on the observed predecessor, not on the index: see active_proposal_exists.
4902 if active_proposal_exists(profile=profile, task_type=SELECT_TERMINATION_TASK, field_key=field_key):
4903 raise ActiveProposalExists("This field already has an active Resolution Proposal.")
4904 inventory = self.inventories.get(field_key)
4905 device = inventory.resolved_device
4906 if device is None:
4907 raise PreviewCommandRefused("The resolved Device is unavailable or outside your view permission.", 409)
4908 if inventory.candidate_error is not None:
4909 raise inventory.candidate_error
4910 snapshot = inventory.candidate_snapshot
4911 if snapshot is None:
4912 raise PreviewCommandRefused("The eligible candidates are no longer available.", 409)
4913 # A dense Device is searched in turns, from after the last page that used itself up.
4914 snapshot = snapshot.with_page(
4915 offset=next_page_offset(
4916 profile=profile, task_type=SELECT_TERMINATION_TASK, field_key=field_key, inventory=inventory
4917 ),
4918 size=proposal_candidate_limit(),
4919 )
4920 request_proposal(
4921 profile=profile,
4922 task_type=SELECT_TERMINATION_TASK,
4923 field_key=field_key,
4924 source_evidence={**parse_termination_field_key(field_key), "label": field["label"]},
4925 resolved_device_type=ObjectType.objects.get_for_model(device),
4926 resolved_device_id=device.pk,
4927 prompt_version=PROMPT_VERSION,
4928 response_schema_version=RESPONSE_SCHEMA_VERSION,
4929 candidate_snapshot=snapshot,
4930 requested_by=actor,
4931 )
4934@dataclass(frozen=True)
4935class _RequestProposal(PreviewCommand):
4936 """Freeze one unresolved field's evidence and queue its inference; the plan itself does not change."""
4938 field_key: str
4939 replans = False
4940 # The plan is unchanged, so every claim the page holds stays current and the next field can ask.
4941 advances = False
4942 reviews_policy = False
4944 def apply(self, preview):
4945 """Queue one proposal for the field."""
4946 _refuse_without_backend()
4947 _ProposalRequests(preview).queue(self.field_key)
4948 return CommandOutcome(payload={"field_key": self.field_key})
4951@dataclass(frozen=True)
4952class _RequestAllProposals(PreviewCommand):
4953 """Queue a proposal for every open termination of the preview that can be asked now."""
4955 replans = False
4956 advances = False
4957 reviews_policy = False
4959 def apply(self, preview: LockedPreview) -> CommandOutcome:
4960 """Ask about each open termination once, and count each refusal by its reason instead of stopping."""
4961 from .cable_target import UNRESOLVED
4962 from .field_keys import TERMINATION_ROLE
4963 from .proposal_tasks import UnusableCandidateSet
4964 from .resolution_proposals import ActiveProposalExists
4965 from .termination_proposal import InvalidProposalCandidate, UnsupportedProposalRole
4967 _refuse_without_backend()
4968 fields = dict.fromkeys(
4969 item["field_key"]
4970 for trace in preview.workspace.traces
4971 for item in trace.terminations
4972 if item.get("state", UNRESOLVED) == UNRESOLVED
4973 and parse_termination_field_key(item["field_key"])["role"] == TERMINATION_ROLE
4974 )
4975 requests = _ProposalRequests(preview)
4976 queued = 0
4977 skipped: Counter[str] = Counter()
4978 for field_key in fields:
4979 try:
4980 requests.queue(field_key)
4981 except (InvalidProposalTarget, InvalidProposalCandidate, UnsupportedProposalRole):
4982 skipped[TERMINATION_UNRESOLVABLE] += 1
4983 except (PreviewCommandRefused, ActiveProposalExists, UnusableCandidateSet) as exc:
4984 skipped[exc.operator_message] += 1
4985 else:
4986 queued += 1
4987 return CommandOutcome(payload={"queued": queued, "skipped": skipped})
4990class TraceRequestProposalView(_TraceProposalMixin, PermissionRequiredMixin, View):
4991 """Freeze one unresolved field's evidence before dispatching inference."""
4993 permission_required = "netbox_data_import.change_importprofile"
4995 def post(self, request):
4996 """Create the attempt row, give it a Job when a request slot is free, and answer with the field's card."""
4997 from .proposal_dispatch import dispatch_proposals
4999 field_key = request.POST.get("field_key", "").strip()
5000 try:
5001 result = apply_preview_command(
5002 request, PreviewClaim.posted(request.POST), _RequestProposal(field_key=field_key)
5003 )
5004 dispatch_proposals()
5005 except (RedisConnectionError, RedisTimeoutError):
5006 logger.exception("Failed to enqueue a resolution proposal")
5007 return JsonResponse({"ok": False, "error": PROPOSAL_QUEUE_UNAVAILABLE}, status=503)
5008 return _proposal_card_after(request, result, field_key)
5011class TraceRequestAllProposalsView(_TraceWorkspaceMixin, PermissionRequiredMixin, View):
5012 """Ask AI about every open termination of the preview, and report what it skipped and why."""
5014 permission_required = "netbox_data_import.change_importprofile"
5016 def post(self, request):
5017 """Queue the proposals in one command, then show the workspace with the outcome."""
5018 from .proposal_dispatch import dispatch_proposals
5020 try:
5021 result = apply_preview_command(request, PreviewClaim.posted(request.POST), _RequestAllProposals())
5022 dispatch_proposals()
5023 except (RedisConnectionError, RedisTimeoutError):
5024 logger.exception("Failed to enqueue resolution proposals")
5025 return self._refusal(request, PROPOSAL_QUEUE_UNAVAILABLE, 503)
5026 queued, skipped = result.outcome.payload["queued"], result.outcome.payload["skipped"]
5027 if queued:
5028 messages.success(request, f"Asked AI about {queued} termination{pluralize(queued)}.")
5029 if skipped:
5030 total = sum(skipped.values())
5031 reasons = "; ".join(f"{reason} ({count})" for reason, count in skipped.items())
5032 messages.warning(request, f"Skipped {total} termination{pluralize(total)}: {reasons}")
5033 if not queued and not skipped:
5034 messages.info(request, "No open termination needs a proposal.")
5035 return redirect(_trace_workspace_url(request.POST.get("trace", "")))
5038class TraceProposalView(_TraceProposalMixin, PermissionRequiredMixin, View):
5039 """Read all attempts for a field, independent of their originating plans."""
5041 permission_required = "netbox_data_import.view_importprofile"
5043 def get(self, request):
5044 """Answer with the field's card as it stands now, which a pending card polls."""
5045 from .proposal_dispatch import dispatch_proposals_on_poll
5047 snapshot = _claimed_snapshot(request, profile_action="view")
5048 try:
5049 dispatch_proposals_on_poll()
5050 except (RedisConnectionError, RedisTimeoutError):
5051 logger.warning("The proposal queue did not answer; the card shows its proposal as it stands.")
5052 return _proposal_card(request, snapshot, request.GET.get("field_key", "").strip(), request.GET.get("trace", ""))
5055@dataclass(frozen=True)
5056class _ProposalAction(PreviewCommand):
5057 """Apply one operator action to a proposal of this preview's profile."""
5059 proposal_id: int
5060 replans = False
5061 # An action that leaves the plan unchanged keeps every claim on the page current.
5062 advances = False
5063 reviews_policy = False
5064 message: ClassVar[str] = ""
5066 def apply(self, preview):
5067 """Refuse an unavailable attempt or a transition another operator already took."""
5068 from .models import ResolutionProposal
5070 proposal = ResolutionProposal.objects.get(
5071 pk=self.proposal_id, profile=preview.profile, task_type=SELECT_TERMINATION_TASK
5072 )
5073 if proposal.field_key not in preview.workspace.termination_sources:
5074 raise InvalidProposalTarget("This preview asked no question about that termination.")
5075 if not self.decide(proposal, preview):
5076 raise PreviewCommandRefused(
5077 "This proposal no longer permits that action. Re-read it before continuing.", 409
5078 )
5079 return CommandOutcome(message=self.message, payload={"field_key": proposal.field_key})
5081 def decide(self, proposal, preview) -> bool:
5082 """Apply the action, returning whether this caller moved the proposal."""
5083 raise NotImplementedError
5086class _TraceProposalActionView(_TraceProposalMixin, PermissionRequiredMixin, View):
5087 """Run one proposal action as a coordinated command."""
5089 permission_required = "netbox_data_import.change_importprofile"
5090 command_class: type[_ProposalAction]
5092 def post(self, request):
5093 """Run the action, then answer with what it changed."""
5094 try:
5095 proposal_id = int(request.POST.get("proposal_id", ""))
5096 except ValueError:
5097 raise InvalidProposalId(INVALID_PROPOSAL_ID_ERROR) from None
5098 result = apply_preview_command(request, PreviewClaim.posted(request.POST), self.command_class(proposal_id))
5099 return self.answer(request, result)
5101 def answer(self, request, result):
5102 """Answer with the card the action changed, because the plan and the claim stay the same."""
5103 return _proposal_card_after(request, result, result.outcome.payload["field_key"])
5106@dataclass(frozen=True)
5107class _CancelProposal(_ProposalAction):
5108 """Cancel active work, for any operator with preview and resolved Device access."""
5110 def decide(self, proposal, preview):
5111 """Cancel through the lifecycle's conditional transition."""
5112 from .proposal_tasks import proposal_task
5113 from .resolution_proposals import cancel_proposal
5115 reader = _trace_reader(preview.actor, preview.profile, preview.planning_context)
5116 if (
5117 proposal_task(proposal.task_type).resolved_device(
5118 profile=proposal.profile, field_key=proposal.field_key, netbox_reader=reader
5119 )
5120 is None
5121 ):
5122 raise ObjectPermissionDenied("dcim.view_device")
5123 return cancel_proposal(proposal.pk)
5126@dataclass(frozen=True)
5127class _AcceptProposal(_ProposalAction):
5128 """Write the accepted resolution; the coordinator replans in the same transaction."""
5130 replans = True
5131 advances = True
5132 reviews_policy = True
5133 message = "The proposal was accepted and the preview was replanned."
5135 def decide(self, proposal, preview):
5136 """Record the accepted resolution through the existing acceptance transaction."""
5137 from .proposal_decisions import accept_proposal
5139 if preview.workspace.termination_sources[proposal.field_key] is None:
5140 raise InvalidProposalTarget(TERMINATION_UNRESOLVABLE)
5141 return accept_proposal(
5142 proposal.pk,
5143 source=preview.workspace.termination_sources[proposal.field_key],
5144 operator=preview.actor,
5145 netbox_reader=_trace_reader(preview.actor, preview.profile, preview.planning_context),
5146 reviewed_fingerprint=preview.reviewed_fingerprint,
5147 )
5150@dataclass(frozen=True)
5151class _RejectProposal(_ProposalAction):
5152 """Record the explicit rejection without binding it to the requesting operator."""
5154 profile_action = "view"
5156 def decide(self, proposal, preview):
5157 """Reject through the permission-checked decision service."""
5158 from .proposal_decisions import reject_proposal
5160 return reject_proposal(proposal.pk, operator=preview.actor)
5163class TraceCancelProposalView(_TraceProposalActionView):
5164 """Allow any operator with preview and resolved Device access to cancel active work."""
5166 command_class = _CancelProposal
5169class TraceAcceptProposalView(_TraceProposalActionView):
5170 """Accept a proposal and replan the preview in one coordinated transaction."""
5172 command_class = _AcceptProposal
5174 def answer(self, request, result):
5175 """Show the replanned workspace, whose every form carries the claim the acceptance advanced to."""
5176 return _command_response(request, result, _trace_workspace_url(request.POST.get("trace", "")))
5179class TraceRejectProposalView(_TraceProposalActionView):
5180 """Record the explicit rejection without binding it to the requesting operator."""
5182 permission_required = "netbox_data_import.view_importprofile"
5183 command_class = _RejectProposal
5186@dataclass(frozen=True)
5187class _MapManufacturer(PreviewCommand):
5188 """Save a ManufacturerMapping (source make to NetBox manufacturer slug) from the preview page."""
5190 source_make: str
5191 netbox_mfg_slug: str
5193 def apply(self, preview):
5194 """Validate and save the mapping."""
5195 mapping = _get_or_init(ManufacturerMapping, profile=preview.profile, source_make=self.source_make)
5196 mapping.netbox_manufacturer_slug = self.netbox_mfg_slug
5197 _validate_model_instance(mapping, f"manufacturer mapping '{self.source_make}'")
5198 result = save_permission_scoped_object(
5199 preview.actor,
5200 ManufacturerMapping,
5201 {"profile": preview.profile, "source_make": self.source_make},
5202 {"netbox_manufacturer_slug": self.netbox_mfg_slug},
5203 )
5204 verb = "Created" if result.created else "Updated"
5205 return CommandOutcome(message=f"{verb} manufacturer mapping: '{self.source_make}' → {self.netbox_mfg_slug}")
5208class QuickResolveManufacturerView(_PreviewCommandMixin, PermissionRequiredMixin, View):
5209 """Save a ManufacturerMapping (source make → NetBox manufacturer slug) from the preview page."""
5211 permission_required = "netbox_data_import.change_importprofile"
5213 def post(self, request):
5214 """Save the manufacturer mapping and replan the preview."""
5215 source_make = " ".join(request.POST.get("source_make", "").split())
5216 netbox_mfg_slug = request.POST.get("netbox_mfg_slug", "").strip()
5217 if not source_make or not netbox_mfg_slug:
5218 raise PreviewCommandRefused("Source make and NetBox manufacturer slug are required.")
5219 command = _MapManufacturer(source_make=source_make, netbox_mfg_slug=netbox_mfg_slug)
5220 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
5221 return _command_response(request, result, reverse("plugins:netbox_data_import:import_preview"))
5223 def refusal_url(self, request):
5224 """Return the preview the quick fix came from."""
5225 return reverse("plugins:netbox_data_import:import_preview")
5228@dataclass(frozen=True)
5229class _MapDeviceType(PreviewCommand):
5230 """Save a DeviceTypeMapping (source make and model to NetBox slugs) from the preview page."""
5232 source_make: str
5233 source_model: str
5234 netbox_mfg_slug: str
5235 netbox_dt_slug: str
5237 def apply(self, preview):
5238 """Validate and save the mapping."""
5239 mapping = _get_or_init(
5240 DeviceTypeMapping, profile=preview.profile, source_make=self.source_make, source_model=self.source_model
5241 )
5242 mapping.netbox_manufacturer_slug = self.netbox_mfg_slug
5243 mapping.netbox_device_type_slug = self.netbox_dt_slug
5244 _validate_model_instance(mapping, f"device type mapping '{self.source_make} / {self.source_model}'")
5245 result = save_permission_scoped_object(
5246 preview.actor,
5247 DeviceTypeMapping,
5248 {"profile": preview.profile, "source_make": self.source_make, "source_model": self.source_model},
5249 {"netbox_manufacturer_slug": self.netbox_mfg_slug, "netbox_device_type_slug": self.netbox_dt_slug},
5250 )
5251 verb = "created" if result.created else "updated"
5252 return CommandOutcome(
5253 message=(
5254 f"DeviceType mapping {verb}: '{self.source_make} / {self.source_model}' "
5255 f"→ {self.netbox_mfg_slug}/{self.netbox_dt_slug}"
5256 )
5257 )
5260class QuickResolveDeviceTypeView(_PreviewCommandMixin, PermissionRequiredMixin, View):
5261 """Save a DeviceTypeMapping (source make/model → NetBox slugs) from the preview page."""
5263 permission_required = "netbox_data_import.change_importprofile"
5265 def post(self, request):
5266 """Save the device type mapping and replan the preview."""
5267 from .device_identity import default_identity_slugs
5269 source_make = " ".join(request.POST.get("source_make", "").split())
5270 source_model = " ".join(request.POST.get("source_model", "").split())
5271 if not source_make or not source_model:
5272 raise PreviewCommandRefused("Source make and model are required.")
5273 if request.POST.get("action", "map") != "map":
5274 raise PreviewCommandRefused("The requested Device Type action is not supported.")
5275 # The importer derives both slugs the same way, so a default that differs maps to nothing.
5276 default_mfg_slug, default_dt_slug = default_identity_slugs(source_make, source_model)
5277 command = _MapDeviceType(
5278 source_make=source_make,
5279 source_model=source_model,
5280 netbox_mfg_slug=request.POST.get("netbox_mfg_slug", "").strip() or default_mfg_slug,
5281 netbox_dt_slug=request.POST.get("netbox_dt_slug", "").strip() or default_dt_slug,
5282 )
5283 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
5284 return _command_response(request, result, reverse("plugins:netbox_data_import:import_preview"))
5286 def refusal_url(self, request):
5287 """Return the preview the quick fix came from."""
5288 return reverse("plugins:netbox_data_import:import_preview")
5291@dataclass(frozen=True)
5292class _MapClassRole(PreviewCommand):
5293 """Save a ClassRoleMapping (ignore, role or rack) from an error row of the preview."""
5295 source_class: str
5296 mapping_action: str
5297 role_slug: str
5298 rack_type_id: int | None
5300 def apply(self, preview):
5301 """Validate and save the mapping."""
5302 from dcim.models import RackType
5304 rack_type = None
5305 if self.rack_type_id is not None:
5306 rack_type = RackType.objects.filter(pk=self.rack_type_id).first()
5307 if rack_type is None:
5308 raise PreviewCommandRefused(
5309 f"Invalid rack type selected for class '{self.source_class}'. Please choose a valid rack type."
5310 )
5311 values = {
5312 "ignore": self.mapping_action == "ignore",
5313 "creates_rack": self.mapping_action == "rack",
5314 "rack_type": rack_type,
5315 "role_slug": self.role_slug if self.mapping_action == "role" else "",
5316 }
5317 mapping = _get_or_init(ClassRoleMapping, profile=preview.profile, source_class=self.source_class)
5318 for field_name, value in values.items():
5319 setattr(mapping, field_name, value)
5320 _validate_model_instance(mapping, f"class role mapping '{self.source_class}'")
5321 result = save_permission_scoped_object(
5322 preview.actor,
5323 ClassRoleMapping,
5324 {"profile": preview.profile, "source_class": self.source_class},
5325 values,
5326 )
5327 verb = "Created" if result.created else "Updated"
5328 if self.mapping_action == "ignore":
5329 action_label = "ignore"
5330 elif self.mapping_action == "rack":
5331 action_label = f"creates rack{f' (type: {rack_type})' if rack_type else ''}"
5332 else:
5333 action_label = f"role '{self.role_slug}'"
5334 return CommandOutcome(message=f"{verb} mapping: class '{self.source_class}' → {action_label}")
5337class QuickAddClassRoleMappingView(_PreviewCommandMixin, PermissionRequiredMixin, View):
5338 """Quickly add a ClassRoleMapping (ignore / role) directly from an error row in preview."""
5340 permission_required = "netbox_data_import.change_importprofile"
5342 def post(self, request):
5343 """Save the class-to-role mapping and replan the preview."""
5344 source_class = request.POST.get("source_class", "").strip()
5345 mapping_action = request.POST.get("mapping_action", "ignore") # "ignore", "role", or "rack"
5346 role_slug = request.POST.get("role_slug", "").strip()
5347 raw_rack_type = request.POST.get("rack_type_id", "").strip()
5348 rack_type_id = None
5349 if mapping_action == "rack" and raw_rack_type:
5350 try:
5351 rack_type_id = int(raw_rack_type)
5352 except (TypeError, ValueError):
5353 raise PreviewCommandRefused(
5354 f"Invalid rack type selected for class '{source_class}'. Please choose a valid rack type."
5355 ) from None
5356 if not source_class:
5357 raise PreviewCommandRefused("Source class is required.")
5358 valid_actions = ("ignore", "role", "rack")
5359 if mapping_action not in valid_actions:
5360 raise PreviewCommandRefused(
5361 f"Invalid mapping action '{mapping_action}'. Must be one of: {', '.join(valid_actions)}."
5362 )
5363 if mapping_action == "role" and not role_slug:
5364 raise PreviewCommandRefused("A role slug is required when mapping action is 'role'.")
5365 command = _MapClassRole(
5366 source_class=source_class, mapping_action=mapping_action, role_slug=role_slug, rack_type_id=rack_type_id
5367 )
5368 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
5369 return _command_response(request, result, reverse("plugins:netbox_data_import:import_preview"))
5371 def refusal_url(self, request):
5372 """Return the preview the quick fix came from."""
5373 return reverse("plugins:netbox_data_import:import_preview")
5376@dataclass(frozen=True)
5377class _MapColumn(PreviewCommand):
5378 """Map one unmapped source column to a NetBox target field from the preview panel."""
5380 source_column: str
5381 target_field: str
5383 def apply(self, preview):
5384 """Validate and save the mapping, replacing the column that supplied a direct target before."""
5385 if not CATALOG.is_valid(self.target_field, output_kinds=preview.profile.output_kinds):
5386 raise PreviewCommandRefused("Valid source column and target field are required.")
5387 # The catalog accepts any non-empty name after a family prefix, so it cannot bound length.
5388 # Validate before the displaced row is deleted: an invalid write must strand nothing.
5389 _validate_model_instance(
5390 ColumnMapping(profile=preview.profile, source_column=self.source_column, target_field=self.target_field),
5391 f"column mapping '{self.source_column}' -> {self.target_field}",
5392 )
5393 lookup = {"profile": preview.profile, "source_column": self.source_column, "target_field": self.target_field}
5394 if self.target_field.startswith(CANDIDATE_TARGET_PREFIX):
5395 result = save_permission_scoped_object(preview.actor, ColumnMapping, lookup, {}, on_existing="keep")
5396 verb = "Created" if result.created else "Kept"
5397 return CommandOutcome(message=f"{verb} candidate mapping: '{self.source_column}' → {self.target_field}")
5398 # A quick direct mapping replaces the source column that supplied the target before it.
5399 displaced = ColumnMapping.objects.filter(profile=preview.profile, target_field=self.target_field).exclude(
5400 source_column=self.source_column
5401 )
5402 displaced_source = displaced.values_list("source_column", flat=True).first()
5403 delete_permission_scoped_objects(preview.actor, displaced)
5404 result = save_permission_scoped_object(preview.actor, ColumnMapping, lookup, {}, on_existing="keep")
5405 if displaced_source:
5406 return CommandOutcome(
5407 message=(
5408 f"Reassigned: '{self.source_column}' → {self.target_field} "
5409 f"(previously mapped from '{displaced_source}')"
5410 )
5411 )
5412 verb = "Created" if result.created else "Kept"
5413 return CommandOutcome(message=f"{verb} mapping: '{self.source_column}' → {self.target_field}")
5416class QuickAddColumnMappingView(_PreviewCommandMixin, PermissionRequiredMixin, View):
5417 """Quickly map an unmapped source column to a NetBox target field from the preview panel."""
5419 permission_required = "netbox_data_import.change_importprofile"
5421 def post(self, request):
5422 """Save the column mapping and replan the preview."""
5423 source_column = request.POST.get("source_column", "").strip()
5424 target_field = request.POST.get("target_field", "").strip()
5425 if not source_column or not target_field:
5426 raise PreviewCommandRefused("Valid source column and target field are required.")
5427 command = _MapColumn(source_column=source_column, target_field=target_field)
5428 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
5429 return _command_response(request, result, reverse("plugins:netbox_data_import:import_preview"))
5431 def refusal_url(self, request):
5432 """Return the preview the quick fix came from."""
5433 return reverse("plugins:netbox_data_import:import_preview")
5436@dataclass(frozen=True)
5437class _MatchExistingDevice(PreviewCommand):
5438 """Link one source row to an existing NetBox Device, so the next plan updates that Device."""
5440 source_id: str
5441 netbox_device_id: str
5443 def apply(self, preview):
5444 """Check the Device against the import site and the profile's other links, then save the link."""
5445 from dcim.models import Device
5447 source_row = _source_row(preview.workspace, self.source_id)
5448 try:
5449 device = Device.objects.restrict(preview.actor, "view").get(pk=int(self.netbox_device_id))
5450 except (Device.DoesNotExist, ValueError):
5451 raise PreviewCommandRefused(f"Device #{self.netbox_device_id} not found.") from None
5452 if device.site_id != preview.planning_context["site_id"]:
5453 raise PreviewCommandRefused("The selected device is outside the active import site.")
5454 conflicting = preview.profile.device_matches.filter(netbox_device_id=device.pk).exclude(
5455 source_id=self.source_id
5456 )
5457 if conflicting_match := conflicting.first():
5458 raise PreviewCommandRefused(
5459 f"Device '{device.name}' is already linked to source '{conflicting_match.source_id}'."
5460 )
5461 try:
5462 save_permission_scoped_object(
5463 preview.actor,
5464 DeviceExistingMatch,
5465 {"profile": preview.profile, "source_id": self.source_id},
5466 {
5467 "netbox_device_id": device.pk,
5468 "device_name": device.name,
5469 "source_asset_tag": source_text(source_row.get("asset_tag"))[:50],
5470 },
5471 )
5472 except ObjectPermissionDenied as exc:
5473 raise PreviewCommandRefused("Permission denied: cannot create or change this device link.", 403) from exc
5474 except ValidationError as exc:
5475 raise PreviewCommandRefused("; ".join(exc.messages)) from exc
5476 return CommandOutcome(message=f"Source '{self.source_id}' linked to existing device '{device.name}'.")
5479class MatchExistingDeviceView(_PreviewCommandMixin, PermissionRequiredMixin, View):
5480 """Link a source row to an existing NetBox device (by device ID)."""
5482 permission_required = (
5483 "netbox_data_import.change_importprofile",
5484 "dcim.view_device",
5485 )
5487 def post(self, request):
5488 """Save the device match and go back to the replanned preview."""
5489 source_id = source_text(request.POST.get("source_id"))
5490 netbox_device_id = request.POST.get("netbox_device_id", "").strip()
5491 if not source_id or not netbox_device_id:
5492 raise PreviewCommandRefused("source_id and netbox_device_id are required.")
5493 command = _MatchExistingDevice(source_id=source_id, netbox_device_id=netbox_device_id)
5494 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
5495 return _command_response(request, result, reverse("plugins:netbox_data_import:import_preview"))
5497 def refusal_url(self, request):
5498 """Return the preview the link came from."""
5499 return reverse("plugins:netbox_data_import:import_preview")
5502def _device_name_filter(q: str):
5503 """Build a Django Q filter for device name search.
5505 Exact icontains is tried first; when the query contains separators (-, _, .)
5506 individual tokens (≥3 chars) are OR-ed in so that e.g. "EXAMPLE-SITE03-SW3"
5507 matches "edge-site03-switch03.lab.example.invalid" via the "SITE03" token.
5508 """
5509 import re as _re
5511 from django.db.models import Q as _Q
5513 base = _Q(name__icontains=q)
5514 tokens = [t for t in _re.split(r"[-_.\s]+", q) if len(t) >= 3]
5515 if len(tokens) > 1:
5516 token_q = _Q()
5517 for tok in tokens:
5518 token_q |= _Q(name__icontains=tok)
5519 return base | token_q
5520 return base
5523class SearchNetBoxObjectsView(_AjaxPermissionView):
5524 """AJAX search endpoint for NetBox objects used in preview quick-fix modals.
5526 GET params: type (manufacturer|device_type|device|role|rack_type), q (search string).
5527 Returns JSON list of {id, name, slug, url} dicts.
5528 """
5530 permission_required = "netbox_data_import.view_importprofile"
5532 def get(self, request):
5533 """Return a JSON list of matching NetBox objects for the given type and query."""
5534 from dcim.models import DeviceRole, DeviceType, Manufacturer, RackType
5535 from django.http import JsonResponse
5537 obj_type = request.GET.get("type", "device")
5538 q = request.GET.get("q", "").strip()
5539 limit = 20
5541 _perm_map = {
5542 "manufacturer": "dcim.view_manufacturer",
5543 "device_type": "dcim.view_devicetype",
5544 "device": "dcim.view_device",
5545 "role": "dcim.view_devicerole",
5546 "rack_type": "dcim.view_racktype",
5547 }
5548 required_perm = _perm_map.get(obj_type)
5549 if required_perm and not request.user.has_perm(required_perm): # pragma: no cover
5550 return JsonResponse({"results": [], "error": "permission_denied"}, status=403)
5552 if not q:
5553 return JsonResponse({"results": []})
5555 results = []
5556 if obj_type == "manufacturer":
5557 for mfg in Manufacturer.objects.filter(name__icontains=q)[:limit]:
5558 results.append(
5559 {
5560 "id": mfg.pk,
5561 "name": mfg.name,
5562 "slug": mfg.slug,
5563 "url": request.build_absolute_uri(mfg.get_absolute_url()),
5564 }
5565 )
5566 elif obj_type == "device_type":
5567 mfg_filter = request.GET.get("mfg_slug", "")
5568 qs = DeviceType.objects.select_related("manufacturer")
5569 if mfg_filter:
5570 qs = qs.filter(manufacturer__slug=mfg_filter)
5571 for dt in qs.filter(model__icontains=q)[:limit]:
5572 results.append(
5573 {
5574 "id": dt.pk,
5575 "name": f"{dt.manufacturer.name} / {dt.model}",
5576 "slug": dt.slug,
5577 "mfg_slug": dt.manufacturer.slug,
5578 "url": request.build_absolute_uri(dt.get_absolute_url()),
5579 }
5580 )
5581 elif obj_type == "device":
5582 self._search_devices(request, q, limit, results)
5583 elif obj_type == "role":
5584 for role in DeviceRole.objects.filter(name__icontains=q)[:limit]:
5585 results.append(
5586 {
5587 "id": role.pk,
5588 "name": role.name,
5589 "slug": role.slug,
5590 "url": request.build_absolute_uri(role.get_absolute_url()),
5591 }
5592 )
5593 elif obj_type == "rack_type":
5594 from django.db.models import Q
5596 qs = RackType.objects.select_related("manufacturer").filter(
5597 Q(model__icontains=q) | Q(manufacturer__name__icontains=q) | Q(slug__icontains=q)
5598 )[:limit]
5599 for rt in qs:
5600 results.append(
5601 {
5602 "id": rt.pk,
5603 "name": f"{rt.manufacturer.name} / {rt.model}" if rt.manufacturer else rt.model,
5604 "slug": rt.slug,
5605 "url": request.build_absolute_uri(rt.get_absolute_url()),
5606 }
5607 )
5609 return JsonResponse({"results": results})
5611 def _search_devices(self, request, q, limit, results):
5612 """Two-phase device search: full-string matches first, then token matches.
5614 This prevents a relevant exact-substring match (e.g. "example-zone03d-rc1")
5615 from being pushed off the result list by noisy short tokens like "rc1"
5616 or "prod" that match many devices.
5617 """
5618 from dcim.models import Device
5620 visible_devices = Device.objects.restrict(request.user, "view")
5621 base_qs = visible_devices.filter(name__icontains=q).distinct().select_related("site").order_by("name")
5622 seen_ids = set()
5623 for dev in base_qs[:limit]:
5624 seen_ids.add(dev.pk)
5625 results.append(
5626 {
5627 "id": dev.pk,
5628 "name": dev.name,
5629 "serial": dev.serial or None,
5630 "site": dev.site.name if dev.site else "",
5631 "url": request.build_absolute_uri(dev.get_absolute_url()),
5632 }
5633 )
5634 if len(results) >= limit:
5635 return
5636 token_qs = (
5637 visible_devices.filter(_device_name_filter(q))
5638 .exclude(pk__in=seen_ids)
5639 .distinct()
5640 .select_related("site")
5641 .order_by("name")
5642 )
5643 for dev in token_qs[: limit - len(results)]:
5644 results.append(
5645 {
5646 "id": dev.pk,
5647 "name": dev.name,
5648 "serial": dev.serial or None,
5649 "site": dev.site.name if dev.site else "",
5650 "url": request.build_absolute_uri(dev.get_absolute_url()),
5651 }
5652 )
5655@dataclass(frozen=True)
5656class _CreateDeviceRole(PreviewCommand):
5657 """Create one DeviceRole from the Configure Class modal; no profile policy or plan changes."""
5659 name: str
5660 slug: str
5661 color: str
5662 replans = False
5663 reviews_policy = False
5664 reads_plan = False
5666 def apply(self, preview):
5667 """Create the role, or keep one that already holds the slug."""
5668 from dcim.models import DeviceRole
5670 role = _get_or_init(DeviceRole, slug=self.slug)
5671 if role.pk is None:
5672 role.name = self.name
5673 role.color = self.color
5674 role.full_clean(validate_unique=False)
5675 result = save_permission_scoped_object(
5676 preview.actor, DeviceRole, {"slug": self.slug}, {"name": self.name, "color": self.color}, on_existing="keep"
5677 )
5678 role = result.instance
5679 return CommandOutcome(payload={"id": role.pk, "name": role.name, "slug": role.slug, "created": result.created})
5682class QuickCreateDeviceRoleView(_PreviewCommandMixin, _AjaxPermissionView):
5683 """AJAX endpoint: create a new DeviceRole and return its details as JSON.
5685 Used by the Configure Class modal so operators can create missing roles
5686 without leaving the import preview page.
5687 """
5689 permission_required = "netbox_data_import.change_importprofile"
5691 def post(self, request):
5692 """Create the DeviceRole and return JSON {id, name, slug}."""
5693 import re
5695 name = request.POST.get("name", "").strip()
5696 slug = request.POST.get("slug", "").strip()
5697 color = request.POST.get("color", "9e9e9e").strip() or "9e9e9e"
5698 if not name or not slug:
5699 return JsonResponse({"error": "Role name and slug are required."}, status=400)
5700 if not re.match(r"^[-a-z0-9_]+$", slug):
5701 return JsonResponse(
5702 {"error": "Slug may only contain lowercase letters, numbers, hyphens, and underscores."}, status=400
5703 )
5704 try:
5705 result = apply_preview_command(
5706 request, PreviewClaim.posted(request.POST), _CreateDeviceRole(name=name, slug=slug, color=color)
5707 )
5708 except IntegrityError:
5709 logger.exception("QuickCreateDeviceRoleView: integrity error creating role slug=%s", slug)
5710 return JsonResponse({"error": "A device role with that slug already exists."}, status=400)
5711 except ValidationError:
5712 logger.exception("QuickCreateDeviceRoleView: validation error creating role slug=%s", slug)
5713 return JsonResponse({"error": "Invalid role data."}, status=400)
5714 except DatabaseError:
5715 logger.exception("QuickCreateDeviceRoleView: database error creating role slug=%s", slug)
5716 return JsonResponse({"error": "An internal error occurred."}, status=500)
5717 messages.success(request, f"Device role '{result.outcome.payload['name']}' is available.")
5718 return JsonResponse(dict(result.outcome.payload))
5721@dataclass(frozen=True)
5722class _AutoMatchDevices(PreviewCommand):
5723 """Save every safe exact Device match for the preview's eligible source rows."""
5725 def apply(self, preview):
5726 """Match against the import target the preview names, as the operator sees it now."""
5727 from .review_workspace import auto_match_devices
5729 target = _resolved_import_target(preview.planning_context, preview.actor)
5730 if target is None:
5731 raise PreviewCommandRefused(TARGET_GONE, 409)
5732 summary = auto_match_devices(preview.workspace, preview.profile, preview.actor, target)
5733 return CommandOutcome(message=summary.message())
5736class AutoMatchDevicesView(_PreviewCommandMixin, PermissionRequiredMixin, View):
5737 """Run the Review Workspace device auto-match command."""
5739 permission_required = (
5740 "netbox_data_import.change_importprofile",
5741 "netbox_data_import.add_deviceexistingmatch",
5742 "dcim.view_device",
5743 )
5745 def post(self, request):
5746 """Run auto-matching and go back to the replanned preview with a summary message."""
5747 result = apply_preview_command(request, PreviewClaim.posted(request.POST), _AutoMatchDevices())
5748 return _command_response(request, result, reverse("plugins:netbox_data_import:import_preview"))
5750 def refusal_url(self, request):
5751 """Return the preview the command came from."""
5752 return reverse("plugins:netbox_data_import:import_preview")
5755@dataclass(frozen=True)
5756class _SyncSingleRow(PreviewCommand):
5757 """Execute one row's Synchronization Unit and replan inside the execution savepoint."""
5759 row_number: int
5761 def apply(self, preview):
5762 """Run the selection; a failure keeps its audit row and leaves the preview as it was."""
5763 unit = next(
5764 (
5765 unit
5766 for unit in preview.workspace.units
5767 if unit.row_number == self.row_number and unit.object_type in {"device", "rack"}
5768 ),
5769 None,
5770 )
5771 if unit is None:
5772 raise PreviewCommandRefused("Row not found in current preview data")
5773 if unit.action not in {"create", "update"}:
5774 raise PreviewCommandRefused("Only 'create' and 'update' rows can be synced individually")
5775 try:
5776 _execution, plan = ImportEngine.execute_and_replan(
5777 preview.profile,
5778 preview.document,
5779 preview.plan.to_dict(),
5780 [unit.identity],
5781 uuid.uuid4().hex,
5782 preview.actor,
5783 validate_replan=validate_preview_plan,
5784 )
5785 except Exception as exc: # noqa: BLE001 - the coordinator commits the failed audit, then raises this
5786 # The savepoint rolled back, so NetBox must not send the events its writes queued.
5787 clear_events.send(sender=type(self))
5788 return CommandOutcome(refusal=_RowWriteRefused(_refused_row_write_response(exc, self.row_number)))
5789 verb = "created" if unit.action == "create" else "updated"
5790 return CommandOutcome(
5791 message="Synchronized.",
5792 payload={"detail": f"{unit.object_type.capitalize()} '{unit.name}' was {verb} in NetBox."},
5793 plan=plan,
5794 )
5797class _RowWriteRefused(Exception):
5798 """A single-row execution that failed after its audit row was written, carrying the answer for it."""
5800 def __init__(self, response):
5801 super().__init__("The row synchronization was refused.")
5802 self.response = response
5805def _refused_row_write_response(exc, row_number):
5806 """Return the answer one refused row write gives the operator.
5808 The worker reports the same failures, so both read the message from one place.
5809 """
5810 if isinstance(exc, PreviewCommandRefused):
5811 return JsonResponse({"ok": False, "error": exc.operator_message}, status=exc.status)
5812 if isinstance(exc, PlanError):
5813 logger.warning("SyncSingleRowView: the Import Plan for row_number=%s is unreadable.", row_number, exc_info=exc)
5814 return JsonResponse({"ok": False, "error": UNREADABLE_PREVIEW}, status=409)
5815 if isinstance(exc, (PlanningTargetUnavailable, PreconditionFailed, SelectionError, StalePlan, StaleSourceDocument)):
5816 return JsonResponse({"ok": False, "error": exc.operator_message}, status=409)
5817 if isinstance(exc, (DatabaseError, ObjectPermissionDenied, ValidationError)):
5818 if isinstance(exc, DatabaseError):
5819 logger.error("SyncSingleRowView: database error for row_number=%s", row_number, exc_info=exc)
5820 return JsonResponse({"ok": False, "error": operator_failure_message(exc)}, status=400)
5821 logger.error("SyncSingleRowView: unexpected error for row_number=%s", row_number, exc_info=exc)
5822 return JsonResponse({"ok": False, "error": "An unexpected error occurred. See server logs."}, status=500)
5825class SyncSingleRowView(_PreviewCommandMixin, _AjaxPermissionView):
5826 """AJAX endpoint: execute a single row of the active preview.
5828 POST body: the preview claim and row_number=<int>
5829 Returns the preview-command JSON envelope.
5830 """
5832 permission_required = "netbox_data_import.change_importprofile"
5834 def post(self, request):
5835 """Execute one selected Synchronization Unit and return JSON."""
5836 raw_row_number = request.POST.get("row_number")
5837 if raw_row_number is None:
5838 return JsonResponse({"ok": False, "error": "row_number is required"}, status=400)
5839 try:
5840 row_number = int(raw_row_number)
5841 except (TypeError, ValueError):
5842 return JsonResponse({"ok": False, "error": "Invalid row number"}, status=400)
5843 try:
5844 result = apply_preview_command(request, PreviewClaim.posted(request.POST), _SyncSingleRow(row_number))
5845 except _RowWriteRefused as refused:
5846 return refused.response
5847 messages.success(request, f"{result.outcome.message} {result.outcome.payload['detail']}")
5848 return JsonResponse(
5849 {
5850 "ok": True,
5851 "row_number": row_number,
5852 "preview_state": "replanned",
5853 "message": result.outcome.message,
5854 "detail": result.outcome.payload["detail"],
5855 }
5856 )
5859@dataclass(frozen=True)
5860class _UnlinkDevice(PreviewCommand):
5861 """Remove one source row's Device link, and the field reviews that rest on it."""
5863 source_id: str
5865 def apply(self, preview):
5866 """Delete the link and its reviews under their row locks, refusing either the actor may not delete."""
5867 _source_row(preview.workspace, self.source_id)
5868 binding = (
5869 DeviceExistingMatch.objects.select_for_update()
5870 .filter(profile=preview.profile, source_id=self.source_id)
5871 .first()
5872 )
5873 reviews = list(
5874 IgnoredFieldDifference.objects.select_for_update().filter(profile=preview.profile, source_id=self.source_id)
5875 )
5876 if binding is not None and not preview.actor.has_perm("netbox_data_import.delete_deviceexistingmatch", binding):
5877 raise PreviewCommandRefused("Permission denied: cannot delete this device link.", 403)
5878 if any(not preview.actor.has_perm("netbox_data_import.delete_ignoredfielddifference", r) for r in reviews):
5879 raise PreviewCommandRefused("Permission denied: cannot remove the dependent field reviews.", 403)
5880 if binding is None and not reviews:
5881 raise PreviewCommandRefused(f"Source '{self.source_id}' has no device link to remove.")
5882 if reviews:
5883 IgnoredFieldDifference.objects.filter(pk__in=[review.pk for review in reviews]).delete()
5884 if binding is not None:
5885 binding.delete()
5886 if reviews:
5887 return CommandOutcome(
5888 message=f"Unlinked source '{self.source_id}' and removed {len(reviews)} field review(s)."
5889 )
5890 return CommandOutcome(message=f"Unlinked source '{self.source_id}'.")
5893class UnlinkDeviceView(_PreviewCommandMixin, _AjaxPermissionView):
5894 """Remove a DeviceExistingMatch (unlink a manually-linked device)."""
5896 permission_required = "netbox_data_import.delete_deviceexistingmatch"
5897 permission_denied_response_format = "redirect"
5899 def post(self, request):
5900 """Delete the DeviceExistingMatch and go back to the replanned preview."""
5901 command = _UnlinkDevice(source_id=source_text(request.POST.get("source_id")))
5902 result = apply_preview_command(request, PreviewClaim.posted(request.POST), command)
5903 return _command_response(request, result, self.refusal_url(request))