Coverage for netbox_data_import/models.py: 98.65%
962 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 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 utilities.querysets import RestrictedQuerySet
18from .adapters import (
19 DEFAULT_ADAPTER_KEY,
20 UnknownSourceAdapter,
21 adapter_choices,
22 get_adapter,
23 output_kinds_for,
24)
25from . import branching, plan
26from .cable_policy import cable_profile_label, cable_type_label, policy_choice_errors
27from .catalog import CATALOG, POLICY_SECTIONS, has_implemented_module, policy_section
28from .field_keys import (
29 PORT_CLASS_CHOICES,
30 SELECT_TERMINATION_TASK,
31 parse_termination_field_key,
32 port_class_mapping_errors,
33)
34from . import inference_settings as _inference_settings
35from .proposal_contract import OUTCOME_CANDIDATE, OUTCOME_CHOICES, OUTCOME_NO_MATCH
36from .trace_schema import TRACE_EXPORT_TIMESTAMP_MAX_LENGTH
37from .identity import identity_text
39CONTACT_RESOLUTION_FIELDS = frozenset({"name", "email", "phone"})
40CONTACT_RESOLUTION_REQUIRED_KEYS = frozenset({"contact_resolution_applied", "contact_field_sources"})
41CONTACT_RESOLUTION_KEYS = CONTACT_RESOLUTION_REQUIRED_KEYS | frozenset({"contact_field_values", "contact_id"})
42SOURCE_RESOLUTION_RESERVED_FIELDS = frozenset({"source_id", "_conflicts"})
45def _invalid_contact_resolution_key_message(key):
46 """Describe one unknown Contact resolution key using the Contact-policy vocabulary."""
47 allowed = ", ".join(f"'{field}'" for field in sorted(CONTACT_RESOLUTION_KEYS))
48 return f"'{key}' is not a Contact resolution field. Use one of {allowed}."
51def validate_adapter_target_module(adapter_key):
52 """Reject a Source Adapter whose Target Module this release does not implement yet."""
53 if not has_implemented_module(output_kinds_for(adapter_key)):
54 raise ValidationError(
55 {"source_adapter": f"This release cannot import from the '{adapter_key}' source adapter yet."}
56 )
59def validate_registered_adapter(profile):
60 """Reject a profile whose stored Source Adapter this release does not register."""
61 if profile is not None and profile.adapter is None:
62 raise ValidationError(
63 f"This profile uses the source adapter '{profile.source_adapter}', which this release does not register."
64 )
67def validate_section_applicability(profile, section_key):
68 """Reject a policy row whose section does not apply to its profile's Source Adapter."""
69 section = policy_section(section_key)
70 if section is None or profile is None:
71 return
72 if not section.applies_to(profile.output_kinds):
73 raise ValidationError(
74 f"{section.label} do not apply to a profile using the '{profile.source_adapter}' source adapter."
75 )
78def _validated_contact_id(contact_id):
79 """Return a saved Contact ID as a positive int, rejecting anything int() would reshape."""
80 if contact_id in ("", None):
81 return None
82 # int() truncates, so a JSON float would silently select a different Contact.
83 if isinstance(contact_id, bool) or (isinstance(contact_id, float) and not contact_id.is_integer()):
84 raise ValidationError("The selected Contact ID is invalid.")
85 try:
86 contact_id = int(contact_id)
87 except (TypeError, ValueError) as exc:
88 raise ValidationError("The selected Contact ID is invalid.") from exc
89 if contact_id < 1:
90 raise ValidationError("The selected Contact ID is invalid.")
91 return contact_id
94def validate_contact_candidate_resolution(
95 resolved_fields,
96 lookup_field: str,
97 available_source_columns,
98) -> dict:
99 """Validate and normalize one saved Contact candidate resolution."""
100 if not isinstance(resolved_fields, dict) or not set(resolved_fields) >= CONTACT_RESOLUTION_REQUIRED_KEYS:
101 raise ValidationError("The Contact candidate resolution has an invalid structure.")
102 invalid = sorted(set(resolved_fields) - CONTACT_RESOLUTION_KEYS)
103 if invalid:
104 raise ValidationError(_invalid_contact_resolution_key_message(invalid[0]))
105 if resolved_fields.get("contact_resolution_applied") is not True:
106 raise ValidationError("The Contact candidate resolution is not marked as applied.")
108 field_sources = resolved_fields.get("contact_field_sources")
109 if not isinstance(field_sources, dict) or set(field_sources) - CONTACT_RESOLUTION_FIELDS:
110 raise ValidationError("The Contact candidate resolution contains an unknown field.")
111 if any(not isinstance(source_column, str) or not source_column for source_column in field_sources.values()):
112 raise ValidationError("Each resolved Contact field must select one source column.")
113 missing_sources = set(field_sources.values()) - set(available_source_columns)
114 if missing_sources:
115 missing = min(missing_sources)
116 raise ValidationError(f"The source column '{missing}' has no candidate value in this row.")
118 field_values = resolved_fields.get("contact_field_values", {})
119 if not isinstance(field_values, dict) or set(field_values) - CONTACT_RESOLUTION_FIELDS:
120 raise ValidationError("The Contact candidate resolution contains an unknown literal field.")
121 if any(not isinstance(value, str) or not value.strip() for value in field_values.values()):
122 raise ValidationError("Each literal Contact field must contain text.")
123 overlap = set(field_sources) & set(field_values)
124 if overlap:
125 raise ValidationError(f"Select a source column or enter a value for Contact {min(overlap)}, not both.")
127 contact_id = _validated_contact_id(resolved_fields.get("contact_id"))
129 supplied_fields = set(field_sources) | set(field_values)
130 if supplied_fields and "name" not in supplied_fields:
131 raise ValidationError("Select a source column or enter a value for the Contact name.")
132 if supplied_fields and lookup_field not in supplied_fields:
133 raise ValidationError(f"Select a source column or enter a value for the Contact {lookup_field} lookup field.")
134 return {
135 "field_sources": field_sources,
136 "field_values": {field: value.strip() for field, value in field_values.items()},
137 "contact_id": contact_id,
138 }
141def validate_source_resolution_fields(profile, source_column, resolved_fields):
142 """Reject resolved fields that cannot safely merge into one source row."""
143 if not isinstance(resolved_fields, dict):
144 raise ValidationError({"resolved_fields": "Enter the resolved fields as a JSON object."})
145 if any(not isinstance(key, str) for key in resolved_fields):
146 raise ValidationError({"resolved_fields": "Each resolved field name must be text."})
148 reserved = sorted(set(resolved_fields) & SOURCE_RESOLUTION_RESERVED_FIELDS)
149 if reserved:
150 raise ValidationError(
151 {"resolved_fields": f"The resolved field '{reserved[0]}' is reserved for import planning."}
152 )
154 if source_column == "candidate:contact":
155 invalid = sorted(set(resolved_fields) - CONTACT_RESOLUTION_KEYS)
156 if invalid:
157 raise ValidationError({"resolved_fields": _invalid_contact_resolution_key_message(invalid[0])})
158 return
159 output_kinds = profile.output_kinds if profile is not None else None
160 invalid = sorted(
161 key for key in resolved_fields if not CATALOG.is_valid(key, output_kinds=output_kinds, allow_candidates=False)
162 )
163 if invalid:
164 raise ValidationError({"resolved_fields": CATALOG.invalid_key_message(invalid[0])})
167@contextmanager
168def locked_profile_policy(*profile_ids):
169 """Hold the given profile rows for a policy write or for an import execution.
171 Every SourceResolution write and the import worker take this same lock, so a decision cannot
172 commit between the worker's check and its writes. Locking the resolution rows alone would leave
173 an insert free to land in that window, because a row that does not exist yet cannot be locked.
175 The rows lock in primary-key order, so two callers naming several profiles cannot deadlock by
176 taking them in opposite orders.
177 """
178 branching.refuse_branch()
179 wanted = sorted({profile_id for profile_id in profile_ids if profile_id is not None})
180 # Django short-circuits `pk__in=[]`, so an empty set would yield without ever taking a lock.
181 if not wanted:
182 raise ImportProfile.DoesNotExist("A policy write must name at least one ImportProfile to lock.")
183 with transaction.atomic():
184 locked = ImportProfile.objects.select_for_update().filter(pk__in=wanted).order_by("pk")
185 if len(locked) != len(wanted):
186 raise ImportProfile.DoesNotExist(f"No ImportProfile matches every id in {wanted}.")
187 yield
190@contextmanager
191def locked_resolution_policy(resolution_pk):
192 """Hold the profile a saved resolution belongs to, read from the database rather than trusted.
194 A caller reaches this holding an instance it fetched earlier, whose profile may be a stale copy.
195 The row is read again under the lock, so the caller acts on a row that still exists and still
196 belongs to the locked profile.
197 """
198 branching.refuse_branch()
199 gone = SourceResolution.DoesNotExist(f"No SourceResolution matches id {resolution_pk}.")
200 profile_id = SourceResolution.objects.filter(pk=resolution_pk).values_list("profile_id", flat=True).first()
201 if profile_id is None:
202 raise gone
203 with locked_profile_policy(profile_id):
204 # A delete can still commit in the gap above, and a write that saw the row would resurrect it.
205 if not SourceResolution.objects.filter(pk=resolution_pk, profile_id=profile_id).exists():
206 raise gone
207 yield
210class ImportProfile(NetBoxModel):
211 """Named configuration for one source file format."""
213 name = models.CharField(max_length=100, unique=True)
214 description = models.TextField(blank=True)
215 source_adapter = models.CharField(
216 max_length=50,
217 choices=adapter_choices,
218 default=DEFAULT_ADAPTER_KEY,
219 help_text="Source format this profile reads. It cannot change after creation.",
220 )
221 adapter_config = models.JSONField(
222 default=dict,
223 blank=True,
224 help_text="Scalar settings the selected Source Adapter declares.",
225 )
227 class Meta:
228 ordering = ["name"]
229 verbose_name = "Import Profile"
230 verbose_name_plural = "Import Profiles"
232 def __str__(self):
233 return self.name
235 def get_absolute_url(self):
236 """Return the detail URL for this import profile."""
237 return reverse("plugins:netbox_data_import:importprofile", args=[self.pk])
239 def _validate_source_adapter_immutability(self):
240 """Return the persisted adapter and reject a different selected adapter."""
241 # A set pk does not prove that the row exists. An unsaved instance can carry a pk.
242 stored = (
243 type(self).objects.filter(pk=self.pk).values_list("source_adapter", flat=True).first()
244 if self.pk is not None
245 else None
246 )
247 if stored is not None and stored != self.source_adapter:
248 raise ValidationError({"source_adapter": "The source adapter cannot change after the profile is created."})
249 return stored
251 def save(self, *args, **kwargs):
252 """Normalize adapter configuration on every supported write that stores it."""
253 update_fields = kwargs.get("update_fields")
254 updated = set(update_fields) if update_fields is not None else None
255 if updated is None or updated & {"source_adapter", "adapter_config"}:
256 self._validate_source_adapter_immutability()
257 if updated is None or "adapter_config" in updated:
258 adapter = self.adapter
259 if adapter is None:
260 raise ValidationError({"source_adapter": f"Unknown source adapter '{self.source_adapter}'."})
261 self.adapter_config = adapter.config_form_class().validate_config(self.adapter_config)
262 return super().save(*args, **kwargs)
264 def delete(self, *args, **kwargs):
265 """Take the policy lock before the cascade, which would otherwise take the child rows first.
267 A policy write holds this row and then writes a child, so a cascade in the opposite order
268 deadlocks against it. NetBox deletes each object through this method, in bulk as well.
269 """
270 with locked_profile_policy(self.pk):
271 # atomic-exit-safe: locked-cascade-committed
272 return super().delete(*args, **kwargs)
274 @property
275 def adapter(self):
276 """Return the registered Source Adapter class for this profile."""
277 return get_adapter(self.source_adapter)
279 @property
280 def output_kinds(self) -> frozenset[str]:
281 """Return the adapter output kinds this profile can supply."""
282 return output_kinds_for(self.source_adapter)
284 @property
285 def adapter_settings(self):
286 """Return attribute access over ``adapter_config`` backed by the adapter's declared defaults."""
287 cache_key = (self.source_adapter, id(self.adapter_config))
288 cached = self.__dict__.get("_adapter_settings_cache")
289 if cached is not None and cached[0] == cache_key:
290 return cached[1]
291 settings = AdapterSettings(self.adapter, self.adapter_config, self.source_adapter)
292 self.__dict__["_adapter_settings_cache"] = (cache_key, settings)
293 return settings
295 @property
296 def adapter_config_display(self):
297 """Return (label, value) pairs for the adapter's declared settings, in declaration order."""
298 from django import forms
299 from django.forms.utils import pretty_name
301 config = _require_adapter_config_mapping(self.adapter_config)
302 adapter = self.adapter
303 if adapter is None:
304 return []
305 rows = []
306 for name, field in adapter.config_form_class().base_fields.items():
307 value = config.get(name, field.initial)
308 if isinstance(field, forms.ChoiceField) and not isinstance(field, forms.ModelChoiceField):
309 value = dict(field.choices).get(value, value)
310 rows.append((field.label or pretty_name(name), value))
311 return rows
313 def grouped_column_map(self) -> dict[str, list[str]]:
314 """Return this profile's mapped source columns, keyed by Target Field."""
315 grouped: dict[str, list[str]] = {}
316 # Two columns can feed one Target Field, and which one wins must not be a query-order accident.
317 for mapping in self.column_mappings.order_by("target_field", "pk"):
318 grouped.setdefault(mapping.target_field, []).append(mapping.source_column)
319 return grouped
321 @property
322 def planning_fingerprint(self) -> str:
323 """Return the fingerprint of every profile value planning depends on."""
324 related_sections = {
325 getattr(relation.related_model, "POLICY_SECTION", ""): relation.get_accessor_name()
326 for relation in self._meta.related_objects
327 }
328 sections = []
329 for section in POLICY_SECTIONS:
330 accessor = related_sections[section.key]
331 serialized_rows = []
332 for row in getattr(self, accessor).all():
333 serialized_rows.append(
334 {
335 field.name: field.value_from_object(row)
336 for field in row._meta.concrete_fields
337 if field.name not in {"id", "profile"}
338 }
339 )
340 serialized_rows.sort(key=plan.canonical_json)
341 sections.append({"key": section.key, "rows": serialized_rows})
342 return plan.fingerprint_of(
343 {
344 "profile_id": self.pk,
345 "source_adapter": self.source_adapter,
346 "adapter_config": self.adapter_config,
347 "policy_sections": sections,
348 }
349 )
351 @property
352 def resolved_primary_contact_role(self):
353 """Return the referenced Contact Role object, or None when unset or dangling.
355 Planning reads this once per row, so the lookup is memoized against the configured name. A
356 plain instance cache would keep returning the old role after ``adapter_config`` changes.
357 """
358 name = self.adapter_settings.primary_contact_role
359 if not name:
360 return None
361 cached = self.__dict__.get("_primary_contact_role_cache")
362 if cached is not None and cached[0] == name:
363 return cached[1]
364 from tenancy.models import ContactRole
366 role = ContactRole.objects.filter(name=name).first()
367 self.__dict__["_primary_contact_role_cache"] = (name, role)
368 return role
370 def clean(self):
371 """Reject an adapter change after creation and validate the adapter configuration."""
372 super().clean()
373 adapter = self.adapter
374 if adapter is None:
375 raise ValidationError({"source_adapter": f"Unknown source adapter '{self.source_adapter}'."})
376 stored = self._validate_source_adapter_immutability()
377 if stored is None:
378 # A creation rule only: the adapter is immutable, so a stored profile keeps validating.
379 validate_adapter_target_module(self.source_adapter)
380 self.adapter_config = adapter.config_form_class().validate_config(self.adapter_config)
383def _require_adapter_config_mapping(config):
384 """Return a stored adapter configuration mapping or expose corrupt JSON state."""
385 if not isinstance(config, dict):
386 raise ValidationError(
387 "The stored adapter configuration must be a mapping. Replace it with a valid JSON mapping."
388 )
389 return config
392class AdapterSettings:
393 """Read one adapter setting, falling back to the adapter form's declared default."""
395 def __init__(self, adapter, config, adapter_key):
396 self._adapter_key = adapter_key
397 self._fields = adapter.config_form_class().base_fields if adapter is not None else None
398 self._config = _require_adapter_config_mapping(config)
400 def get(self, name, default):
401 """Return one setting, or *default* when this profile's adapter declares none.
403 Only a caller that serves every adapter may ask this way. Adapter-specific code reads the
404 attribute, so a setting it depends on cannot go missing quietly.
405 """
406 fields = object.__getattribute__(self, "_fields")
407 return getattr(self, name) if fields is not None and name in fields else default
409 def __getattr__(self, name):
410 fields = object.__getattribute__(self, "_fields")
411 if fields is None:
412 key = object.__getattribute__(self, "_adapter_key")
413 raise UnknownSourceAdapter(f"This release does not register the source adapter '{key}'.")
414 if name not in fields:
415 raise AttributeError(f"'{name}' is not a setting of this profile's source adapter")
416 config = object.__getattribute__(self, "_config")
417 field = fields[name]
418 if name not in config:
419 return field.initial
420 value = config[name]
421 if field.required and (value is None or value == ""):
422 raise ValidationError(f"The required adapter setting '{name}' is empty. Edit and save this import profile.")
423 return value
426class PolicySectionModel(models.Model):
427 """A profile policy table scoped to the adapter output kinds its catalog section declares."""
429 POLICY_SECTION = ""
431 # NetBox's generic views scope a queryset with `restrict()`, which only this manager provides.
432 objects = RestrictedQuerySet.as_manager()
434 class Meta:
435 abstract = True
437 def clean(self):
438 """Reject a row whose section does not apply to the profile's Source Adapter."""
439 super().clean()
440 validate_section_applicability(self.profile if self.profile_id else None, self.POLICY_SECTION)
443class ColumnMapping(PolicySectionModel):
444 """Maps one source column header to one semantic NetBox field."""
446 POLICY_SECTION = "column_mappings"
448 profile = models.ForeignKey(
449 ImportProfile,
450 on_delete=models.CASCADE,
451 related_name="column_mappings",
452 )
453 source_column = models.CharField(
454 max_length=200,
455 help_text="Exact column header in the source file (case-sensitive)",
456 )
457 target_field = models.CharField(max_length=100)
459 class Meta:
460 ordering = ["profile", "target_field"]
461 verbose_name = "Column Mapping"
462 verbose_name_plural = "Column Mappings"
464 def clean(self):
465 """Resolve the target field through the catalog and reject an inapplicable row."""
466 super().clean()
467 value = self.target_field or ""
468 if not CATALOG.is_valid(value, output_kinds=self.profile.output_kinds if self.profile_id else None):
469 raise ValidationError({"target_field": CATALOG.invalid_key_message(value)})
470 if not self.profile_id or not value:
471 return
472 conflict = (
473 ColumnTransformRule.objects.filter(profile_id=self.profile_id)
474 .filter(models.Q(group_1_target=value) | models.Q(group_2_target=value))
475 .only("source_column")
476 .first()
477 )
478 if conflict is not None:
479 raise ValidationError(
480 {
481 "target_field": (
482 f"Target field '{value}' is already assigned by the transform rule "
483 f"for source column '{conflict.source_column}'."
484 )
485 }
486 )
488 def get_target_field_display(self):
489 """Return the human-readable name for the target_field value."""
490 return CATALOG.display(self.target_field)
492 def __str__(self):
493 return f"{self.source_column} → {self.get_target_field_display()}"
495 def get_absolute_url(self):
496 """Return the edit URL for this column mapping."""
497 return reverse("plugins:netbox_data_import:columnmapping_edit", args=[self.pk])
500class ClassRoleMapping(PolicySectionModel):
501 """Maps a source 'class' value to a NetBox outcome (rack or device role)."""
503 POLICY_SECTION = "class_role_mappings"
505 profile = models.ForeignKey(
506 ImportProfile,
507 on_delete=models.CASCADE,
508 related_name="class_role_mappings",
509 )
510 source_class = models.CharField(
511 max_length=200,
512 help_text="Value from the class column (e.g. 'Server', 'Cabinet')",
513 )
514 creates_rack = models.BooleanField(
515 default=False,
516 help_text="If checked, rows with this class create a Rack instead of a Device",
517 )
518 rack_type = models.ForeignKey(
519 to="dcim.RackType",
520 on_delete=models.SET_NULL,
521 null=True,
522 blank=True,
523 related_name="+",
524 help_text="Optional rack type assigned when creating racks",
525 )
526 role_slug = models.CharField(
527 max_length=100,
528 blank=True,
529 help_text="NetBox device role slug (ignored when 'creates rack' is checked)",
530 )
531 ignore = models.BooleanField(
532 default=False,
533 help_text="If checked, rows with this class are silently skipped (not shown as errors)",
534 )
536 class Meta:
537 ordering = ["profile", "source_class"]
538 constraints = [
539 models.UniqueConstraint(fields=["profile", "source_class"], name="ndi_classrolemapping_profile_class"),
540 ]
541 verbose_name = "Class → Role Mapping"
542 verbose_name_plural = "Class → Role Mappings"
544 def __str__(self):
545 if self.creates_rack:
546 suffix = f" ({self.rack_type})" if self.rack_type_id else ""
547 return f"{self.source_class} → Rack{suffix}"
548 return f"{self.source_class} → {self.role_slug}"
550 def get_absolute_url(self):
551 """Return the edit URL for this class→role mapping."""
552 return reverse("plugins:netbox_data_import:classrolemapping_edit", args=[self.pk])
555class CableClassMapping(PolicySectionModel):
556 """Map one source CableClass to independent Cable Type and Cable Profile decisions."""
558 POLICY_SECTION = "cable_class_mappings"
560 profile = models.ForeignKey(
561 ImportProfile,
562 on_delete=models.CASCADE,
563 related_name="cable_class_mappings",
564 )
565 cable_class = models.CharField(max_length=200)
566 cable_type_resolved = models.BooleanField(default=False)
567 cable_type = models.CharField(max_length=50, null=True, blank=True)
568 cable_profile_resolved = models.BooleanField(default=False)
569 cable_profile = models.CharField(max_length=50, null=True, blank=True)
571 class Meta:
572 ordering = ["profile", "cable_class"]
573 constraints = [
574 models.UniqueConstraint(
575 fields=["profile", "cable_class"],
576 name="ndi_cableclassmapping_profile_class",
577 ),
578 models.CheckConstraint(
579 condition=models.Q(cable_type__isnull=True)
580 | (models.Q(cable_type_resolved=True) & ~models.Q(cable_type="")),
581 name="ndi_cableclassmapping_type_resolved_value",
582 ),
583 models.CheckConstraint(
584 condition=models.Q(cable_profile__isnull=True)
585 | (models.Q(cable_profile_resolved=True) & ~models.Q(cable_profile="")),
586 name="ndi_cableclassmapping_profile_resolved_value",
587 ),
588 ]
589 verbose_name = "CableClass Mapping"
590 verbose_name_plural = "CableClass Mappings"
592 def clean(self):
593 """Reject inapplicable, inconsistent, stale, or incompatible mapping values."""
594 super().clean()
595 self.cable_type = self.cable_type or None
596 self.cable_profile = self.cable_profile or None
597 errors = policy_choice_errors(self.cable_type, self.cable_profile)
598 if not self.cable_type_resolved and self.cable_type is not None:
599 errors["cable_type"] = ValidationError(
600 "An unresolved Cable Type cannot store a selected value.",
601 code="invalid",
602 )
603 if not self.cable_profile_resolved and self.cable_profile is not None:
604 errors["cable_profile"] = ValidationError(
605 "An unresolved Cable Profile cannot store a selected value.",
606 code="invalid",
607 )
608 if errors:
609 raise ValidationError(errors)
611 def decided_policy(self):
612 """Return the Cable policy this row decides, or None while either dimension is unresolved."""
613 if not (self.cable_type_resolved and self.cable_profile_resolved):
614 return None
615 return {"cable_type": self.cable_type, "cable_profile": self.cable_profile}
617 def cable_type_display(self):
618 """Return the operator-facing Cable Type decision."""
619 return "Unresolved" if not self.cable_type_resolved else cable_type_label(self.cable_type)
621 def cable_profile_display(self):
622 """Return the operator-facing Cable Profile decision."""
623 return "Unresolved" if not self.cable_profile_resolved else cable_profile_label(self.cable_profile)
625 def __str__(self):
626 return self.cable_class
628 def get_absolute_url(self):
629 """Return the edit URL for this CableClass mapping."""
630 return reverse("plugins:netbox_data_import:cableclassmapping_edit", args=[self.pk])
633class PortClassMapping(PolicySectionModel):
634 """Map one source PortClass label outside the fixed vocabulary to one fixed PortClass."""
636 POLICY_SECTION = "port_class_mappings"
638 profile = models.ForeignKey(
639 ImportProfile,
640 on_delete=models.CASCADE,
641 related_name="port_class_mappings",
642 )
643 source_label = models.CharField(max_length=200, help_text="PortClass label as the source file states it")
644 port_class = models.CharField(max_length=50, choices=PORT_CLASS_CHOICES, verbose_name="PortClass")
646 class Meta:
647 ordering = ["profile", "source_label"]
648 constraints = [
649 models.UniqueConstraint(fields=["profile", "source_label"], name="ndi_portclassmapping_profile_label"),
650 ]
651 verbose_name = "PortClass Mapping"
652 verbose_name_plural = "PortClass Mappings"
654 def clean(self):
655 """Reject a fixed PortClass as the source label, and a target outside the fixed vocabulary."""
656 super().clean()
657 if errors := port_class_mapping_errors(self.source_label, self.port_class):
658 raise ValidationError(errors)
660 def __str__(self):
661 return f"{self.source_label} → {self.port_class}"
663 def get_absolute_url(self):
664 """Return the edit URL for this PortClass mapping."""
665 return reverse("plugins:netbox_data_import:portclassmapping_edit", args=[self.pk])
668class CableSegmentOverride(PolicySectionModel):
669 """Force the Cable policy of one planned segment, keyed by its pair of resolved terminations."""
671 POLICY_SECTION = "cable_segment_overrides"
673 profile = models.ForeignKey(
674 ImportProfile,
675 on_delete=models.CASCADE,
676 related_name="cable_segment_overrides",
677 )
678 segment_key = models.CharField(
679 max_length=200,
680 help_text="Direction-independent pair of resolved termination identities, derived by the planner",
681 )
682 cable_type = models.CharField(max_length=50, null=True, blank=True)
683 cable_profile = models.CharField(max_length=50, null=True, blank=True)
684 source_trace_identity = models.TextField(
685 help_text="Source Trace the decision was made from, so a replan can report the override it lost",
686 )
687 segment_index = models.PositiveIntegerField(
688 help_text="Segment position in that Source Trace when the decision was made",
689 )
691 # An override decides both dimensions, so it reads beside a CableClass row wherever one is read.
692 cable_type_resolved = True
693 cable_profile_resolved = True
695 class Meta:
696 ordering = ["profile", "segment_key"]
697 constraints = [
698 models.UniqueConstraint(
699 fields=["profile", "segment_key"],
700 name="ndi_cablesegmentoverride_profile_segment",
701 ),
702 ]
703 verbose_name = "Cable Segment Override"
704 verbose_name_plural = "Cable Segment Overrides"
706 def clean(self):
707 """Reject a value the running instance no longer offers or cannot terminate on one side."""
708 super().clean()
709 self.cable_type = self.cable_type or None
710 self.cable_profile = self.cable_profile or None
711 errors = policy_choice_errors(self.cable_type, self.cable_profile)
712 if errors:
713 raise ValidationError(errors)
715 def decided_policy(self):
716 """Return the Cable policy this row decides, which an override always decides completely."""
717 return {"cable_type": self.cable_type, "cable_profile": self.cable_profile}
719 def cable_type_display(self):
720 """Return the operator-facing Cable Type this override forces."""
721 return cable_type_label(self.cable_type)
723 def cable_profile_display(self):
724 """Return the operator-facing Cable Profile this override forces."""
725 return cable_profile_label(self.cable_profile)
727 def __str__(self):
728 return f"segment {self.segment_index} of {self.source_trace_identity}"
731def index_digest(value: str) -> str:
732 """Return the fixed-width index key one piece of unbounded source text is stored under.
734 PostgreSQL refuses a btree entry past about 2704 bytes, so a constraint over source text
735 carries this digest instead of the text.
736 """
737 return hashlib.sha256(value.encode("utf-8")).hexdigest()
740class DigestIndexedMixin(models.Model):
741 """Keep a source field and its fixed-width index digest synchronized."""
743 DIGEST_SOURCE_FIELD: str
744 DIGEST_FIELD: str
746 class Meta:
747 abstract = True
749 def _derive_digest(self):
750 """Derive the configured index digest from the current source text."""
751 setattr(self, self.DIGEST_FIELD, index_digest(getattr(self, self.DIGEST_SOURCE_FIELD)))
753 def save(self, *args, **kwargs):
754 """Derive the digest and include it when a partial save writes its source."""
755 update_fields = kwargs.get("update_fields")
756 if update_fields is None:
757 self._derive_digest()
758 elif self.DIGEST_SOURCE_FIELD in update_fields:
759 self._derive_digest()
760 kwargs["update_fields"] = {*update_fields, self.DIGEST_FIELD}
761 elif self.DIGEST_FIELD in update_fields:
762 raise ValueError(f"{self.DIGEST_FIELD} cannot be saved without {self.DIGEST_SOURCE_FIELD}.")
763 return super().save(*args, **kwargs)
766def _canonical_termination_field_key(value):
767 """Return *value* when it is an exact canonical termination field key."""
768 try:
769 parse_termination_field_key(value)
770 except (TypeError, ValueError) as exc:
771 raise ValidationError(
772 "Enter the canonical JSON termination field key.",
773 code="invalid",
774 ) from exc
775 return value
778class TerminationResolution(DigestIndexedMixin, PolicySectionModel):
779 """Store one selected NetBox termination for a trace field-key role."""
781 POLICY_SECTION = "termination_resolutions"
782 DIGEST_SOURCE_FIELD = "field_key"
783 DIGEST_FIELD = "field_key_digest"
785 profile = models.ForeignKey(
786 ImportProfile,
787 on_delete=models.CASCADE,
788 related_name="termination_resolutions",
789 )
790 task_type = models.CharField(
791 max_length=50,
792 choices=((SELECT_TERMINATION_TASK, "Select termination"),),
793 )
794 field_key = models.TextField()
795 field_key_digest = models.CharField(
796 max_length=64,
797 blank=True,
798 editable=False,
799 help_text="Fixed-width digest of field_key, which is what the index and constraint carry",
800 )
801 source_device = models.TextField(
802 blank=True,
803 default="",
804 help_text="Source Device spelling the decision was made for; empty on a row saved before it was kept",
805 )
806 source_cards = models.TextField(
807 blank=True,
808 default="",
809 help_text="Source cards spelling the decision was made for; empty on a row saved before it was kept",
810 )
811 source_port = models.TextField(
812 blank=True,
813 default="",
814 help_text="Source port spelling the decision was made for; empty on a row saved before it was kept",
815 )
816 selected_object_type = models.ForeignKey(
817 to="core.ObjectType",
818 on_delete=models.PROTECT,
819 related_name="+",
820 )
821 selected_object_id = models.PositiveBigIntegerField()
822 selected_display_name = models.CharField(max_length=200)
824 class Meta:
825 ordering = ["profile", "task_type", "field_key"]
826 constraints = [
827 models.UniqueConstraint(
828 fields=["profile", "task_type", "field_key_digest"],
829 name="ndi_termresolution_profile_task_key",
830 ),
831 ]
832 verbose_name = "Termination Resolution"
833 verbose_name_plural = "Termination Resolutions"
835 def clean(self):
836 """Reject an inapplicable row, a noncanonical field key, or a source spelling of another key."""
837 super().clean()
838 try:
839 _canonical_termination_field_key(self.field_key)
840 except ValidationError as exc:
841 raise ValidationError({"field_key": exc}) from exc
842 parsed = parse_termination_field_key(self.field_key)
843 for part in ("device", "cards", "port"):
844 spelling = getattr(self, f"source_{part}")
845 if (self._state.adding or spelling) and identity_text(spelling) != parsed[part]:
846 raise ValidationError({f"source_{part}": "Enter the source spelling of this field key."})
847 self._derive_digest()
849 def __str__(self):
850 return f"{self.task_type}: {self.selected_display_name}"
853class TraceDeviceResolution(DigestIndexedMixin, PolicySectionModel):
854 """Store one selected NetBox Device for a canonical source Device label."""
856 POLICY_SECTION = "trace_device_resolutions"
857 DIGEST_SOURCE_FIELD = "source_device_key"
858 DIGEST_FIELD = "source_device_key_digest"
860 profile = models.ForeignKey(
861 ImportProfile,
862 on_delete=models.CASCADE,
863 related_name="trace_device_resolutions",
864 )
865 source_device_key = models.TextField(
866 help_text="Canonical source Device label shared by every port on that Device",
867 )
868 source_device_key_digest = models.CharField(
869 max_length=64,
870 blank=True,
871 editable=False,
872 help_text="Fixed-width digest of source_device_key, which the index and constraint carry",
873 )
874 source_device_label = models.TextField(
875 blank=True,
876 default="",
877 help_text="Source Device label the decision was made for; empty on a row saved before it was kept",
878 )
879 selected_device_id = models.PositiveBigIntegerField(
880 help_text="Primary key of the selected NetBox Device",
881 )
882 selected_display_name = models.CharField(
883 max_length=200,
884 help_text="Device display name at selection time; it can become stale",
885 )
887 class Meta:
888 ordering = ["profile", "source_device_key"]
889 constraints = [
890 models.UniqueConstraint(
891 fields=["profile", "source_device_key_digest"],
892 name="ndi_tracedevresolution_profile_key",
893 ),
894 ]
895 verbose_name = "Trace Device Resolution"
896 verbose_name_plural = "Trace Device Resolutions"
898 def clean(self):
899 """Require a nonempty exact canonical key on a trace-only profile."""
900 super().clean()
901 canonical = identity_text(self.source_device_key)
902 if not canonical or canonical != self.source_device_key:
903 raise ValidationError(
904 {"source_device_key": "Enter the canonical source Device key."},
905 code="invalid",
906 )
907 if (self._state.adding or self.source_device_label) and identity_text(self.source_device_label) != canonical:
908 raise ValidationError({"source_device_label": "Enter the source label of this Device key."})
909 self._derive_digest()
911 def __str__(self):
912 return f"{self.source_device_key} → Device #{self.selected_device_id}"
915class TraceLocationResolution(DigestIndexedMixin, PolicySectionModel):
916 """Map one source Location path or prefix to one NetBox Location."""
918 POLICY_SECTION = "trace_location_resolutions"
919 DIGEST_SOURCE_FIELD = "source_location_key"
920 DIGEST_FIELD = "source_location_key_digest"
922 profile = models.ForeignKey(
923 ImportProfile,
924 on_delete=models.CASCADE,
925 related_name="trace_location_resolutions",
926 )
927 source_location_key = models.TextField(
928 help_text="Canonical key of one source Location path or prefix",
929 )
930 source_location_key_digest = models.CharField(
931 max_length=64,
932 blank=True,
933 editable=False,
934 help_text="Fixed-width digest of source_location_key, which the index and constraint carry",
935 )
936 source_location_path = models.TextField(
937 blank=True,
938 default="",
939 help_text="Source text of the path or prefix the decision was made for; empty on a row saved before it was kept",
940 )
941 selected_location_id = models.PositiveBigIntegerField(
942 help_text="Primary key of the mapped NetBox Location",
943 )
944 selected_display_name = models.CharField(
945 max_length=200,
946 help_text="Location display name at selection time; it can become stale",
947 )
949 class Meta:
950 ordering = ["profile", "source_location_key"]
951 constraints = [
952 models.UniqueConstraint(
953 fields=["profile", "source_location_key_digest"],
954 name="ndi_tracelocresolution_profile_key",
955 ),
956 ]
957 verbose_name = "Trace Location Resolution"
958 verbose_name_plural = "Trace Location Resolutions"
960 def clean(self):
961 """Require a nonempty exact canonical key on a trace-only profile."""
962 super().clean()
963 canonical = identity_text(self.source_location_key)
964 if not canonical or canonical != self.source_location_key:
965 raise ValidationError(
966 {"source_location_key": "Enter the canonical source Location key."},
967 code="invalid",
968 )
969 if (self._state.adding or self.source_location_path) and identity_text(self.source_location_path) != canonical:
970 raise ValidationError({"source_location_path": "Enter the source path of this Location key."})
971 self._derive_digest()
973 def __str__(self):
974 return f"{self.source_location_key} → Location #{self.selected_location_id}"
977class SourceDocument(models.Model):
978 """The stored uploaded workbook that a plan and its executions read.
980 Preview, replanning, a background execution, and an audit read all resolve the same bytes, so the
981 plan carries a reference instead of the content.
982 """
984 RETENTION = timedelta(days=30)
986 # Audit input outlives its profile, so a delete orphans the row and retention reclaims it.
987 profile = models.ForeignKey(
988 ImportProfile, on_delete=models.SET_NULL, null=True, blank=True, related_name="source_documents"
989 )
990 content = models.BinaryField()
991 content_fingerprint = models.CharField(max_length=64)
992 filename = models.CharField(max_length=255, blank=True)
993 uploaded_by = models.ForeignKey(
994 settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
995 )
996 created = models.DateTimeField(auto_now_add=True)
998 class Meta:
999 ordering = ["-created"]
1000 indexes = [models.Index(fields=["profile", "content_fingerprint"])]
1001 verbose_name = "Source Document"
1002 verbose_name_plural = "Source Documents"
1004 def __str__(self):
1005 return f"{self.filename or 'upload'} ({self.content_fingerprint[:12]})"
1007 @staticmethod
1008 def fingerprint(content: bytes) -> str:
1009 """Return the content fingerprint, which is what a plan compares against."""
1010 return hashlib.sha256(bytes(content)).hexdigest()
1012 @classmethod
1013 def store(cls, *, profile, content, filename="", uploaded_by=None):
1014 """Store one upload. A newer upload never removes an older one."""
1015 return cls.objects.create(
1016 profile=profile,
1017 content=bytes(content),
1018 content_fingerprint=cls.fingerprint(content),
1019 filename=filename,
1020 uploaded_by=uploaded_by,
1021 )
1023 @classmethod
1024 def purge_unreferenced(cls, *, now=None) -> int:
1025 """Delete unreferenced uploads past the retention window and return the count.
1027 A document an Import Execution references is permanent audit input, so the queryset excludes
1028 it and the protecting foreign key backs that up.
1029 """
1030 reference_time = now or timezone.now()
1031 if timezone.is_naive(reference_time):
1032 raise ValueError("now must be timezone-aware")
1033 cutoff = reference_time - cls.RETENTION
1034 # The protecting relation forces a row-by-row collect, so defer the bytes one purge would load.
1035 stale = cls.objects.filter(import_executions__isnull=True, created__lt=cutoff).defer("content")
1036 return stale.delete()[0]
1039class PreviewState:
1040 """The states of one Preview Coordinator (ADR 0004)."""
1042 EMPTY = "empty"
1043 READY = "ready"
1044 SYNC_PENDING = "sync_pending"
1045 SUBMITTED = "submitted"
1046 EXPIRED = "expired"
1048 CHOICES = (
1049 (EMPTY, "Empty"),
1050 (READY, "Ready"),
1051 (SYNC_PENDING, "Trace sync pending"),
1052 (SUBMITTED, "Import submitted"),
1053 (EXPIRED, "Expired"),
1054 )
1055 # A state in this set names a profile, a Source Document and a plan.
1056 ACTIVE = frozenset({READY, SYNC_PENDING, SUBMITTED})
1057 HOLDS_JOB = frozenset({SYNC_PENDING, SUBMITTED})
1060class PreviewCoordinator(models.Model):
1061 """One browser session's active preview: the compare-and-set row every preview command locks (ADR 0004).
1063 Only `preview_coordinator` reads or writes it. The profile and the document are scalar ids, so a
1064 delete never cascades the record away; the coordinator validates both on every access.
1065 """
1067 # A preview lives no longer than the Source Document it plans, which retention may delete.
1068 PAYLOAD_LIFETIME = SourceDocument.RETENTION
1070 session_binding = models.CharField(max_length=64, unique=True)
1071 owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="+")
1072 preview_token = models.CharField(max_length=64)
1073 revision = models.PositiveBigIntegerField(default=1)
1074 state = models.CharField(max_length=16, choices=PreviewState.CHOICES, default=PreviewState.EMPTY)
1075 profile_id = models.PositiveBigIntegerField(null=True, blank=True)
1076 source_document_id = models.PositiveBigIntegerField(null=True, blank=True)
1077 context = models.JSONField(default=dict, blank=True)
1078 plan = models.JSONField(null=True, blank=True)
1079 job_id = models.PositiveBigIntegerField(null=True, blank=True)
1080 created = models.DateTimeField(auto_now_add=True)
1081 updated = models.DateTimeField(auto_now=True)
1082 expires_at = models.DateTimeField(db_index=True)
1084 class Meta:
1085 constraints = [
1086 models.CheckConstraint(condition=models.Q(revision__gte=1), name="ndi_previewcoordinator_revision"),
1087 models.CheckConstraint(
1088 condition=models.Q(state__in=[value for value, _label in PreviewState.CHOICES]),
1089 name="ndi_previewcoordinator_state",
1090 ),
1091 models.CheckConstraint(
1092 condition=~models.Q(state__in=sorted(PreviewState.ACTIVE))
1093 | (
1094 models.Q(profile_id__isnull=False)
1095 & models.Q(source_document_id__isnull=False)
1096 & models.Q(plan__isnull=False)
1097 ),
1098 name="ndi_previewcoordinator_active_payload",
1099 ),
1100 models.CheckConstraint(
1101 condition=models.Q(state__in=sorted(PreviewState.ACTIVE))
1102 | (
1103 models.Q(profile_id__isnull=True)
1104 & models.Q(source_document_id__isnull=True)
1105 & models.Q(plan__isnull=True)
1106 & models.Q(job_id__isnull=True)
1107 ),
1108 name="ndi_previewcoordinator_inactive_payload",
1109 ),
1110 models.CheckConstraint(
1111 condition=models.Q(state__in=sorted(PreviewState.HOLDS_JOB), job_id__isnull=False)
1112 | (~models.Q(state__in=sorted(PreviewState.HOLDS_JOB)) & models.Q(job_id__isnull=True)),
1113 name="ndi_previewcoordinator_job",
1114 ),
1115 ]
1116 verbose_name = "Preview Coordinator"
1117 verbose_name_plural = "Preview Coordinators"
1119 def __str__(self):
1120 return f"Preview {self.preview_token[:8]} revision {self.revision} ({self.state})"
1123class ExecutionOutcome:
1124 """The outcome vocabulary of an Import Execution (section 9.2)."""
1126 PENDING = "pending"
1127 SUCCEEDED = "succeeded"
1128 FAILED = "failed"
1130 CHOICES = ((PENDING, "Pending"), (SUCCEEDED, "Succeeded"), (FAILED, "Failed"))
1133class FailureReason:
1134 """Typed failure reasons an Import Execution records."""
1136 ABANDONED = "abandoned"
1137 DATABASE = "database"
1138 PERMISSION = "permission"
1139 PRECONDITION = "precondition"
1140 PLANNING = "planning"
1141 SELECTION = "selection"
1142 STALE_PLAN = "stale_plan"
1143 VALIDATION = "validation"
1146class ImportExecution(models.Model):
1147 """The audit record of one selective or final execution.
1149 Rows created before the plan cutover keep their historical columns, have null new fields, and are
1150 display-only: they never satisfy an idempotency lookup and never take part in plan comparison.
1151 """
1153 #: A synchronous attempt cannot outlive the web request bound, so an older pending row is gone.
1154 SYNCHRONOUS_BOUND = timedelta(minutes=10)
1156 objects = RestrictedQuerySet.as_manager()
1158 profile = models.ForeignKey(
1159 ImportProfile,
1160 on_delete=models.SET_NULL,
1161 null=True,
1162 blank=True,
1163 related_name="import_executions",
1164 )
1165 created = models.DateTimeField(auto_now_add=True)
1166 input_filename = models.CharField(max_length=255, blank=True)
1167 site_name = models.CharField(max_length=100, blank=True)
1168 result_counts = models.JSONField(default=dict)
1170 source_document = models.ForeignKey(
1171 SourceDocument,
1172 on_delete=models.PROTECT,
1173 null=True,
1174 blank=True,
1175 related_name="import_executions",
1176 )
1177 actor = models.ForeignKey(
1178 settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
1179 )
1180 idempotency_key = models.CharField(max_length=64, null=True, blank=True)
1181 plan_schema_version = models.PositiveIntegerField(null=True, blank=True)
1182 accepted_plan_fingerprint = models.CharField(max_length=64, null=True, blank=True)
1183 selected_units = models.JSONField(null=True, blank=True)
1184 outcome = models.CharField(max_length=16, choices=ExecutionOutcome.CHOICES, null=True, blank=True)
1185 applied_changes = models.JSONField(null=True, blank=True)
1186 failure_detail = models.JSONField(null=True, blank=True)
1187 # Set by link_job: the reservation commits before the Job exists, and outlives a deleted Job.
1188 job_backed = models.BooleanField(default=False)
1189 job = models.OneToOneField(
1190 "core.Job", on_delete=models.SET_NULL, null=True, blank=True, related_name="import_execution"
1191 )
1193 class Meta:
1194 ordering = ["-created"]
1195 constraints = [
1196 models.UniqueConstraint(
1197 fields=["profile", "idempotency_key"],
1198 condition=models.Q(idempotency_key__isnull=False),
1199 name="ndi_execution_profile_idempotency_key",
1200 ),
1201 ]
1202 verbose_name = "Import Execution"
1203 verbose_name_plural = "Import Executions"
1205 def __str__(self):
1206 return f"Import {self.pk} — {self.created:%Y-%m-%d %H:%M} ({self.input_filename})"
1208 def get_absolute_url(self):
1209 """Return the associated profile's URL (no per-execution detail view exists)."""
1210 if not self.profile_id:
1211 return reverse("plugins:netbox_data_import:importprofile_list")
1212 return reverse("plugins:netbox_data_import:importprofile", args=[self.profile_id])
1214 @classmethod
1215 def reserve(cls, *, committed_by_caller=False, **fields):
1216 """Insert and commit the pending row, or return the row already holding this key.
1218 The insert reserves the unique (Import Profile, idempotency key), so a duplicate submission
1219 or job delivery loses the race and returns the existing row in any outcome.
1220 `committed_by_caller` is for a caller whose transaction commits the failed row, with the
1221 target writes in a savepoint below it.
1222 """
1223 if transaction.get_connection().in_atomic_block != committed_by_caller:
1224 raise RuntimeError("The Import Execution reservation must commit before the target transaction opens.")
1225 if not fields.get("idempotency_key"):
1226 raise ValueError("An Import Execution reservation requires an idempotency key.")
1227 # PostgreSQL treats two NULL profiles as distinct, so the partial unique index cannot hold.
1228 if not fields.get("profile"):
1229 raise ValueError("An Import Execution reservation requires an Import Profile.")
1230 required_audit_fields = {
1231 "source_document": "a Source Document",
1232 "actor": "an actor",
1233 "plan_schema_version": "a plan schema version",
1234 "accepted_plan_fingerprint": "an accepted plan fingerprint",
1235 "selected_units": "selected Synchronization Unit identities",
1236 }
1237 for field_name, label in required_audit_fields.items():
1238 if fields.get(field_name) is None:
1239 raise ValueError(f"An Import Execution reservation requires {label}.")
1240 existing = cls.for_idempotency(fields["profile"], fields["idempotency_key"])
1241 if existing is not None:
1242 return existing, False
1243 try:
1244 # A savepoint, so a lost race leaves a caller's transaction usable.
1245 with transaction.atomic():
1246 # atomic-exit-safe: reservation-inserted
1247 return cls.objects.create(outcome=ExecutionOutcome.PENDING, **fields), True
1248 except IntegrityError:
1249 # Only a lost race for this key is recoverable; any other constraint failure must surface.
1250 winner = cls.for_idempotency(fields["profile"], fields["idempotency_key"])
1251 if winner is None:
1252 raise
1253 return winner, False
1255 def link_job(self, job):
1256 """Record the native Job that runs this execution, after the reservation has committed."""
1257 self.job = job
1258 self.job_backed = True
1259 self.save(update_fields=["job", "job_backed"])
1260 return self
1262 @classmethod
1263 def for_idempotency(cls, profile, idempotency_key):
1264 """Return the reserved row for this key, reconciled, or None. A legacy row never matches."""
1265 if not idempotency_key:
1266 return None
1267 found = cls.objects.filter(profile=profile, idempotency_key=idempotency_key).first()
1268 return found.reconcile_pending() if found is not None else None
1270 def reconcile_pending(self, *, now=None):
1271 """Transition an abandoned pending row to failed, so no sweeper is needed."""
1272 if self.outcome != ExecutionOutcome.PENDING:
1273 return self
1274 if self.job_backed:
1275 # The Job decides once it exists; a deleted Job leaves no way to finish the attempt.
1276 job = Job.objects.filter(pk=self.job_id).first() if self.job_id else None
1277 live = job is not None and job.status in JobStatusChoices.ENQUEUED_STATE_CHOICES
1278 else:
1279 # Either a synchronous attempt or a background one still between reserving and enqueuing.
1280 live = self.created > (now or timezone.now()) - self.SYNCHRONOUS_BOUND
1281 if live:
1282 return self
1283 # A worker finished the row between this read and the transition; its outcome wins.
1284 with suppress(ValueError):
1285 self.mark_failed(reason=FailureReason.ABANDONED)
1286 return self
1288 def _finish(self, **values):
1289 """Transition this row out of pending exactly once, with the database as the arbiter.
1291 Two instances can both hold a pending copy, so an in-memory check would let the second
1292 write overwrite a committed outcome and destroy the audit evidence.
1293 """
1294 updated = type(self).objects.filter(pk=self.pk, outcome=ExecutionOutcome.PENDING).update(**values)
1295 if not updated:
1296 self.refresh_from_db()
1297 raise ValueError(f"Import Execution {self.pk} already finished as '{self.outcome}'.")
1298 for name, value in values.items():
1299 setattr(self, name, value)
1300 return self
1302 def mark_succeeded(self, *, applied_changes, result_counts=None):
1303 """Record the applied identities and the deleted-object snapshot."""
1304 return self._finish(
1305 outcome=ExecutionOutcome.SUCCEEDED,
1306 applied_changes=applied_changes,
1307 failure_detail=None,
1308 result_counts=result_counts or {},
1309 )
1311 def mark_failed(self, *, reason, failed_change=None, rolled_back=(), not_attempted=()):
1312 """Record what failed, what rolled back, and what was never attempted."""
1313 return self._finish(
1314 outcome=ExecutionOutcome.FAILED,
1315 applied_changes=None,
1316 result_counts={"created": {}, "errors": 1},
1317 failure_detail={
1318 "failed_change": failed_change,
1319 "rolled_back": list(rolled_back),
1320 "not_attempted": list(not_attempted),
1321 "reason": reason,
1322 },
1323 )
1326class DeviceTypeMapping(PolicySectionModel):
1327 """Explicit (make, model) override when source naming doesn't slugify cleanly."""
1329 POLICY_SECTION = "device_type_mappings"
1331 profile = models.ForeignKey(
1332 ImportProfile,
1333 on_delete=models.CASCADE,
1334 related_name="device_type_mappings",
1335 )
1336 source_make = models.CharField(max_length=200)
1337 source_model = models.CharField(max_length=200)
1338 netbox_manufacturer_slug = models.CharField(max_length=100)
1339 netbox_device_type_slug = models.CharField(max_length=100)
1341 class Meta:
1342 ordering = ["profile", "source_make", "source_model"]
1343 constraints = [
1344 models.UniqueConstraint(
1345 fields=["profile", "source_make", "source_model"], name="ndi_dtm_profile_make_model"
1346 ),
1347 ]
1348 verbose_name = "Device Type Mapping"
1349 verbose_name_plural = "Device Type Mappings"
1351 def __str__(self):
1352 return (
1353 f"{self.source_make} / {self.source_model} → {self.netbox_manufacturer_slug}/{self.netbox_device_type_slug}"
1354 )
1356 def get_absolute_url(self):
1357 """Return the edit URL for this device type mapping."""
1358 return reverse("plugins:netbox_data_import:devicetypemapping_edit", args=[self.pk])
1361class ManufacturerMapping(PolicySectionModel):
1362 """Maps a source 'make' value to an existing NetBox manufacturer slug."""
1364 POLICY_SECTION = "manufacturer_mappings"
1366 profile = models.ForeignKey(
1367 ImportProfile,
1368 on_delete=models.CASCADE,
1369 related_name="manufacturer_mappings",
1370 )
1371 source_make = models.CharField(
1372 max_length=200,
1373 help_text="Exact source make value (e.g. 'Dell EMC')",
1374 )
1375 netbox_manufacturer_slug = models.CharField(
1376 max_length=100,
1377 help_text="NetBox manufacturer slug to map this make to (e.g. 'dell')",
1378 )
1380 class Meta:
1381 ordering = ["profile", "source_make"]
1382 constraints = [
1383 models.UniqueConstraint(fields=["profile", "source_make"], name="ndi_mfgmapping_profile_make"),
1384 ]
1385 verbose_name = "Manufacturer Mapping"
1386 verbose_name_plural = "Manufacturer Mappings"
1388 def __str__(self):
1389 return f"{self.source_make} → {self.netbox_manufacturer_slug}"
1392class InferenceBackend(NetBoxModel):
1393 """One named Inference Backend definition; the enabled row is the active backend (section 8.2)."""
1395 ADAPTER_TYPES = _inference_settings.ADAPTER_TYPES
1396 AUTHENTICATION_METHODS = _inference_settings.AUTHENTICATION_METHODS
1397 RESPONSE_MODES = _inference_settings.RESPONSE_MODES
1399 backend_key = models.SlugField(
1400 max_length=100,
1401 unique=True,
1402 help_text="The unique name used in operator-visible results and backend metadata.",
1403 )
1404 display_name = models.CharField(max_length=200)
1405 adapter_type = models.CharField(max_length=50, choices=ADAPTER_TYPES, default="openai_compatible")
1406 api_root = models.CharField(
1407 max_length=500,
1408 help_text="Exact API root without a trailing slash. The client appends /chat/completions.",
1409 )
1410 model = models.CharField(max_length=200, help_text="Exact backend model id. The worker never chooses one.")
1411 authentication = models.CharField(max_length=20, choices=AUTHENTICATION_METHODS, default="bearer")
1412 response_mode = models.CharField(
1413 max_length=20,
1414 choices=RESPONSE_MODES,
1415 default="prompt_json",
1416 help_text="Select a mode other than prompt_json only after verifying the exact backend and model.",
1417 )
1418 credential_reference = models.JSONField(help_text="A typed Vault KV v2 reference. It never holds a secret value.")
1419 # A zero timeout raises in the transport, so the row carries the same floor as the fallback.
1420 connect_timeout = models.PositiveIntegerField(
1421 default=5, validators=[MinValueValidator(_inference_settings.TIMEOUT_MIN)]
1422 )
1423 read_timeout = models.PositiveIntegerField(
1424 default=60, validators=[MinValueValidator(_inference_settings.TIMEOUT_MIN)]
1425 )
1426 max_concurrent_requests = models.PositiveIntegerField(
1427 default=_inference_settings.MAX_CONCURRENT_REQUESTS_DEFAULT,
1428 validators=[MinValueValidator(_inference_settings.MAX_CONCURRENT_REQUESTS_MIN)],
1429 help_text="How many Resolution Proposals may call this backend at the same time. The others wait.",
1430 )
1431 enabled = models.BooleanField(default=False, help_text="Whether Ask AI may use this backend.")
1433 class Meta:
1434 ordering = ["backend_key"]
1435 constraints = [
1436 # A partial unique index over one column value permits exactly one enabled row.
1437 models.UniqueConstraint(
1438 fields=["enabled"],
1439 condition=models.Q(enabled=True),
1440 name="ndi_inferencebackend_one_enabled",
1441 ),
1442 ]
1443 verbose_name = "AI backend"
1444 verbose_name_plural = "AI backends"
1446 def __str__(self):
1447 return self.display_name or self.backend_key
1449 def get_absolute_url(self):
1450 """Return the detail URL for this Inference Backend."""
1451 return reverse("plugins:netbox_data_import:inferencebackend", args=[self.pk])
1453 def serialize_object(self, exclude=None):
1454 """Keep the Vault reference out of NetBox change-log snapshots."""
1455 return super().serialize_object(exclude=[*(exclude or ()), "credential_reference"])
1457 def clean(self):
1458 """Reject a second enabled row, an unapproved api_root, and a reference that is not typed."""
1459 super().clean()
1460 from .inference_backend import validate_backend_fields
1462 if self.enabled:
1463 competing = type(self).objects.filter(enabled=True).exclude(pk=self.pk)
1464 if competing.exists():
1465 raise ValidationError({"enabled": "Another Inference Backend is already enabled. Disable it first."})
1466 validate_backend_fields(
1467 api_root=self.api_root,
1468 authentication=self.authentication,
1469 credential_reference=self.credential_reference,
1470 )
1473class ProposalStatus:
1474 """The five Resolution Proposal statuses and their edges (section 7.2)."""
1476 QUEUED = "queued"
1477 RUNNING = "running"
1478 COMPLETED = "completed"
1479 FAILED = "failed"
1480 CANCELLED = "cancelled"
1482 CHOICES = (
1483 (QUEUED, "Queued"),
1484 (RUNNING, "Running"),
1485 (COMPLETED, "Completed"),
1486 (FAILED, "Failed"),
1487 (CANCELLED, "Cancelled"),
1488 )
1489 ACTIVE = (QUEUED, RUNNING)
1490 TERMINAL = (COMPLETED, FAILED, CANCELLED)
1492 #: The only permitted edges. Every other transition is refused, and a retry is a new row.
1493 EDGES = {
1494 QUEUED: (RUNNING, CANCELLED, FAILED),
1495 RUNNING: (COMPLETED, CANCELLED, FAILED),
1496 COMPLETED: (),
1497 FAILED: (),
1498 CANCELLED: (),
1499 }
1502class ProposalOutcome:
1503 """What a completed Resolution Proposal concluded (section 7.8)."""
1505 CANDIDATE = OUTCOME_CANDIDATE
1506 NO_MATCH = OUTCOME_NO_MATCH
1508 CHOICES = OUTCOME_CHOICES
1511class ProposalDecision:
1512 """The one-shot operator decision, which is not a status (section 7.2)."""
1514 ACCEPTED = "accepted"
1515 REJECTED = "rejected"
1517 CHOICES = ((ACCEPTED, "Accepted"), (REJECTED, "Rejected"))
1520class ProposalFailureReason:
1521 """Typed failure reasons a Resolution Proposal stores (section 13.3).
1523 These are persisted values with their own vocabulary. The job maps adapter categories onto them
1524 explicitly, so renaming an adapter category cannot silently change what is already stored.
1525 """
1527 BACKEND_REFUSAL = "backend_refusal"
1528 INVALID_RESPONSE = "invalid_response"
1529 RATE_LIMIT = "rate_limit"
1530 TEMPORARY_BACKEND_FAILURE = "temporary_backend_failure"
1531 QUEUE_UNAVAILABLE = "queue_unavailable"
1532 TIMEOUT = "timeout"
1533 AUTHENTICATION_FAILURE = "authentication_failure"
1534 INVALID_CONFIGURATION = "invalid_configuration"
1535 CREDENTIAL_UNAVAILABLE = "credential_unavailable"
1536 CREDENTIAL_DENIED = "credential_denied"
1537 CREDENTIAL_INVALID = "credential_invalid"
1538 SUPERSEDED_REQUEST = "superseded_request"
1539 BRANCH_ACTIVE = "branch_active"
1541 CHOICES = (
1542 (BACKEND_REFUSAL, "Backend refusal"),
1543 (INVALID_RESPONSE, "Invalid backend response"),
1544 (RATE_LIMIT, "Rate limited"),
1545 (TEMPORARY_BACKEND_FAILURE, "Temporary backend failure"),
1546 (QUEUE_UNAVAILABLE, "Queue unavailable"),
1547 (TIMEOUT, "Timeout"),
1548 (AUTHENTICATION_FAILURE, "Backend authentication failure"),
1549 (INVALID_CONFIGURATION, "Invalid configuration"),
1550 (CREDENTIAL_UNAVAILABLE, "Credential infrastructure unavailable"),
1551 (CREDENTIAL_DENIED, "Credential denied"),
1552 (CREDENTIAL_INVALID, "Invalid credential reference or secret"),
1553 (SUPERSEDED_REQUEST, "Superseded by a newer request contract"),
1554 (BRANCH_ACTIVE, "Refused inside a netbox-branching branch"),
1555 )
1557 #: Section 7.5: these retry at most twice inside the same proposal; every other reason fails at once.
1558 TRANSIENT = (RATE_LIMIT, TEMPORARY_BACKEND_FAILURE, TIMEOUT, CREDENTIAL_UNAVAILABLE)
1561class ResolutionProposal(DigestIndexedMixin, models.Model):
1562 """One Resolution Proposal request, attempt, and operator decision (section 7).
1564 A plain audit model on purpose: it carries no profile policy and must stay out of the profile
1565 YAML policy export that `PolicySectionModel` subclasses enter.
1566 """
1568 DIGEST_SOURCE_FIELD = "field_key"
1569 DIGEST_FIELD = "field_key_digest"
1571 # NetBox's generic views scope a queryset with `restrict()`, which only this manager provides.
1572 objects = RestrictedQuerySet.as_manager()
1574 profile = models.ForeignKey(
1575 ImportProfile,
1576 on_delete=models.CASCADE,
1577 related_name="resolution_proposals",
1578 )
1579 task_type = models.CharField(
1580 max_length=50,
1581 choices=((SELECT_TERMINATION_TASK, "Select termination"),),
1582 )
1583 field_key = models.TextField()
1584 field_key_digest = models.CharField(
1585 max_length=64,
1586 blank=True,
1587 editable=False,
1588 help_text="Fixed-width digest of field_key, which is what the index and constraint carry",
1589 )
1590 status = models.CharField(max_length=20, choices=ProposalStatus.CHOICES, default=ProposalStatus.QUEUED)
1592 source_evidence = models.JSONField(help_text="The source values for the bound field, frozen at request time.")
1593 resolved_device_type = models.ForeignKey(to="core.ObjectType", on_delete=models.PROTECT, related_name="+")
1594 resolved_device_id = models.PositiveBigIntegerField()
1595 prompt_version = models.PositiveIntegerField()
1596 response_schema_version = models.PositiveIntegerField()
1597 candidate_snapshot = models.JSONField(help_text="The eligible candidate set as it stood at request time.")
1598 requested_by = models.ForeignKey(
1599 settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
1600 )
1601 created = models.DateTimeField(auto_now_add=True)
1602 last_updated = models.DateTimeField(auto_now=True)
1604 outcome = models.CharField(max_length=20, choices=ProposalOutcome.CHOICES, blank=True, default="")
1605 selected_candidate_id = models.CharField(max_length=100, blank=True, default="")
1606 selected_object_type = models.ForeignKey(
1607 to="core.ObjectType", on_delete=models.PROTECT, null=True, blank=True, related_name="+"
1608 )
1609 selected_object_id = models.PositiveBigIntegerField(null=True, blank=True)
1610 explanation = models.TextField(blank=True, default="")
1611 backend_metadata = models.JSONField(null=True, blank=True)
1612 # The adapter's whole diagnostic, so absent, empty and interrupted receipts stay distinguishable.
1613 response_diagnostic = models.JSONField(null=True, blank=True)
1615 failure_reason = models.CharField(max_length=50, choices=ProposalFailureReason.CHOICES, blank=True, default="")
1616 # Null while the row waits for a request slot; the dispatcher sets it when it enqueues the Job.
1617 job = models.OneToOneField(
1618 "core.Job", on_delete=models.SET_NULL, null=True, blank=True, related_name="resolution_proposal"
1619 )
1621 decision = models.CharField(max_length=20, choices=ProposalDecision.CHOICES, blank=True, default="")
1622 decided_by = models.ForeignKey(
1623 settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
1624 )
1625 decided_at = models.DateTimeField(null=True, blank=True)
1626 written_resolution = models.ForeignKey(
1627 TerminationResolution, on_delete=models.SET_NULL, null=True, blank=True, related_name="+"
1628 )
1630 class Meta:
1631 ordering = ["-created"]
1632 constraints = [
1633 models.UniqueConstraint(
1634 fields=["profile", "task_type", "field_key_digest"],
1635 condition=models.Q(status__in=ProposalStatus.ACTIVE),
1636 name="ndi_resolutionproposal_one_active",
1637 ),
1638 models.CheckConstraint(
1639 condition=(models.Q(status=ProposalStatus.COMPLETED) & ~models.Q(outcome=""))
1640 | (~models.Q(status=ProposalStatus.COMPLETED) & models.Q(outcome="")),
1641 name="ndi_resolutionproposal_completed_iff_outcome",
1642 ),
1643 models.CheckConstraint(
1644 condition=(models.Q(status=ProposalStatus.FAILED) & ~models.Q(failure_reason=""))
1645 | (~models.Q(status=ProposalStatus.FAILED) & models.Q(failure_reason="")),
1646 name="ndi_resolutionproposal_failed_iff_reason",
1647 ),
1648 models.CheckConstraint(
1649 condition=~models.Q(outcome=ProposalOutcome.CANDIDATE)
1650 | (
1651 models.Q(selected_object_type__isnull=False, selected_object_id__isnull=False)
1652 & ~models.Q(selected_candidate_id="")
1653 ),
1654 name="ndi_resolutionproposal_candidate_has_selection",
1655 ),
1656 models.CheckConstraint(
1657 condition=~models.Q(outcome=ProposalOutcome.NO_MATCH)
1658 | models.Q(
1659 selected_object_type__isnull=True, selected_object_id__isnull=True, selected_candidate_id=""
1660 ),
1661 name="ndi_resolutionproposal_no_match_has_no_selection",
1662 ),
1663 models.CheckConstraint(
1664 condition=models.Q(decision="", decided_at__isnull=True)
1665 | (~models.Q(decision="") & models.Q(decided_at__isnull=False)),
1666 name="ndi_resolutionproposal_decision_group",
1667 ),
1668 models.CheckConstraint(
1669 condition=models.Q(decision="") | models.Q(status=ProposalStatus.COMPLETED),
1670 name="ndi_resolutionproposal_decision_needs_completed",
1671 ),
1672 ]
1673 verbose_name = "Resolution Proposal"
1674 verbose_name_plural = "Resolution Proposals"
1676 def clean(self):
1677 """Reject a noncanonical field key, and derive the index key before validate_unique reads it."""
1678 super().clean()
1679 try:
1680 _canonical_termination_field_key(self.field_key)
1681 except ValidationError as exc:
1682 raise ValidationError({"field_key": exc}) from exc
1683 self._derive_digest()
1685 def __str__(self):
1686 return f"{self.task_type}: {self.status}"
1689class IgnoredDevice(PolicySectionModel):
1690 """Per-device ignore record — prevents a specific source device from being imported."""
1692 POLICY_SECTION = "ignored_devices"
1694 profile = models.ForeignKey(
1695 ImportProfile,
1696 on_delete=models.CASCADE,
1697 related_name="ignored_devices",
1698 )
1699 source_id = models.CharField(
1700 max_length=200,
1701 help_text="Source ID value that identifies this device",
1702 )
1703 device_name = models.CharField(
1704 max_length=200,
1705 blank=True,
1706 help_text="Original device name (for display only)",
1707 )
1709 class Meta:
1710 ordering = ["profile", "source_id"]
1711 constraints = [
1712 models.UniqueConstraint(fields=["profile", "source_id"], name="ndi_ignoreddevice_profile_srcid"),
1713 ]
1714 verbose_name = "Ignored Device"
1715 verbose_name_plural = "Ignored Devices"
1717 def __str__(self):
1718 return f"{self.device_name or self.source_id} (ignored)"
1721class ColumnTransformRule(PolicySectionModel):
1722 r"""Regex-based transform applied to a source column during parse.
1724 Example: source_column='Name', pattern='^(\w{4,8}) - (.+)$',
1725 group_1_target='asset_tag', group_2_target='device_name'
1726 transforms "TEST0001 - EXAMPLE-SWITCH-01" into asset_tag="TEST0001", device_name="EXAMPLE-SWITCH-01".
1727 """
1729 POLICY_SECTION = "column_transform_rules"
1731 profile = models.ForeignKey(
1732 ImportProfile,
1733 on_delete=models.CASCADE,
1734 related_name="column_transform_rules",
1735 )
1736 source_column = models.CharField(
1737 max_length=200,
1738 help_text="Source Excel column to transform (exact header name)",
1739 )
1740 pattern = models.CharField(
1741 max_length=500,
1742 help_text=(
1743 r"RE2 pattern with capture groups and full-match semantics. "
1744 r"Backreferences and look-around are not supported. E.g. ^(\w+) - (.+)$"
1745 ),
1746 )
1747 group_1_target = models.CharField(
1748 max_length=100,
1749 blank=True,
1750 help_text="Target field for capture group 1 (leave blank to ignore)",
1751 )
1752 group_2_target = models.CharField(
1753 max_length=100,
1754 blank=True,
1755 help_text="Target field for capture group 2 (leave blank to ignore)",
1756 )
1758 class Meta:
1759 ordering = ["profile", "source_column"]
1760 constraints = [
1761 models.UniqueConstraint(fields=["profile", "source_column"], name="ndi_ctr_profile_column"),
1762 ]
1763 verbose_name = "Column Transform Rule"
1764 verbose_name_plural = "Column Transform Rules"
1766 def clean(self):
1767 """Validate the regex, capture groups, and exclusive ownership of group targets."""
1768 from django.core.exceptions import ValidationError
1770 from .transform_regex import TransformPattern, TransformPatternError
1772 super().clean()
1774 try:
1775 compiled = TransformPattern.compile(self.pattern)
1776 except TransformPatternError as exc:
1777 raise ValidationError({"pattern": f"Regex pattern is not supported: {exc}"}) from exc
1779 required_groups = 0
1780 if self.group_1_target:
1781 required_groups = 1
1782 if self.group_2_target:
1783 required_groups = 2
1784 if compiled.group_count < required_groups:
1785 raise ValidationError(
1786 {
1787 "pattern": (
1788 f"Regex must contain at least {required_groups} capture group(s) "
1789 f"for the configured group target(s), but found {compiled.group_count}."
1790 )
1791 }
1792 )
1794 output_kinds = self.profile.output_kinds if self.profile_id else None
1795 target_attrs = {}
1796 for attr in ("group_1_target", "group_2_target"):
1797 value = getattr(self, attr) or ""
1798 if not value:
1799 continue
1800 # A capture group yields text, so a candidate target is not a valid group target.
1801 if not CATALOG.is_valid(value, output_kinds=output_kinds, allow_candidates=False):
1802 raise ValidationError({attr: CATALOG.invalid_key_message(value)})
1803 previous_attr = target_attrs.get(value)
1804 if previous_attr is not None:
1805 previous_group = previous_attr.removeprefix("group_").removesuffix("_target")
1806 raise ValidationError(
1807 {attr: f"Target field '{value}' is already assigned to capture group {previous_group}."}
1808 )
1809 target_attrs[value] = attr
1811 if not self.profile_id or not target_attrs:
1812 return
1813 conflicts = (
1814 type(self)
1815 .objects.filter(profile_id=self.profile_id)
1816 .exclude(pk=self.pk)
1817 .filter(models.Q(group_1_target__in=target_attrs) | models.Q(group_2_target__in=target_attrs))
1818 .only("source_column", "group_1_target", "group_2_target")
1819 )
1820 errors = {}
1821 for conflict in conflicts:
1822 for target in (conflict.group_1_target, conflict.group_2_target):
1823 attr = target_attrs.get(target)
1824 if attr is not None:
1825 errors[attr] = (
1826 f"Target field '{target}' is already assigned by the transform rule "
1827 f"for source column '{conflict.source_column}'."
1828 )
1829 errors.update(self._column_mapping_target_errors(target_attrs))
1830 if errors:
1831 raise ValidationError(errors)
1833 def _column_mapping_target_errors(self, target_attrs):
1834 """Return capture errors for targets already owned by direct mappings."""
1835 mapped_targets = {
1836 mapping.target_field: mapping.source_column
1837 for mapping in ColumnMapping.objects.filter(
1838 profile_id=self.profile_id,
1839 target_field__in=target_attrs,
1840 ).only("source_column", "target_field")
1841 }
1842 return {
1843 attr: (
1844 f"Target field '{target}' is already assigned by the column mapping "
1845 f"for source column '{mapped_targets[target]}'."
1846 )
1847 for target, attr in target_attrs.items()
1848 if target in mapped_targets
1849 }
1851 def __str__(self):
1852 return f"{self.source_column}: {self.pattern}"
1854 def get_absolute_url(self):
1855 """Return the edit URL for this column transform rule."""
1856 return reverse("plugins:netbox_data_import:columntransformrule_edit", args=[self.pk])
1859class SourceResolution(PolicySectionModel):
1860 """Saved target-field decision for one source row.
1862 A resolution can split one source value or select candidate source columns
1863 for structured target fields. The import reapplies it when the same source
1864 row appears in a later file.
1865 """
1867 POLICY_SECTION = "source_resolutions"
1869 profile = models.ForeignKey(
1870 ImportProfile,
1871 on_delete=models.CASCADE,
1872 related_name="source_resolutions",
1873 )
1874 source_id = models.CharField(
1875 max_length=200,
1876 help_text="Source ID of the row this resolution applies to",
1877 )
1878 source_column = models.CharField(
1879 max_length=200,
1880 help_text="Column name this resolution applies to",
1881 )
1882 original_value = models.TextField(
1883 help_text="Original cell value before resolution",
1884 )
1885 resolved_fields = models.JSONField(
1886 default=dict,
1887 help_text="Dict of target_field -> resolved_value (e.g. {'device_name': 'SW1', 'asset_tag': 'TEST0001'})",
1888 )
1890 class Meta:
1891 ordering = ["profile", "source_id"]
1892 constraints = [
1893 models.UniqueConstraint(
1894 fields=["profile", "source_id", "source_column"], name="ndi_srcresolution_profile_id_col"
1895 ),
1896 ]
1897 verbose_name = "Source Resolution"
1898 verbose_name_plural = "Source Resolutions"
1900 def clean(self):
1901 """Require target-field decisions that can safely merge into the profile's source rows."""
1902 super().clean()
1903 validate_source_resolution_fields(
1904 self.profile if self.profile_id else None,
1905 self.source_column,
1906 self.resolved_fields,
1907 )
1909 def __str__(self):
1910 return f"{self.source_id}/{self.source_column}: {self.original_value!r}"
1913class DeviceExistingMatch(PolicySectionModel):
1914 """Explicit match between a source row and an existing NetBox device.
1916 When a user clicks "Link existing" on a device preview row, this record is saved.
1917 On re-import, the engine uses this to emit action='update' against the matched device
1918 instead of action='create', even if the device has no source-ID custom field yet.
1919 """
1921 POLICY_SECTION = "device_existing_matches"
1923 profile = models.ForeignKey(
1924 ImportProfile,
1925 on_delete=models.CASCADE,
1926 related_name="device_matches",
1927 )
1928 source_id = models.CharField(
1929 max_length=200,
1930 help_text="Source ID value that identifies this row",
1931 )
1932 source_asset_tag = models.CharField(
1933 max_length=100,
1934 blank=True,
1935 default="",
1936 help_text="Asset tag from source row (for display / lookup; may become stale)",
1937 )
1938 netbox_device_id = models.PositiveIntegerField(
1939 help_text="Primary key of the matched NetBox Device",
1940 )
1941 device_name = models.CharField(
1942 max_length=200,
1943 blank=True,
1944 help_text="NetBox device name (for display only; may become stale)",
1945 )
1947 class Meta:
1948 ordering = ["profile", "source_id"]
1949 constraints = [
1950 models.UniqueConstraint(fields=["profile", "source_id"], name="ndi_devicematch_profile_srcid"),
1951 models.UniqueConstraint(
1952 fields=["profile", "netbox_device_id"],
1953 name="ndi_devicematch_profile_device",
1954 ),
1955 ]
1956 verbose_name = "Device Existing Match"
1957 verbose_name_plural = "Device Existing Matches"
1959 def __str__(self):
1960 tag = f" / {self.source_asset_tag}" if self.source_asset_tag else ""
1961 return f"{self.source_id}{tag} → Device #{self.netbox_device_id} ({self.device_name})"
1964class IgnoredFieldDifference(PolicySectionModel):
1965 """Preserve one exact file/NetBox value pair for a device field difference."""
1967 POLICY_SECTION = "ignored_field_differences"
1969 profile = models.ForeignKey(
1970 ImportProfile,
1971 on_delete=models.CASCADE,
1972 related_name="ignored_field_differences",
1973 )
1974 source_id = models.CharField(
1975 max_length=200,
1976 help_text="Source ID of the row this review applies to",
1977 )
1978 netbox_device_id = models.PositiveIntegerField(
1979 help_text="Primary key of the matched NetBox Device",
1980 )
1981 target_field = models.CharField(
1982 max_length=100,
1983 help_text="Target field whose current difference is ignored",
1984 )
1985 file_snapshot = models.JSONField(
1986 default=dict,
1987 help_text="Normalized and display values from the source row",
1988 )
1989 netbox_snapshot = models.JSONField(
1990 default=dict,
1991 help_text="Normalized and display values from the matched NetBox device",
1992 )
1994 class Meta:
1995 ordering = ["profile", "source_id", "target_field"]
1996 constraints = [
1997 models.UniqueConstraint(
1998 fields=["profile", "source_id", "netbox_device_id", "target_field"],
1999 name="ndi_ignored_diff_profile_source_device_field",
2000 ),
2001 ]
2002 verbose_name = "Ignored Field Difference"
2003 verbose_name_plural = "Ignored Field Differences"
2005 def __str__(self):
2006 return f"{self.source_id}/{self.target_field} on device #{self.netbox_device_id} (ignored)"
2009class DeviceImportSource(models.Model):
2010 """Import provenance the plugin keeps for one Device.
2012 Replaces the plugin-managed ``data_import_source`` custom field. The per-profile custom
2013 field an operator configures (``ImportProfile.custom_field_name``) is separate and stays.
2014 """
2016 device = models.OneToOneField(
2017 to="dcim.Device",
2018 on_delete=models.CASCADE,
2019 related_name="data_import_source",
2020 )
2021 profile = models.ForeignKey(
2022 ImportProfile,
2023 on_delete=models.CASCADE,
2024 related_name="device_sources",
2025 )
2026 source_id = models.CharField(
2027 max_length=200,
2028 blank=True,
2029 default="",
2030 help_text="Source ID of the row that wrote this device",
2031 )
2032 extra_columns = models.JSONField(
2033 default=dict,
2034 blank=True,
2035 help_text="Source column values that no mapping consumes",
2036 )
2037 unassigned_ips = models.JSONField(
2038 default=dict,
2039 blank=True,
2040 help_text="IP values the import could not assign to a NetBox IP field",
2041 )
2043 class Meta:
2044 ordering = ["device"]
2045 indexes = [models.Index(fields=["profile", "source_id"])]
2046 verbose_name = "Device Import Source"
2047 verbose_name_plural = "Device Import Sources"
2049 def __str__(self):
2050 return f"{self.source_id or '(no source ID)'} → Device #{self.device_id}"
2053class CableImportSource(DigestIndexedMixin):
2054 """Import provenance the plugin keeps for one Cable and one contributing Source Trace.
2056 Two Source Traces that state one identical segment share one created Cable, so the Cable
2057 reference is a plain foreign key and the row is keyed by the trace as well (section 5.7).
2058 """
2060 DIGEST_SOURCE_FIELD = "trace_identity"
2061 DIGEST_FIELD = "trace_key"
2063 objects = RestrictedQuerySet.as_manager()
2065 cable = models.ForeignKey(
2066 to="dcim.Cable",
2067 on_delete=models.CASCADE,
2068 related_name="data_import_sources",
2069 )
2070 profile = models.ForeignKey(
2071 ImportProfile,
2072 on_delete=models.CASCADE,
2073 related_name="cable_sources",
2074 )
2075 trace_identity = models.TextField(
2076 help_text="Canonical JSON identity of the Source Trace that states this segment",
2077 )
2078 trace_key = models.CharField(
2079 max_length=64,
2080 blank=True,
2081 editable=False,
2082 help_text="Fixed-width digest of trace_identity, which is what the index and constraint carry",
2083 )
2084 segment_index = models.PositiveIntegerField(
2085 null=True,
2086 blank=True,
2087 help_text=(
2088 "Position of this segment in the Source Trace, in canonical order; empty when an identity "
2089 "upgrade reversed that order and the trace length was unknown"
2090 ),
2091 )
2092 from_text = models.TextField(blank=True, default="")
2093 to_text = models.TextField(blank=True, default="")
2094 direction = models.CharField(max_length=20, blank=True, default="")
2095 workbook_fingerprint = models.CharField(max_length=64, blank=True, default="")
2096 sheet = models.CharField(max_length=100, blank=True, default="")
2097 block_ordinal = models.PositiveIntegerField(null=True, blank=True)
2098 row_start = models.PositiveIntegerField(null=True, blank=True)
2099 row_end = models.PositiveIntegerField(null=True, blank=True)
2100 export_timestamp = models.CharField(
2101 max_length=TRACE_EXPORT_TIMESTAMP_MAX_LENGTH,
2102 blank=True,
2103 default="",
2104 )
2106 def clean(self):
2107 """Derive the index key before validate_unique reads the constraint's own fields."""
2108 super().clean()
2109 self._derive_digest()
2111 class Meta:
2112 ordering = ["cable", "profile", "trace_identity"]
2113 constraints = [
2114 models.UniqueConstraint(
2115 fields=["cable", "profile", "trace_key"],
2116 name="ndi_cableimportsource_cable_profile_trace",
2117 ),
2118 ]
2119 indexes = [models.Index(fields=["profile", "trace_key"])]
2120 verbose_name = "Cable Import Source"
2121 verbose_name_plural = "Cable Import Sources"
2123 def __str__(self):
2124 return f"segment {self.segment_index} of {self.from_text} to {self.to_text} \u2192 Cable #{self.cable_id}"
2127def stored_import_source(obj):
2128 """Return the plugin's import record for one object, or None when it holds none."""
2129 from dcim.models import Device
2131 if not isinstance(obj, Device) or obj.pk is None:
2132 return None
2133 return DeviceImportSource.objects.filter(device_id=obj.pk).first()