Coverage for netbox_data_import/models.py: 99%
766 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3import hashlib
4from contextlib import contextmanager, suppress
5from datetime import timedelta
7from django.conf import settings
8from django.core.exceptions import ValidationError
9from django.core.validators import MinValueValidator
10from django.db import IntegrityError, models, transaction
11from django.urls import reverse
12from django.utils import timezone
13from core.choices import JobStatusChoices
14from core.models import Job
15from netbox.models import NetBoxModel
16from netbox.models.features import JobsMixin
17from utilities.querysets import RestrictedQuerySet
19from .adapters import (
20 DEFAULT_ADAPTER_KEY,
21 UnknownSourceAdapter,
22 adapter_choices,
23 get_adapter,
24 output_kinds_for,
25)
26from . import plan
27from .catalog import CATALOG, POLICY_SECTIONS, has_implemented_module, policy_section
28from .field_keys import SELECT_TERMINATION_TASK, parse_termination_field_key
29from . import inference_settings as _inference_settings
30from .trace_schema import TRACE_EXPORT_TIMESTAMP_MAX_LENGTH
32CONTACT_RESOLUTION_FIELDS = frozenset({"name", "email", "phone"})
33CONTACT_RESOLUTION_REQUIRED_KEYS = frozenset({"contact_resolution_applied", "contact_field_sources"})
34CONTACT_RESOLUTION_KEYS = CONTACT_RESOLUTION_REQUIRED_KEYS | frozenset({"contact_field_values", "contact_id"})
35SOURCE_RESOLUTION_RESERVED_FIELDS = frozenset({"source_id", "_conflicts"})
38def _invalid_contact_resolution_key_message(key):
39 """Describe one unknown Contact resolution key using the Contact-policy vocabulary."""
40 allowed = ", ".join(f"'{field}'" for field in sorted(CONTACT_RESOLUTION_KEYS))
41 return f"'{key}' is not a Contact resolution field. Use one of {allowed}."
44def validate_adapter_target_module(adapter_key):
45 """Reject a Source Adapter whose Target Module this release does not implement yet."""
46 if not has_implemented_module(output_kinds_for(adapter_key)):
47 raise ValidationError(
48 {"source_adapter": f"This release cannot import from the '{adapter_key}' source adapter yet."}
49 )
52def validate_registered_adapter(profile):
53 """Reject a profile whose stored Source Adapter this release does not register."""
54 if profile is not None and profile.adapter is None:
55 raise ValidationError(
56 f"This profile uses the source adapter '{profile.source_adapter}', which this release does not register."
57 )
60def validate_section_applicability(profile, section_key):
61 """Reject a policy row whose section does not apply to its profile's Source Adapter."""
62 section = policy_section(section_key)
63 if section is None or profile is None:
64 return
65 if not section.applies_to(profile.output_kinds):
66 raise ValidationError(
67 f"{section.label} do not apply to a profile using the '{profile.source_adapter}' source adapter."
68 )
71def _validated_contact_id(contact_id):
72 """Return a saved Contact ID as a positive int, rejecting anything int() would reshape."""
73 if contact_id in ("", None):
74 return None
75 # int() truncates, so a JSON float would silently select a different Contact.
76 if isinstance(contact_id, bool) or (isinstance(contact_id, float) and not contact_id.is_integer()):
77 raise ValidationError("The selected Contact ID is invalid.")
78 try:
79 contact_id = int(contact_id)
80 except (TypeError, ValueError) as exc:
81 raise ValidationError("The selected Contact ID is invalid.") from exc
82 if contact_id < 1:
83 raise ValidationError("The selected Contact ID is invalid.")
84 return contact_id
87def validate_contact_candidate_resolution(
88 resolved_fields,
89 lookup_field: str,
90 available_source_columns,
91) -> dict:
92 """Validate and normalize one saved Contact candidate resolution."""
93 if not isinstance(resolved_fields, dict) or not set(resolved_fields) >= CONTACT_RESOLUTION_REQUIRED_KEYS:
94 raise ValidationError("The Contact candidate resolution has an invalid structure.")
95 invalid = sorted(set(resolved_fields) - CONTACT_RESOLUTION_KEYS)
96 if invalid:
97 raise ValidationError(_invalid_contact_resolution_key_message(invalid[0]))
98 if resolved_fields.get("contact_resolution_applied") is not True:
99 raise ValidationError("The Contact candidate resolution is not marked as applied.")
101 field_sources = resolved_fields.get("contact_field_sources")
102 if not isinstance(field_sources, dict) or set(field_sources) - CONTACT_RESOLUTION_FIELDS:
103 raise ValidationError("The Contact candidate resolution contains an unknown field.")
104 if any(not isinstance(source_column, str) or not source_column for source_column in field_sources.values()):
105 raise ValidationError("Each resolved Contact field must select one source column.")
106 missing_sources = set(field_sources.values()) - set(available_source_columns)
107 if missing_sources:
108 missing = min(missing_sources)
109 raise ValidationError(f"The source column '{missing}' has no candidate value in this row.")
111 field_values = resolved_fields.get("contact_field_values", {})
112 if not isinstance(field_values, dict) or set(field_values) - CONTACT_RESOLUTION_FIELDS:
113 raise ValidationError("The Contact candidate resolution contains an unknown literal field.")
114 if any(not isinstance(value, str) or not value.strip() for value in field_values.values()):
115 raise ValidationError("Each literal Contact field must contain text.")
116 overlap = set(field_sources) & set(field_values)
117 if overlap:
118 raise ValidationError(f"Select a source column or enter a value for Contact {min(overlap)}, not both.")
120 contact_id = _validated_contact_id(resolved_fields.get("contact_id"))
122 supplied_fields = set(field_sources) | set(field_values)
123 if supplied_fields and "name" not in supplied_fields:
124 raise ValidationError("Select a source column or enter a value for the Contact name.")
125 if supplied_fields and lookup_field not in supplied_fields:
126 raise ValidationError(f"Select a source column or enter a value for the Contact {lookup_field} lookup field.")
127 return {
128 "field_sources": field_sources,
129 "field_values": {field: value.strip() for field, value in field_values.items()},
130 "contact_id": contact_id,
131 }
134def validate_source_resolution_fields(profile, source_column, resolved_fields):
135 """Reject resolved fields that cannot safely merge into one source row."""
136 if not isinstance(resolved_fields, dict):
137 raise ValidationError({"resolved_fields": "Enter the resolved fields as a JSON object."})
138 if any(not isinstance(key, str) for key in resolved_fields):
139 raise ValidationError({"resolved_fields": "Each resolved field name must be text."})
141 reserved = sorted(set(resolved_fields) & SOURCE_RESOLUTION_RESERVED_FIELDS)
142 if reserved:
143 raise ValidationError(
144 {"resolved_fields": f"The resolved field '{reserved[0]}' is reserved for import planning."}
145 )
147 if source_column == "candidate:contact":
148 invalid = sorted(set(resolved_fields) - CONTACT_RESOLUTION_KEYS)
149 if invalid:
150 raise ValidationError({"resolved_fields": _invalid_contact_resolution_key_message(invalid[0])})
151 return
152 output_kinds = profile.output_kinds if profile is not None else None
153 invalid = sorted(
154 key for key in resolved_fields if not CATALOG.is_valid(key, output_kinds=output_kinds, allow_candidates=False)
155 )
156 if invalid:
157 raise ValidationError({"resolved_fields": CATALOG.invalid_key_message(invalid[0])})
160@contextmanager
161def locked_profile_policy(*profile_ids):
162 """Hold the given profile rows for a policy write or for an import execution.
164 Every SourceResolution write and the import worker take this same lock, so a decision cannot
165 commit between the worker's check and its writes. Locking the resolution rows alone would leave
166 an insert free to land in that window, because a row that does not exist yet cannot be locked.
168 The rows lock in primary-key order, so two callers naming several profiles cannot deadlock by
169 taking them in opposite orders.
170 """
171 wanted = sorted({profile_id for profile_id in profile_ids if profile_id is not None})
172 # Django short-circuits `pk__in=[]`, so an empty set would yield without ever taking a lock.
173 if not wanted:
174 raise ImportProfile.DoesNotExist("A policy write must name at least one ImportProfile to lock.")
175 with transaction.atomic():
176 locked = ImportProfile.objects.select_for_update().filter(pk__in=wanted).order_by("pk")
177 if len(locked) != len(wanted):
178 raise ImportProfile.DoesNotExist(f"No ImportProfile matches every id in {wanted}.")
179 yield
182@contextmanager
183def locked_resolution_policy(resolution_pk):
184 """Hold the profile a saved resolution belongs to, read from the database rather than trusted.
186 A caller reaches this holding an instance it fetched earlier, whose profile may be a stale copy.
187 The row is read again under the lock, so the caller acts on a row that still exists and still
188 belongs to the locked profile.
189 """
190 gone = SourceResolution.DoesNotExist(f"No SourceResolution matches id {resolution_pk}.")
191 profile_id = SourceResolution.objects.filter(pk=resolution_pk).values_list("profile_id", flat=True).first()
192 if profile_id is None:
193 raise gone
194 with locked_profile_policy(profile_id):
195 # A delete can still commit in the gap above, and a write that saw the row would resurrect it.
196 if not SourceResolution.objects.filter(pk=resolution_pk, profile_id=profile_id).exists():
197 raise gone
198 yield
201class ImportProfile(NetBoxModel):
202 """Named configuration for one source file format."""
204 name = models.CharField(max_length=100, unique=True)
205 description = models.TextField(blank=True)
206 source_adapter = models.CharField(
207 max_length=50,
208 choices=adapter_choices,
209 default=DEFAULT_ADAPTER_KEY,
210 help_text="Source format this profile reads. It cannot change after creation.",
211 )
212 adapter_config = models.JSONField(
213 default=dict,
214 blank=True,
215 help_text="Scalar settings the selected Source Adapter declares.",
216 )
218 # Override tags reverse accessor to avoid clashes with other plugins
219 tags = models.ManyToManyField(
220 to="extras.Tag",
221 related_name="+",
222 blank=True,
223 )
225 class Meta:
226 ordering = ["name"]
227 verbose_name = "Import Profile"
228 verbose_name_plural = "Import Profiles"
230 def __str__(self):
231 return self.name
233 def get_absolute_url(self):
234 """Return the detail URL for this import profile."""
235 return reverse("plugins:netbox_data_import:importprofile", args=[self.pk])
237 def _validate_source_adapter_immutability(self):
238 """Return the persisted adapter and reject a different selected adapter."""
239 # A set pk does not prove that the row exists. An unsaved instance can carry a pk.
240 stored = (
241 type(self).objects.filter(pk=self.pk).values_list("source_adapter", flat=True).first()
242 if self.pk is not None
243 else None
244 )
245 if stored is not None and stored != self.source_adapter:
246 raise ValidationError({"source_adapter": "The source adapter cannot change after the profile is created."})
247 return stored
249 def save(self, *args, **kwargs):
250 """Normalize adapter configuration on every supported write that stores it."""
251 update_fields = kwargs.get("update_fields")
252 updated = set(update_fields) if update_fields is not None else None
253 if updated is None or updated & {"source_adapter", "adapter_config"}:
254 self._validate_source_adapter_immutability()
255 if updated is None or "adapter_config" in updated:
256 adapter = self.adapter
257 if adapter is None:
258 raise ValidationError({"source_adapter": f"Unknown source adapter '{self.source_adapter}'."})
259 self.adapter_config = adapter.config_form_class().validate_config(self.adapter_config)
260 return super().save(*args, **kwargs)
262 def delete(self, *args, **kwargs):
263 """Take the policy lock before the cascade, which would otherwise take the child rows first.
265 A policy write holds this row and then writes a child, so a cascade in the opposite order
266 deadlocks against it. NetBox deletes each object through this method, in bulk as well.
267 """
268 with locked_profile_policy(self.pk):
269 # atomic-exit-safe: locked-cascade-committed
270 return super().delete(*args, **kwargs)
272 @property
273 def adapter(self):
274 """Return the registered Source Adapter class for this profile."""
275 return get_adapter(self.source_adapter)
277 @property
278 def output_kinds(self) -> frozenset[str]:
279 """Return the adapter output kinds this profile can supply."""
280 return output_kinds_for(self.source_adapter)
282 @property
283 def adapter_settings(self):
284 """Return attribute access over ``adapter_config`` backed by the adapter's declared defaults."""
285 cache_key = (self.source_adapter, id(self.adapter_config))
286 cached = self.__dict__.get("_adapter_settings_cache")
287 if cached is not None and cached[0] == cache_key:
288 return cached[1]
289 settings = AdapterSettings(self.adapter, self.adapter_config, self.source_adapter)
290 self.__dict__["_adapter_settings_cache"] = (cache_key, settings)
291 return settings
293 @property
294 def adapter_config_display(self):
295 """Return (label, value) pairs for the adapter's declared settings, in declaration order."""
296 from django import forms
297 from django.forms.utils import pretty_name
299 config = _require_adapter_config_mapping(self.adapter_config)
300 adapter = self.adapter
301 if adapter is None:
302 return []
303 rows = []
304 for name, field in adapter.config_form_class().base_fields.items():
305 value = config.get(name, field.initial)
306 if isinstance(field, forms.ChoiceField) and not isinstance(field, forms.ModelChoiceField):
307 value = dict(field.choices).get(value, value)
308 rows.append((field.label or pretty_name(name), value))
309 return rows
311 def grouped_column_map(self) -> dict[str, list[str]]:
312 """Return this profile's mapped source columns, keyed by Target Field."""
313 grouped: dict[str, list[str]] = {}
314 # Two columns can feed one Target Field, and which one wins must not be a query-order accident.
315 for mapping in self.column_mappings.order_by("target_field", "pk"):
316 grouped.setdefault(mapping.target_field, []).append(mapping.source_column)
317 return grouped
319 @property
320 def planning_fingerprint(self) -> str:
321 """Return the fingerprint of every profile value planning depends on."""
322 related_sections = {
323 getattr(relation.related_model, "POLICY_SECTION", ""): relation.get_accessor_name()
324 for relation in self._meta.related_objects
325 }
326 sections = []
327 for section in POLICY_SECTIONS:
328 accessor = related_sections[section.key]
329 serialized_rows = []
330 for row in getattr(self, accessor).all():
331 serialized_rows.append(
332 {
333 field.name: field.value_from_object(row)
334 for field in row._meta.concrete_fields
335 if field.name not in {"id", "profile"}
336 }
337 )
338 serialized_rows.sort(key=plan.canonical_json)
339 sections.append({"key": section.key, "rows": serialized_rows})
340 return plan.fingerprint_of(
341 {
342 "profile_id": self.pk,
343 "source_adapter": self.source_adapter,
344 "adapter_config": self.adapter_config,
345 "policy_sections": sections,
346 }
347 )
349 @property
350 def resolved_primary_contact_role(self):
351 """Return the referenced Contact Role object, or None when unset or dangling.
353 Planning reads this once per row, so the lookup is memoized against the configured name. A
354 plain instance cache would keep returning the old role after ``adapter_config`` changes.
355 """
356 name = self.adapter_settings.primary_contact_role
357 if not name:
358 return None
359 cached = self.__dict__.get("_primary_contact_role_cache")
360 if cached is not None and cached[0] == name:
361 return cached[1]
362 from tenancy.models import ContactRole
364 role = ContactRole.objects.filter(name=name).first()
365 self.__dict__["_primary_contact_role_cache"] = (name, role)
366 return role
368 def clean(self):
369 """Reject an adapter change after creation and validate the adapter configuration."""
370 super().clean()
371 adapter = self.adapter
372 if adapter is None:
373 raise ValidationError({"source_adapter": f"Unknown source adapter '{self.source_adapter}'."})
374 stored = self._validate_source_adapter_immutability()
375 if stored is None:
376 # A creation rule only: the adapter is immutable, so a stored profile keeps validating.
377 validate_adapter_target_module(self.source_adapter)
378 self.adapter_config = adapter.config_form_class().validate_config(self.adapter_config)
381def _require_adapter_config_mapping(config):
382 """Return a stored adapter configuration mapping or expose corrupt JSON state."""
383 if not isinstance(config, dict):
384 raise ValidationError(
385 "The stored adapter configuration must be a mapping. Replace it with a valid JSON mapping."
386 )
387 return config
390class AdapterSettings:
391 """Read one adapter setting, falling back to the adapter form's declared default."""
393 def __init__(self, adapter, config, adapter_key):
394 self._adapter_key = adapter_key
395 self._fields = adapter.config_form_class().base_fields if adapter is not None else None
396 self._config = _require_adapter_config_mapping(config)
398 def get(self, name, default):
399 """Return one setting, or *default* when this profile's adapter declares none.
401 Only a caller that serves every adapter may ask this way. Adapter-specific code reads the
402 attribute, so a setting it depends on cannot go missing quietly.
403 """
404 fields = object.__getattribute__(self, "_fields")
405 return getattr(self, name) if fields is not None and name in fields else default
407 def __getattr__(self, name):
408 fields = object.__getattribute__(self, "_fields")
409 if fields is None:
410 key = object.__getattribute__(self, "_adapter_key")
411 raise UnknownSourceAdapter(f"This release does not register the source adapter '{key}'.")
412 if name not in fields:
413 raise AttributeError(f"'{name}' is not a setting of this profile's source adapter")
414 config = object.__getattribute__(self, "_config")
415 field = fields[name]
416 if name not in config:
417 return field.initial
418 value = config[name]
419 if field.required and (value is None or value == ""):
420 raise ValidationError(f"The required adapter setting '{name}' is empty. Edit and save this import profile.")
421 return value
424class PolicySectionModel(models.Model):
425 """A profile policy table scoped to the adapter output kinds its catalog section declares."""
427 POLICY_SECTION = ""
429 # NetBox's generic views scope a queryset with `restrict()`, which only this manager provides.
430 objects = RestrictedQuerySet.as_manager()
432 class Meta:
433 abstract = True
435 def clean(self):
436 """Reject a row whose section does not apply to the profile's Source Adapter."""
437 super().clean()
438 validate_section_applicability(self.profile if self.profile_id else None, self.POLICY_SECTION)
441class ColumnMapping(PolicySectionModel):
442 """Maps one source column header to one semantic NetBox field."""
444 POLICY_SECTION = "column_mappings"
446 profile = models.ForeignKey(
447 ImportProfile,
448 on_delete=models.CASCADE,
449 related_name="column_mappings",
450 )
451 source_column = models.CharField(
452 max_length=200,
453 help_text="Exact column header in the source file (case-sensitive)",
454 )
455 target_field = models.CharField(max_length=100)
457 class Meta:
458 ordering = ["profile", "target_field"]
459 verbose_name = "Column Mapping"
460 verbose_name_plural = "Column Mappings"
462 def clean(self):
463 """Resolve the target field through the catalog and reject an inapplicable row."""
464 super().clean()
465 value = self.target_field or ""
466 if not CATALOG.is_valid(value, output_kinds=self.profile.output_kinds if self.profile_id else None):
467 raise ValidationError({"target_field": CATALOG.invalid_key_message(value)})
468 if not self.profile_id or not value:
469 return
470 conflict = (
471 ColumnTransformRule.objects.filter(profile_id=self.profile_id)
472 .filter(models.Q(group_1_target=value) | models.Q(group_2_target=value))
473 .only("source_column")
474 .first()
475 )
476 if conflict is not None:
477 raise ValidationError(
478 {
479 "target_field": (
480 f"Target field '{value}' is already assigned by the transform rule "
481 f"for source column '{conflict.source_column}'."
482 )
483 }
484 )
486 def get_target_field_display(self):
487 """Return the human-readable name for the target_field value."""
488 return CATALOG.display(self.target_field)
490 def __str__(self):
491 return f"{self.source_column} → {self.get_target_field_display()}"
493 def get_absolute_url(self):
494 """Return the edit URL for this column mapping."""
495 return reverse("plugins:netbox_data_import:columnmapping_edit", args=[self.pk])
498class ClassRoleMapping(PolicySectionModel):
499 """Maps a source 'class' value to a NetBox outcome (rack or device role)."""
501 POLICY_SECTION = "class_role_mappings"
503 profile = models.ForeignKey(
504 ImportProfile,
505 on_delete=models.CASCADE,
506 related_name="class_role_mappings",
507 )
508 source_class = models.CharField(
509 max_length=200,
510 help_text="Value from the class column (e.g. 'Server', 'Cabinet')",
511 )
512 creates_rack = models.BooleanField(
513 default=False,
514 help_text="If checked, rows with this class create a Rack instead of a Device",
515 )
516 rack_type = models.ForeignKey(
517 to="dcim.RackType",
518 on_delete=models.SET_NULL,
519 null=True,
520 blank=True,
521 related_name="+",
522 help_text="Optional rack type assigned when creating racks",
523 )
524 role_slug = models.CharField(
525 max_length=100,
526 blank=True,
527 help_text="NetBox device role slug (ignored when 'creates rack' is checked)",
528 )
529 ignore = models.BooleanField(
530 default=False,
531 help_text="If checked, rows with this class are silently skipped (not shown as errors)",
532 )
534 class Meta:
535 ordering = ["profile", "source_class"]
536 constraints = [
537 models.UniqueConstraint(fields=["profile", "source_class"], name="ndi_classrolemapping_profile_class"),
538 ]
539 verbose_name = "Class → Role Mapping"
540 verbose_name_plural = "Class → Role Mappings"
542 def __str__(self):
543 if self.creates_rack:
544 suffix = f" ({self.rack_type})" if self.rack_type_id else ""
545 return f"{self.source_class} → Rack{suffix}"
546 return f"{self.source_class} → {self.role_slug}"
548 def get_absolute_url(self):
549 """Return the edit URL for this class→role mapping."""
550 return reverse("plugins:netbox_data_import:classrolemapping_edit", args=[self.pk])
553def _flatten_choice_groups(choices):
554 """Return value and label pairs from flat or grouped NetBox choices."""
555 flattened = []
556 for value, label in choices:
557 if isinstance(label, (tuple, list)):
558 flattened.extend(label)
559 else:
560 flattened.append((value, label))
561 return tuple(flattened)
564def cable_type_choices():
565 """Return the Cable Type values offered by the running NetBox instance."""
566 from dcim.choices import CableTypeChoices
568 return _flatten_choice_groups(CableTypeChoices.CHOICES)
571def cable_profile_choices():
572 """Return the Cable Profile values offered by the running NetBox instance."""
573 from dcim.choices import CableProfileChoices
575 return _flatten_choice_groups(CableProfileChoices.CHOICES)
578def cable_profile_accepts_one_termination_per_side(value) -> bool:
579 """Return whether NetBox reports one connector on each side of the Cable Profile."""
580 from dcim.models import Cable
582 profile_class = Cable(profile=value).profile_class
583 return profile_class is not None and len(profile_class.a_connectors) == 1 and len(profile_class.b_connectors) == 1
586def compatible_cable_profile_choices():
587 """Return running Cable Profiles that permit one termination on each side."""
588 return tuple(
589 (value, label)
590 for value, label in cable_profile_choices()
591 if cable_profile_accepts_one_termination_per_side(value)
592 )
595def cable_class_mapping_choice_errors(cable_type, cable_profile):
596 """Return runtime-choice and profile-cardinality errors by model field."""
597 errors = {}
598 type_values = {value for value, _label in cable_type_choices()}
599 profile_values = {value for value, _label in cable_profile_choices()}
600 if cable_type is not None and cable_type not in type_values:
601 errors["cable_type"] = ValidationError(
602 "The selected Cable Type is no longer offered by this NetBox instance.",
603 code="cable.cableclass_stale_mapping",
604 )
605 if cable_profile is not None and cable_profile not in profile_values:
606 errors["cable_profile"] = ValidationError(
607 "The selected Cable Profile is no longer offered by this NetBox instance.",
608 code="cable.cableclass_stale_mapping",
609 )
610 elif cable_profile is not None and not cable_profile_accepts_one_termination_per_side(cable_profile):
611 errors["cable_profile"] = ValidationError(
612 "The selected Cable Profile does not permit one termination on each side.",
613 code="cable.profile_incompatible",
614 )
615 return errors
618class CableClassMapping(PolicySectionModel):
619 """Map one source CableClass to independent Cable Type and Cable Profile decisions."""
621 POLICY_SECTION = "cable_class_mappings"
623 profile = models.ForeignKey(
624 ImportProfile,
625 on_delete=models.CASCADE,
626 related_name="cable_class_mappings",
627 )
628 cable_class = models.CharField(max_length=200)
629 cable_type_resolved = models.BooleanField(default=False)
630 cable_type = models.CharField(max_length=50, null=True, blank=True)
631 cable_profile_resolved = models.BooleanField(default=False)
632 cable_profile = models.CharField(max_length=50, null=True, blank=True)
634 class Meta:
635 ordering = ["profile", "cable_class"]
636 constraints = [
637 models.UniqueConstraint(
638 fields=["profile", "cable_class"],
639 name="ndi_cableclassmapping_profile_class",
640 ),
641 models.CheckConstraint(
642 condition=models.Q(cable_type__isnull=True)
643 | (models.Q(cable_type_resolved=True) & ~models.Q(cable_type="")),
644 name="ndi_cableclassmapping_type_resolved_value",
645 ),
646 models.CheckConstraint(
647 condition=models.Q(cable_profile__isnull=True)
648 | (models.Q(cable_profile_resolved=True) & ~models.Q(cable_profile="")),
649 name="ndi_cableclassmapping_profile_resolved_value",
650 ),
651 ]
652 verbose_name = "CableClass Mapping"
653 verbose_name_plural = "CableClass Mappings"
655 def clean(self):
656 """Reject inapplicable, inconsistent, stale, or incompatible mapping values."""
657 super().clean()
658 self.cable_type = self.cable_type or None
659 self.cable_profile = self.cable_profile or None
660 errors = cable_class_mapping_choice_errors(self.cable_type, self.cable_profile)
661 if not self.cable_type_resolved and self.cable_type is not None:
662 errors["cable_type"] = ValidationError(
663 "An unresolved Cable Type cannot store a selected value.",
664 code="invalid",
665 )
666 if not self.cable_profile_resolved and self.cable_profile is not None:
667 errors["cable_profile"] = ValidationError(
668 "An unresolved Cable Profile cannot store a selected value.",
669 code="invalid",
670 )
671 if errors:
672 raise ValidationError(errors)
674 def cable_type_display(self):
675 """Return the operator-facing Cable Type decision."""
676 if not self.cable_type_resolved:
677 return "Unresolved"
678 if self.cable_type is None:
679 return "None"
680 return str(dict(cable_type_choices()).get(self.cable_type, self.cable_type))
682 def cable_profile_display(self):
683 """Return the operator-facing Cable Profile decision."""
684 if not self.cable_profile_resolved:
685 return "Unresolved"
686 if self.cable_profile is None:
687 return "None"
688 return str(dict(cable_profile_choices()).get(self.cable_profile, self.cable_profile))
690 def __str__(self):
691 return self.cable_class
693 def get_absolute_url(self):
694 """Return the edit URL for this CableClass mapping."""
695 return reverse("plugins:netbox_data_import:cableclassmapping_edit", args=[self.pk])
698def index_digest(value: str) -> str:
699 """Return the fixed-width index key one piece of unbounded source text is stored under.
701 PostgreSQL refuses a btree entry past about 2704 bytes, so a constraint over source text
702 carries this digest instead of the text.
703 """
704 return hashlib.sha256(value.encode("utf-8")).hexdigest()
707def _canonical_termination_field_key(value):
708 """Return *value* when it is an exact canonical termination field key."""
709 try:
710 parse_termination_field_key(value)
711 except (TypeError, ValueError) as exc:
712 raise ValidationError(
713 "Enter the canonical JSON termination field key.",
714 code="invalid",
715 ) from exc
716 return value
719class TerminationResolution(PolicySectionModel):
720 """Store one selected NetBox termination for a trace field-key role."""
722 POLICY_SECTION = "termination_resolutions"
724 profile = models.ForeignKey(
725 ImportProfile,
726 on_delete=models.CASCADE,
727 related_name="termination_resolutions",
728 )
729 task_type = models.CharField(
730 max_length=50,
731 choices=((SELECT_TERMINATION_TASK, "Select termination"),),
732 )
733 field_key = models.TextField()
734 field_key_digest = models.CharField(
735 max_length=64,
736 blank=True,
737 editable=False,
738 help_text="Fixed-width digest of field_key, which is what the index and constraint carry",
739 )
740 selected_object_type = models.ForeignKey(
741 to="core.ObjectType",
742 on_delete=models.PROTECT,
743 related_name="+",
744 )
745 selected_object_id = models.PositiveBigIntegerField()
746 selected_display_name = models.CharField(max_length=200)
748 class Meta:
749 ordering = ["profile", "task_type", "field_key"]
750 constraints = [
751 models.UniqueConstraint(
752 fields=["profile", "task_type", "field_key_digest"],
753 name="ndi_termresolution_profile_task_key",
754 ),
755 ]
756 verbose_name = "Termination Resolution"
757 verbose_name_plural = "Termination Resolutions"
759 def clean(self):
760 """Reject an inapplicable row or a noncanonical field key."""
761 super().clean()
762 try:
763 _canonical_termination_field_key(self.field_key)
764 except ValidationError as exc:
765 raise ValidationError({"field_key": exc}) from exc
766 # Before validate_unique, which reads the constraint's own fields.
767 self.field_key_digest = index_digest(self.field_key)
769 def save(self, *args, **kwargs):
770 """Derive the index key, so no caller can store one that disagrees with the field key."""
771 self.field_key_digest = index_digest(self.field_key)
772 update_fields = kwargs.get("update_fields")
773 # A partial save of the key alone would leave the constraint on the digest it replaced.
774 if update_fields is not None and "field_key" in update_fields:
775 kwargs["update_fields"] = {*update_fields, "field_key_digest"}
776 super().save(*args, **kwargs)
778 def __str__(self):
779 return f"{self.task_type}: {self.selected_display_name}"
782class SourceDocument(models.Model):
783 """The stored uploaded workbook that a plan and its executions read.
785 Preview, replanning, a background execution, and an audit read all resolve the same bytes, so the
786 plan carries a reference instead of the content.
787 """
789 RETENTION = timedelta(days=30)
791 # Audit input outlives its profile, so a delete orphans the row and retention reclaims it.
792 profile = models.ForeignKey(
793 ImportProfile, on_delete=models.SET_NULL, null=True, blank=True, related_name="source_documents"
794 )
795 content = models.BinaryField()
796 content_fingerprint = models.CharField(max_length=64)
797 filename = models.CharField(max_length=255, blank=True)
798 uploaded_by = models.ForeignKey(
799 settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
800 )
801 created = models.DateTimeField(auto_now_add=True)
803 class Meta:
804 ordering = ["-created"]
805 indexes = [models.Index(fields=["profile", "content_fingerprint"])]
806 verbose_name = "Source Document"
807 verbose_name_plural = "Source Documents"
809 def __str__(self):
810 return f"{self.filename or 'upload'} ({self.content_fingerprint[:12]})"
812 @staticmethod
813 def fingerprint(content: bytes) -> str:
814 """Return the content fingerprint, which is what a plan compares against."""
815 return hashlib.sha256(bytes(content)).hexdigest()
817 @classmethod
818 def store(cls, *, profile, content, filename="", uploaded_by=None):
819 """Store one upload. A newer upload never removes an older one."""
820 return cls.objects.create(
821 profile=profile,
822 content=bytes(content),
823 content_fingerprint=cls.fingerprint(content),
824 filename=filename,
825 uploaded_by=uploaded_by,
826 )
828 @classmethod
829 def purge_unreferenced(cls, *, now=None) -> int:
830 """Delete unreferenced uploads past the retention window and return the count.
832 A document an Import Execution references is permanent audit input, so the queryset excludes
833 it and the protecting foreign key backs that up.
834 """
835 reference_time = now or timezone.now()
836 if timezone.is_naive(reference_time):
837 raise ValueError("now must be timezone-aware")
838 cutoff = reference_time - cls.RETENTION
839 # The protecting relation forces a row-by-row collect, so defer the bytes one purge would load.
840 stale = cls.objects.filter(import_executions__isnull=True, created__lt=cutoff).defer("content")
841 return stale.delete()[0]
844class ExecutionOutcome:
845 """The outcome vocabulary of an Import Execution (section 9.2)."""
847 PENDING = "pending"
848 SUCCEEDED = "succeeded"
849 FAILED = "failed"
851 CHOICES = ((PENDING, "Pending"), (SUCCEEDED, "Succeeded"), (FAILED, "Failed"))
854class FailureReason:
855 """Typed failure reasons an Import Execution records."""
857 ABANDONED = "abandoned"
858 DATABASE = "database"
859 PERMISSION = "permission"
860 PRECONDITION = "precondition"
861 PLANNING = "planning"
862 SELECTION = "selection"
863 STALE_PLAN = "stale_plan"
864 VALIDATION = "validation"
867class ImportExecution(models.Model):
868 """The audit record of one selective or final execution.
870 Rows created before the plan cutover keep their historical columns, have null new fields, and are
871 display-only: they never satisfy an idempotency lookup and never take part in plan comparison.
872 """
874 #: A synchronous attempt cannot outlive the web request bound, so an older pending row is gone.
875 SYNCHRONOUS_BOUND = timedelta(minutes=10)
877 profile = models.ForeignKey(
878 ImportProfile,
879 on_delete=models.SET_NULL,
880 null=True,
881 blank=True,
882 related_name="import_executions",
883 )
884 created = models.DateTimeField(auto_now_add=True)
885 input_filename = models.CharField(max_length=255, blank=True)
886 site_name = models.CharField(max_length=100, blank=True)
887 result_counts = models.JSONField(default=dict)
889 source_document = models.ForeignKey(
890 SourceDocument,
891 on_delete=models.PROTECT,
892 null=True,
893 blank=True,
894 related_name="import_executions",
895 )
896 actor = models.ForeignKey(
897 settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
898 )
899 idempotency_key = models.CharField(max_length=64, null=True, blank=True)
900 plan_schema_version = models.PositiveIntegerField(null=True, blank=True)
901 accepted_plan_fingerprint = models.CharField(max_length=64, null=True, blank=True)
902 selected_units = models.JSONField(null=True, blank=True)
903 outcome = models.CharField(max_length=16, choices=ExecutionOutcome.CHOICES, null=True, blank=True)
904 applied_changes = models.JSONField(null=True, blank=True)
905 failure_detail = models.JSONField(null=True, blank=True)
906 # Set by link_job: the reservation commits before the Job exists, and outlives a deleted Job.
907 job_backed = models.BooleanField(default=False)
908 job = models.OneToOneField(
909 "core.Job", on_delete=models.SET_NULL, null=True, blank=True, related_name="import_execution"
910 )
912 class Meta:
913 ordering = ["-created"]
914 constraints = [
915 models.UniqueConstraint(
916 fields=["profile", "idempotency_key"],
917 condition=models.Q(idempotency_key__isnull=False),
918 name="ndi_execution_profile_idempotency_key",
919 ),
920 ]
921 verbose_name = "Import Execution"
922 verbose_name_plural = "Import Executions"
924 def __str__(self):
925 return f"Import {self.pk} — {self.created:%Y-%m-%d %H:%M} ({self.input_filename})"
927 def get_absolute_url(self):
928 """Return the associated profile's URL (no per-execution detail view exists)."""
929 if not self.profile_id:
930 return reverse("plugins:netbox_data_import:importprofile_list")
931 return reverse("plugins:netbox_data_import:importprofile", args=[self.profile_id])
933 @classmethod
934 def reserve(cls, **fields):
935 """Insert and commit the pending row, or return the row already holding this key.
937 The insert reserves the unique (Import Profile, idempotency key), so a duplicate submission
938 or job delivery loses the race and returns the existing row in any outcome.
939 """
940 if transaction.get_connection().in_atomic_block:
941 raise RuntimeError("The Import Execution reservation must commit before the target transaction opens.")
942 if not fields.get("idempotency_key"):
943 raise ValueError("An Import Execution reservation requires an idempotency key.")
944 # PostgreSQL treats two NULL profiles as distinct, so the partial unique index cannot hold.
945 if not fields.get("profile"):
946 raise ValueError("An Import Execution reservation requires an Import Profile.")
947 required_audit_fields = {
948 "source_document": "a Source Document",
949 "actor": "an actor",
950 "plan_schema_version": "a plan schema version",
951 "accepted_plan_fingerprint": "an accepted plan fingerprint",
952 "selected_units": "selected Synchronization Unit identities",
953 }
954 for field_name, label in required_audit_fields.items():
955 if fields.get(field_name) is None:
956 raise ValueError(f"An Import Execution reservation requires {label}.")
957 existing = cls.for_idempotency(fields["profile"], fields["idempotency_key"])
958 if existing is not None:
959 return existing, False
960 try:
961 return cls.objects.create(outcome=ExecutionOutcome.PENDING, **fields), True
962 except IntegrityError:
963 # Only a lost race for this key is recoverable; any other constraint failure must surface.
964 winner = cls.for_idempotency(fields["profile"], fields["idempotency_key"])
965 if winner is None:
966 raise
967 return winner, False
969 def link_job(self, job):
970 """Record the native Job that runs this execution, after the reservation has committed."""
971 self.job = job
972 self.job_backed = True
973 self.save(update_fields=["job", "job_backed"])
974 return self
976 @classmethod
977 def for_idempotency(cls, profile, idempotency_key):
978 """Return the reserved row for this key, reconciled, or None. A legacy row never matches."""
979 if not idempotency_key:
980 return None
981 found = cls.objects.filter(profile=profile, idempotency_key=idempotency_key).first()
982 return found.reconcile_pending() if found is not None else None
984 def reconcile_pending(self, *, now=None):
985 """Transition an abandoned pending row to failed, so no sweeper is needed."""
986 if self.outcome != ExecutionOutcome.PENDING:
987 return self
988 if self.job_backed:
989 # The Job decides once it exists; a deleted Job leaves no way to finish the attempt.
990 job = Job.objects.filter(pk=self.job_id).first() if self.job_id else None
991 live = job is not None and job.status in JobStatusChoices.ENQUEUED_STATE_CHOICES
992 else:
993 # Either a synchronous attempt or a background one still between reserving and enqueuing.
994 live = self.created > (now or timezone.now()) - self.SYNCHRONOUS_BOUND
995 if live:
996 return self
997 # A worker finished the row between this read and the transition; its outcome wins.
998 with suppress(ValueError):
999 self.mark_failed(reason=FailureReason.ABANDONED)
1000 return self
1002 def _finish(self, **values):
1003 """Transition this row out of pending exactly once, with the database as the arbiter.
1005 Two instances can both hold a pending copy, so an in-memory check would let the second
1006 write overwrite a committed outcome and destroy the audit evidence.
1007 """
1008 updated = type(self).objects.filter(pk=self.pk, outcome=ExecutionOutcome.PENDING).update(**values)
1009 if not updated:
1010 self.refresh_from_db()
1011 raise ValueError(f"Import Execution {self.pk} already finished as '{self.outcome}'.")
1012 for name, value in values.items():
1013 setattr(self, name, value)
1014 return self
1016 def mark_succeeded(self, *, applied_changes, result_counts=None):
1017 """Record the applied identities and the deleted-object snapshot."""
1018 return self._finish(
1019 outcome=ExecutionOutcome.SUCCEEDED,
1020 applied_changes=applied_changes,
1021 failure_detail=None,
1022 result_counts=result_counts or {},
1023 )
1025 def mark_failed(self, *, reason, failed_change=None, rolled_back=(), not_attempted=()):
1026 """Record what failed, what rolled back, and what was never attempted."""
1027 return self._finish(
1028 outcome=ExecutionOutcome.FAILED,
1029 applied_changes=None,
1030 result_counts={"created": {}, "errors": 1},
1031 failure_detail={
1032 "failed_change": failed_change,
1033 "rolled_back": list(rolled_back),
1034 "not_attempted": list(not_attempted),
1035 "reason": reason,
1036 },
1037 )
1040class DeviceTypeMapping(PolicySectionModel):
1041 """Explicit (make, model) override when source naming doesn't slugify cleanly."""
1043 POLICY_SECTION = "device_type_mappings"
1045 profile = models.ForeignKey(
1046 ImportProfile,
1047 on_delete=models.CASCADE,
1048 related_name="device_type_mappings",
1049 )
1050 source_make = models.CharField(max_length=200)
1051 source_model = models.CharField(max_length=200)
1052 netbox_manufacturer_slug = models.CharField(max_length=100)
1053 netbox_device_type_slug = models.CharField(max_length=100)
1055 class Meta:
1056 ordering = ["profile", "source_make", "source_model"]
1057 constraints = [
1058 models.UniqueConstraint(
1059 fields=["profile", "source_make", "source_model"], name="ndi_dtm_profile_make_model"
1060 ),
1061 ]
1062 verbose_name = "Device Type Mapping"
1063 verbose_name_plural = "Device Type Mappings"
1065 def __str__(self):
1066 return (
1067 f"{self.source_make} / {self.source_model} → {self.netbox_manufacturer_slug}/{self.netbox_device_type_slug}"
1068 )
1070 def get_absolute_url(self):
1071 """Return the edit URL for this device type mapping."""
1072 return reverse("plugins:netbox_data_import:devicetypemapping_edit", args=[self.pk])
1075class ManufacturerMapping(PolicySectionModel):
1076 """Maps a source 'make' value to an existing NetBox manufacturer slug."""
1078 POLICY_SECTION = "manufacturer_mappings"
1080 profile = models.ForeignKey(
1081 ImportProfile,
1082 on_delete=models.CASCADE,
1083 related_name="manufacturer_mappings",
1084 )
1085 source_make = models.CharField(
1086 max_length=200,
1087 help_text="Exact source make value (e.g. 'Dell EMC')",
1088 )
1089 netbox_manufacturer_slug = models.CharField(
1090 max_length=100,
1091 help_text="NetBox manufacturer slug to map this make to (e.g. 'dell')",
1092 )
1094 class Meta:
1095 ordering = ["profile", "source_make"]
1096 constraints = [
1097 models.UniqueConstraint(fields=["profile", "source_make"], name="ndi_mfgmapping_profile_make"),
1098 ]
1099 verbose_name = "Manufacturer Mapping"
1100 verbose_name_plural = "Manufacturer Mappings"
1102 def __str__(self):
1103 return f"{self.source_make} → {self.netbox_manufacturer_slug}"
1106class InferenceBackend(JobsMixin, NetBoxModel):
1107 """One named Inference Backend definition; the enabled row is the active backend (section 8.2).
1109 JobsMixin attaches the connection test to the row, so its typed result is read where the
1110 configuration lives.
1111 """
1113 ADAPTER_TYPES = _inference_settings.ADAPTER_TYPES
1114 AUTHENTICATION_METHODS = _inference_settings.AUTHENTICATION_METHODS
1115 RESPONSE_MODES = _inference_settings.RESPONSE_MODES
1117 backend_key = models.SlugField(
1118 max_length=100,
1119 unique=True,
1120 help_text="The unique name of this backend, and the only identifier a job payload carries.",
1121 )
1122 display_name = models.CharField(max_length=200)
1123 adapter_type = models.CharField(max_length=50, choices=ADAPTER_TYPES, default="openai_compatible")
1124 api_root = models.CharField(
1125 max_length=500,
1126 help_text="Exact API root without a trailing slash. The client appends /chat/completions.",
1127 )
1128 model = models.CharField(max_length=200, help_text="Exact backend model id. The worker never chooses one.")
1129 authentication = models.CharField(max_length=20, choices=AUTHENTICATION_METHODS, default="bearer")
1130 response_mode = models.CharField(
1131 max_length=20,
1132 choices=RESPONSE_MODES,
1133 default="prompt_json",
1134 help_text="Select a mode other than prompt_json only after verifying the exact backend and model.",
1135 )
1136 credential_reference = models.JSONField(help_text="A typed Vault KV v2 reference. It never holds a secret value.")
1137 # A zero timeout raises in the transport, so the row carries the same floor as the fallback.
1138 connect_timeout = models.PositiveIntegerField(
1139 default=5, validators=[MinValueValidator(_inference_settings.TIMEOUT_MIN)]
1140 )
1141 read_timeout = models.PositiveIntegerField(
1142 default=60, validators=[MinValueValidator(_inference_settings.TIMEOUT_MIN)]
1143 )
1144 enabled = models.BooleanField(default=False, help_text="Whether Ask AI may use this backend.")
1146 # Override tags reverse accessor to avoid clashes with other plugins
1147 tags = models.ManyToManyField(to="extras.Tag", related_name="+", blank=True)
1149 class Meta:
1150 ordering = ["backend_key"]
1151 constraints = [
1152 # A partial unique index over one column value permits exactly one enabled row.
1153 models.UniqueConstraint(
1154 fields=["enabled"],
1155 condition=models.Q(enabled=True),
1156 name="ndi_inferencebackend_one_enabled",
1157 ),
1158 ]
1159 verbose_name = "AI backend"
1160 verbose_name_plural = "AI backends"
1162 def __str__(self):
1163 return self.display_name or self.backend_key
1165 def get_absolute_url(self):
1166 """Return the detail URL for this Inference Backend."""
1167 return reverse("plugins:netbox_data_import:inferencebackend", args=[self.pk])
1169 def clean(self):
1170 """Reject a second enabled row, an unapproved api_root, and a reference that is not typed."""
1171 super().clean()
1172 from .inference_backend import validate_backend_fields
1174 if self.enabled:
1175 competing = type(self).objects.filter(enabled=True).exclude(pk=self.pk)
1176 if competing.exists():
1177 raise ValidationError({"enabled": "Another Inference Backend is already enabled. Disable it first."})
1178 validate_backend_fields(
1179 api_root=self.api_root,
1180 authentication=self.authentication,
1181 credential_reference=self.credential_reference,
1182 )
1185class IgnoredDevice(PolicySectionModel):
1186 """Per-device ignore record — prevents a specific source device from being imported."""
1188 POLICY_SECTION = "ignored_devices"
1190 profile = models.ForeignKey(
1191 ImportProfile,
1192 on_delete=models.CASCADE,
1193 related_name="ignored_devices",
1194 )
1195 source_id = models.CharField(
1196 max_length=200,
1197 help_text="Source ID value that identifies this device",
1198 )
1199 device_name = models.CharField(
1200 max_length=200,
1201 blank=True,
1202 help_text="Original device name (for display only)",
1203 )
1205 class Meta:
1206 ordering = ["profile", "source_id"]
1207 constraints = [
1208 models.UniqueConstraint(fields=["profile", "source_id"], name="ndi_ignoreddevice_profile_srcid"),
1209 ]
1210 verbose_name = "Ignored Device"
1211 verbose_name_plural = "Ignored Devices"
1213 def __str__(self):
1214 return f"{self.device_name or self.source_id} (ignored)"
1217class ColumnTransformRule(PolicySectionModel):
1218 r"""Regex-based transform applied to a source column during parse.
1220 Example: source_column='Name', pattern='^(\w{4,8}) - (.+)$',
1221 group_1_target='asset_tag', group_2_target='device_name'
1222 transforms "TEST0001 - EXAMPLE-SWITCH-01" into asset_tag="TEST0001", device_name="EXAMPLE-SWITCH-01".
1223 """
1225 POLICY_SECTION = "column_transform_rules"
1227 profile = models.ForeignKey(
1228 ImportProfile,
1229 on_delete=models.CASCADE,
1230 related_name="column_transform_rules",
1231 )
1232 source_column = models.CharField(
1233 max_length=200,
1234 help_text="Source Excel column to transform (exact header name)",
1235 )
1236 pattern = models.CharField(
1237 max_length=500,
1238 help_text=(
1239 r"RE2 pattern with capture groups and full-match semantics. "
1240 r"Backreferences and look-around are not supported. E.g. ^(\w+) - (.+)$"
1241 ),
1242 )
1243 group_1_target = models.CharField(
1244 max_length=100,
1245 blank=True,
1246 help_text="Target field for capture group 1 (leave blank to ignore)",
1247 )
1248 group_2_target = models.CharField(
1249 max_length=100,
1250 blank=True,
1251 help_text="Target field for capture group 2 (leave blank to ignore)",
1252 )
1254 class Meta:
1255 ordering = ["profile", "source_column"]
1256 constraints = [
1257 models.UniqueConstraint(fields=["profile", "source_column"], name="ndi_ctr_profile_column"),
1258 ]
1259 verbose_name = "Column Transform Rule"
1260 verbose_name_plural = "Column Transform Rules"
1262 def clean(self):
1263 """Validate the regex, capture groups, and exclusive ownership of group targets."""
1264 from django.core.exceptions import ValidationError
1266 from .transform_regex import TransformPattern, TransformPatternError
1268 super().clean()
1270 try:
1271 compiled = TransformPattern.compile(self.pattern)
1272 except TransformPatternError as exc:
1273 raise ValidationError({"pattern": f"Regex pattern is not supported: {exc}"}) from exc
1275 required_groups = 0
1276 if self.group_1_target:
1277 required_groups = 1
1278 if self.group_2_target:
1279 required_groups = 2
1280 if compiled.group_count < required_groups:
1281 raise ValidationError(
1282 {
1283 "pattern": (
1284 f"Regex must contain at least {required_groups} capture group(s) "
1285 f"for the configured group target(s), but found {compiled.group_count}."
1286 )
1287 }
1288 )
1290 output_kinds = self.profile.output_kinds if self.profile_id else None
1291 target_attrs = {}
1292 for attr in ("group_1_target", "group_2_target"):
1293 value = getattr(self, attr) or ""
1294 if not value:
1295 continue
1296 # A capture group yields text, so a candidate target is not a valid group target.
1297 if not CATALOG.is_valid(value, output_kinds=output_kinds, allow_candidates=False):
1298 raise ValidationError({attr: CATALOG.invalid_key_message(value)})
1299 previous_attr = target_attrs.get(value)
1300 if previous_attr is not None:
1301 previous_group = previous_attr.removeprefix("group_").removesuffix("_target")
1302 raise ValidationError(
1303 {attr: f"Target field '{value}' is already assigned to capture group {previous_group}."}
1304 )
1305 target_attrs[value] = attr
1307 if not self.profile_id or not target_attrs:
1308 return
1309 conflicts = (
1310 type(self)
1311 .objects.filter(profile_id=self.profile_id)
1312 .exclude(pk=self.pk)
1313 .filter(models.Q(group_1_target__in=target_attrs) | models.Q(group_2_target__in=target_attrs))
1314 .only("source_column", "group_1_target", "group_2_target")
1315 )
1316 errors = {}
1317 for conflict in conflicts:
1318 for target in (conflict.group_1_target, conflict.group_2_target):
1319 attr = target_attrs.get(target)
1320 if attr is not None:
1321 errors[attr] = (
1322 f"Target field '{target}' is already assigned by the transform rule "
1323 f"for source column '{conflict.source_column}'."
1324 )
1325 errors.update(self._column_mapping_target_errors(target_attrs))
1326 if errors:
1327 raise ValidationError(errors)
1329 def _column_mapping_target_errors(self, target_attrs):
1330 """Return capture errors for targets already owned by direct mappings."""
1331 mapped_targets = {
1332 mapping.target_field: mapping.source_column
1333 for mapping in ColumnMapping.objects.filter(
1334 profile_id=self.profile_id,
1335 target_field__in=target_attrs,
1336 ).only("source_column", "target_field")
1337 }
1338 return {
1339 attr: (
1340 f"Target field '{target}' is already assigned by the column mapping "
1341 f"for source column '{mapped_targets[target]}'."
1342 )
1343 for target, attr in target_attrs.items()
1344 if target in mapped_targets
1345 }
1347 def __str__(self):
1348 return f"{self.source_column}: {self.pattern}"
1350 def get_absolute_url(self):
1351 """Return the edit URL for this column transform rule."""
1352 return reverse("plugins:netbox_data_import:columntransformrule_edit", args=[self.pk])
1355class SourceResolution(PolicySectionModel):
1356 """Saved target-field decision for one source row.
1358 A resolution can split one source value or select candidate source columns
1359 for structured target fields. The import reapplies it when the same source
1360 row appears in a later file.
1361 """
1363 POLICY_SECTION = "source_resolutions"
1365 profile = models.ForeignKey(
1366 ImportProfile,
1367 on_delete=models.CASCADE,
1368 related_name="source_resolutions",
1369 )
1370 source_id = models.CharField(
1371 max_length=200,
1372 help_text="Source ID of the row this resolution applies to",
1373 )
1374 source_column = models.CharField(
1375 max_length=200,
1376 help_text="Column name this resolution applies to",
1377 )
1378 original_value = models.TextField(
1379 help_text="Original cell value before resolution",
1380 )
1381 resolved_fields = models.JSONField(
1382 default=dict,
1383 help_text="Dict of target_field -> resolved_value (e.g. {'device_name': 'SW1', 'asset_tag': 'TEST0001'})",
1384 )
1386 class Meta:
1387 ordering = ["profile", "source_id"]
1388 constraints = [
1389 models.UniqueConstraint(
1390 fields=["profile", "source_id", "source_column"], name="ndi_srcresolution_profile_id_col"
1391 ),
1392 ]
1393 verbose_name = "Source Resolution"
1394 verbose_name_plural = "Source Resolutions"
1396 def clean(self):
1397 """Require target-field decisions that can safely merge into the profile's source rows."""
1398 super().clean()
1399 validate_source_resolution_fields(
1400 self.profile if self.profile_id else None,
1401 self.source_column,
1402 self.resolved_fields,
1403 )
1405 def __str__(self):
1406 return f"{self.source_id}/{self.source_column}: {self.original_value!r}"
1409class DeviceExistingMatch(PolicySectionModel):
1410 """Explicit match between a source row and an existing NetBox device.
1412 When a user clicks "Link existing" on a device preview row, this record is saved.
1413 On re-import, the engine uses this to emit action='update' against the matched device
1414 instead of action='create', even if the device has no source-ID custom field yet.
1415 """
1417 POLICY_SECTION = "device_existing_matches"
1419 profile = models.ForeignKey(
1420 ImportProfile,
1421 on_delete=models.CASCADE,
1422 related_name="device_matches",
1423 )
1424 source_id = models.CharField(
1425 max_length=200,
1426 help_text="Source ID value that identifies this row",
1427 )
1428 source_asset_tag = models.CharField(
1429 max_length=100,
1430 blank=True,
1431 default="",
1432 help_text="Asset tag from source row (for display / lookup; may become stale)",
1433 )
1434 netbox_device_id = models.PositiveIntegerField(
1435 help_text="Primary key of the matched NetBox Device",
1436 )
1437 device_name = models.CharField(
1438 max_length=200,
1439 blank=True,
1440 help_text="NetBox device name (for display only; may become stale)",
1441 )
1443 class Meta:
1444 ordering = ["profile", "source_id"]
1445 constraints = [
1446 models.UniqueConstraint(fields=["profile", "source_id"], name="ndi_devicematch_profile_srcid"),
1447 models.UniqueConstraint(
1448 fields=["profile", "netbox_device_id"],
1449 name="ndi_devicematch_profile_device",
1450 ),
1451 ]
1452 verbose_name = "Device Existing Match"
1453 verbose_name_plural = "Device Existing Matches"
1455 def __str__(self):
1456 tag = f" / {self.source_asset_tag}" if self.source_asset_tag else ""
1457 return f"{self.source_id}{tag} → Device #{self.netbox_device_id} ({self.device_name})"
1460class IgnoredFieldDifference(PolicySectionModel):
1461 """Preserve one exact file/NetBox value pair for a device field difference."""
1463 POLICY_SECTION = "ignored_field_differences"
1465 profile = models.ForeignKey(
1466 ImportProfile,
1467 on_delete=models.CASCADE,
1468 related_name="ignored_field_differences",
1469 )
1470 source_id = models.CharField(
1471 max_length=200,
1472 help_text="Source ID of the row this review applies to",
1473 )
1474 netbox_device_id = models.PositiveIntegerField(
1475 help_text="Primary key of the matched NetBox Device",
1476 )
1477 target_field = models.CharField(
1478 max_length=100,
1479 help_text="Target field whose current difference is ignored",
1480 )
1481 file_snapshot = models.JSONField(
1482 default=dict,
1483 help_text="Normalized and display values from the source row",
1484 )
1485 netbox_snapshot = models.JSONField(
1486 default=dict,
1487 help_text="Normalized and display values from the matched NetBox device",
1488 )
1490 class Meta:
1491 ordering = ["profile", "source_id", "target_field"]
1492 constraints = [
1493 models.UniqueConstraint(
1494 fields=["profile", "source_id", "netbox_device_id", "target_field"],
1495 name="ndi_ignored_diff_profile_source_device_field",
1496 ),
1497 ]
1498 verbose_name = "Ignored Field Difference"
1499 verbose_name_plural = "Ignored Field Differences"
1501 def __str__(self):
1502 return f"{self.source_id}/{self.target_field} on device #{self.netbox_device_id} (ignored)"
1505class DeviceImportSource(models.Model):
1506 """Import provenance the plugin keeps for one Device.
1508 Replaces the plugin-managed ``data_import_source`` custom field. The per-profile custom
1509 field an operator configures (``ImportProfile.custom_field_name``) is separate and stays.
1510 """
1512 device = models.OneToOneField(
1513 to="dcim.Device",
1514 on_delete=models.CASCADE,
1515 related_name="data_import_source",
1516 )
1517 profile = models.ForeignKey(
1518 ImportProfile,
1519 on_delete=models.CASCADE,
1520 related_name="device_sources",
1521 )
1522 source_id = models.CharField(
1523 max_length=200,
1524 blank=True,
1525 default="",
1526 help_text="Source ID of the row that wrote this device",
1527 )
1528 extra_columns = models.JSONField(
1529 default=dict,
1530 blank=True,
1531 help_text="Source column values that no mapping consumes",
1532 )
1533 unassigned_ips = models.JSONField(
1534 default=dict,
1535 blank=True,
1536 help_text="IP values the import could not assign to a NetBox IP field",
1537 )
1539 class Meta:
1540 ordering = ["device"]
1541 indexes = [models.Index(fields=["profile", "source_id"])]
1542 verbose_name = "Device Import Source"
1543 verbose_name_plural = "Device Import Sources"
1545 def __str__(self):
1546 return f"{self.source_id or '(no source ID)'} → Device #{self.device_id}"
1549class CableImportSource(models.Model):
1550 """Import provenance the plugin keeps for one Cable and one contributing Source Trace.
1552 Two Source Traces that state one identical segment share one created Cable, so the Cable
1553 reference is a plain foreign key and the row is keyed by the trace as well (section 5.7).
1554 """
1556 cable = models.ForeignKey(
1557 to="dcim.Cable",
1558 on_delete=models.CASCADE,
1559 related_name="data_import_sources",
1560 )
1561 profile = models.ForeignKey(
1562 ImportProfile,
1563 on_delete=models.CASCADE,
1564 related_name="cable_sources",
1565 )
1566 trace_identity = models.TextField(
1567 help_text="Canonical JSON identity of the Source Trace that states this segment",
1568 )
1569 trace_key = models.CharField(
1570 max_length=64,
1571 blank=True,
1572 editable=False,
1573 help_text="Fixed-width digest of trace_identity, which is what the index and constraint carry",
1574 )
1575 segment_index = models.PositiveIntegerField(
1576 help_text="Position of this segment in the Source Trace, in canonical order",
1577 )
1578 from_text = models.TextField(blank=True, default="")
1579 to_text = models.TextField(blank=True, default="")
1580 direction = models.CharField(max_length=20, blank=True, default="")
1581 workbook_fingerprint = models.CharField(max_length=64, blank=True, default="")
1582 sheet = models.CharField(max_length=100, blank=True, default="")
1583 block_ordinal = models.PositiveIntegerField(null=True, blank=True)
1584 row_start = models.PositiveIntegerField(null=True, blank=True)
1585 row_end = models.PositiveIntegerField(null=True, blank=True)
1586 export_timestamp = models.CharField(
1587 max_length=TRACE_EXPORT_TIMESTAMP_MAX_LENGTH,
1588 blank=True,
1589 default="",
1590 )
1592 def clean(self):
1593 """Derive the index key before validate_unique reads the constraint's own fields."""
1594 super().clean()
1595 self.trace_key = index_digest(self.trace_identity)
1597 def save(self, *args, **kwargs):
1598 """Derive the index key, so no caller can store one that disagrees with the identity."""
1599 self.trace_key = index_digest(self.trace_identity)
1600 update_fields = kwargs.get("update_fields")
1601 # A partial save of the identity alone would leave the constraint on the digest it replaced.
1602 if update_fields is not None and "trace_identity" in update_fields:
1603 kwargs["update_fields"] = {*update_fields, "trace_key"}
1604 super().save(*args, **kwargs)
1606 class Meta:
1607 ordering = ["cable", "profile", "trace_identity"]
1608 constraints = [
1609 models.UniqueConstraint(
1610 fields=["cable", "profile", "trace_key"],
1611 name="ndi_cableimportsource_cable_profile_trace",
1612 ),
1613 ]
1614 indexes = [models.Index(fields=["profile", "trace_key"])]
1615 verbose_name = "Cable Import Source"
1616 verbose_name_plural = "Cable Import Sources"
1618 def __str__(self):
1619 return f"segment {self.segment_index} of {self.from_text} to {self.to_text} \u2192 Cable #{self.cable_id}"
1622def stored_import_source(obj):
1623 """Return the plugin's import record for one object, or None when it holds none."""
1624 from dcim.models import Device
1626 if not isinstance(obj, Device) or obj.pk is None:
1627 return None
1628 return DeviceImportSource.objects.filter(device_id=obj.pk).first()