Coverage for netbox_data_import/target_modules.py: 98.07%
1137 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# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""The Rack, Device and Cable Target Modules, and the registry the coordinator resolves them through.
5Section 2.3 gives a Target Module target-specific matching, ORM queries, permission checks,
6preconditions, locking and writes. It plans against the complete relevant Source Batch and applies
7one Planned Change at a time. It never commits a transaction and never calls another module.
9`catalog.TargetModule` stays the static declaration a profile derives its Target Fields from. This
10module is the behaviour behind that declaration.
11"""
13from __future__ import annotations
15import datetime as dt
16import math
17import re
18from copy import copy
19from dataclasses import dataclass
20from decimal import Decimal
21from typing import Any, Protocol
23from django.core.exceptions import ValidationError
25from . import ip_assignment
26from .cable_target import CableModule
27from .catalog import OutputKind
28from .contact_resolution import (
29 ContactResolutionRequired,
30 ContactReview,
31 ContactSelection,
32 DanglingProfileReference,
33 PrimaryContactResolver,
34)
35from .device_field_review import DeviceFieldReviewer
36from .device_identity import DEVICE_TYPE_AMBIGUOUS, MANUFACTURER_AMBIGUOUS, DeviceTypeIdentityResolver
37from .identity import identity_expression, identity_in, identity_text
38from .netbox_reader import PlanningTargetUnavailable
39from .object_permissions import ObjectPermissionDenied, ProspectiveRelation, assess_permission_scoped_save
40from .plan import Diagnostic, Disposition, PlannedChange, Severity, SynchronizationUnit
41from .target_runtime import ExecutionContext, PreconditionFailed, TargetModuleRuntime
42from .values import (
43 effective_device_name,
44 normalize_for_compare,
45 source_position,
46 source_text,
47 translation_maps,
48)
50DEFAULT_RACK_HEIGHT = 42
51# Two mappings that share one name identity and name different targets leave the row for the operator.
52_AMBIGUOUS_MAPPING_CODES = {
53 DEVICE_TYPE_AMBIGUOUS: "device.device_type_mapping_ambiguous",
54 MANUFACTURER_AMBIGUOUS: "device.manufacturer_mapping_ambiguous",
55}
58class _RackComparison(Protocol):
59 """The Rack state that planning compares before it schedules a write."""
61 u_height: int
62 serial: str
63 rack_type_id: int | None
64 location_id: int | None
65 tenant_id: int | None
68def _candidate_save_is_allowed(actor, candidate, prospective_relations=None) -> bool:
69 """Return whether one candidate stays inside the actor's add or change scope."""
70 primary_key = candidate._meta.pk.attname
71 values = {
72 field.attname: field.value_from_object(candidate)
73 for field in candidate._meta.concrete_fields
74 if not field.primary_key
75 }
76 return assess_permission_scoped_save(
77 actor,
78 type(candidate),
79 {primary_key: candidate.pk},
80 values,
81 prospective_relations=prospective_relations,
82 ).allowed
85def _prospective_ip_relations(candidate, ip_fields) -> dict[str, ProspectiveRelation]:
86 """Return final placeable address rows and connect them during permission assessment."""
87 relations = {}
88 for field, address in ip_assignment.prospective_addresses(candidate, ip_fields).items():
89 setattr(candidate, f"{field}_id", None)
90 relations[field] = ProspectiveRelation(address.instance, address.generated_fields)
91 return relations
94def _planned_device_relations(payload, dependencies) -> dict[str, Any]:
95 """Return only planned relations that the reviewed Device payload still uses."""
96 relations = {}
97 planned_rack = dependencies.planned_rack
98 if planned_rack is not None and (
99 (planned_rack.pk is None and payload["rack_id"] is None and payload["rack_name"])
100 or (planned_rack.pk is not None and payload["rack_id"] == planned_rack.pk)
101 ):
102 relations["rack"] = planned_rack
103 if payload["role_id"] is None and dependencies.role.pk is None:
104 relations["role"] = dependencies.role
105 return relations
108def _text(value) -> str:
109 """Return stripped stored or target text, empty for None."""
110 return "" if value is None else str(value).strip()
113def _source_text(value) -> str:
114 """Return source text, including an empty value for spreadsheet null markers."""
115 return source_text(value)
118def _duplicate_value_detail(label: str, value: str, other_rows: list[int]) -> str:
119 """Name a duplicated identity value and every other source row that carries it."""
120 where = ", ".join(f"row {number}" for number in other_rows)
121 return f"Duplicate {label} '{value}' appears more than once in this import" + (
122 f", also on {where}." if where else "."
123 )
126def _display_value(value):
127 """Return detached JSON display data for a source value."""
128 if value is None or isinstance(value, (str, int, bool)):
129 return value
130 if isinstance(value, float):
131 return value if math.isfinite(value) else str(value)
132 if isinstance(value, Decimal):
133 if not value.is_finite():
134 return str(value)
135 return int(value) if value == value.to_integral_value() else float(value)
136 if isinstance(value, (dt.date, dt.datetime, dt.time)):
137 return value.isoformat()
138 if isinstance(value, dict):
139 return {str(key): _display_value(item) for key, item in value.items()}
140 if isinstance(value, (list, tuple)):
141 return [_display_value(item) for item in value]
142 return str(value)
145def _unit_display(row, object_type: str, name: str, rack_name: str = "") -> dict:
146 """Return source and presentation facts shared by all unit dispositions."""
147 source_row = {str(key): _display_value(value) for key, value in row.items()}
148 return {
149 "row_number": row.get("_row_number"),
150 "source_id": _source_text(row.get("source_id")),
151 "name": name,
152 "rack_name": rack_name,
153 "source_row": source_row,
154 "extra_data": {
155 "asset_tag": _source_text(row.get("asset_tag"))[:50],
156 "candidate_values": _display_value(row.get("_candidate_values") or {}),
157 "conflicts": _display_value(row.get("_conflicts") or {}),
158 "extra_columns": _display_value(row.get("_extra_columns") or {}),
159 "source_class": _source_text(row.get("device_class")),
160 "source_make": _source_text(row.get("make")),
161 "source_model": _source_text(row.get("model")),
162 "source_serial": _source_text(row.get("serial")),
163 },
164 "object_type": object_type,
165 }
168def _class_mapping_display(mapping) -> dict:
169 """Return the stored class policy, so its editor reopens on what the operator saved."""
170 if mapping is None:
171 return {"class_mapping_action": "", "class_mapping_role_slug": ""}
172 if mapping.ignore:
173 action = "ignore"
174 elif mapping.creates_rack:
175 action = "rack"
176 else:
177 action = "role"
178 return {"class_mapping_action": action, "class_mapping_role_slug": _text(mapping.role_slug)}
181def _ignored_source_ids(profile) -> frozenset[str]:
182 """Return the source identities the operator has chosen to skip."""
183 return frozenset(_source_text(value) for value in profile.ignored_devices.values_list("source_id", flat=True))
186def _repeated(values) -> frozenset[str]:
187 """Return the non-empty values that appear more than once in one batch."""
188 seen: set[str] = set()
189 repeated: set[str] = set()
190 for value in values:
191 if not value:
192 continue
193 if value in seen:
194 repeated.add(value)
195 seen.add(value)
196 return frozenset(repeated)
199def _translate(value, table) -> str:
200 """Return the NetBox value a source word names, or the word itself when it is already one."""
201 text = _source_text(value).lower()
202 if not text:
203 return ""
204 mapped = table.get(text)
205 if mapped is not None:
206 return mapped
207 return text if text in set(table.values()) else ""
210def _coerce_rack_height(value) -> int:
211 """Return the rack height the row asks for, never below one unit."""
212 try:
213 return max(1, int(float(_source_text(value) or DEFAULT_RACK_HEIGHT)))
214 except (OverflowError, TypeError, ValueError):
215 return DEFAULT_RACK_HEIGHT
218def rack_row_name(row) -> str:
219 """Return the rack name one rack row carries."""
220 return _source_text(row.get("rack_name")) or _source_text(row.get("device_name"))
223def rack_unit_identity(row) -> str:
224 """Return the stable Synchronization Unit identity for one rack row."""
225 source_id = _source_text(row.get("source_id"))
226 if source_id:
227 return f"rack:source:{source_id}"
228 return f"rack:name:{identity_text(rack_row_name(row))}"
231def rack_duplicate_keys(rows) -> tuple[frozenset[str], frozenset[str]]:
232 """Return the rack names and source IDs more than one row in this batch claims."""
233 return (
234 _repeated(identity_text(rack_row_name(row)) for row in rows),
235 _repeated(_source_text(row.get("source_id")) for row in rows),
236 )
239def rack_row_rejection(row, ignored, duplicate_names, duplicate_source_ids) -> tuple[str, dict] | None:
240 """Keep the rack-pass rejection order, including its empty-ID ignore match, for cutover parity."""
241 name = rack_row_name(row)
242 source_id = _source_text(row.get("source_id"))
243 if not name:
244 return "rack.missing_name", {"source_id": source_id}
245 if source_id in ignored:
246 return "rack.ignored", {"rack_name": name, "source_id": source_id}
247 if identity_text(name) in duplicate_names:
248 return "rack.duplicate_name", {"rack_name": name, "source_id": source_id}
249 if source_id and source_id in duplicate_source_ids:
250 return "rack.duplicate_source_id", {"rack_name": name, "source_id": source_id}
251 return None
254def _racks_by_comparison_name(netbox_reader) -> dict[str, list[Any]]:
255 """Return visible racks at the exact import target, grouped by comparison name."""
256 if netbox_reader.site is None:
257 return {}
258 location_filter = (
259 {"location": netbox_reader.location} if netbox_reader.location is not None else {"location__isnull": True}
260 )
261 grouped: dict[str, list[Any]] = {}
262 for rack in netbox_reader.racks().filter(site=netbox_reader.site, **location_filter):
263 grouped.setdefault(identity_text(rack.name), []).append(rack)
264 return grouped
267class RackModule:
268 """Plans and applies the Racks a flat source batch describes."""
270 key = "rack"
271 consumes = frozenset({OutputKind.RACK_SOURCE_ROW})
273 def plan(
274 self,
275 source_batch,
276 profile,
277 catalog,
278 netbox_reader,
279 *,
280 lock_plan_references: bool = False,
281 ) -> list[SynchronizationUnit]:
282 """Return one Synchronization Unit per rack row, with the disposition its state earns."""
283 # Planned Change preconditions carry every target row this module depends on, so no read-only reference remains.
284 del lock_plan_references
285 rows = self._rack_rows(source_batch, profile)
286 if not rows:
287 return []
288 ignored = _ignored_source_ids(profile)
289 duplicate_names, duplicate_source_ids = rack_duplicate_keys(rows)
290 existing = _racks_by_comparison_name(netbox_reader)
291 mappings = {mapping.source_class: mapping for mapping in profile.class_role_mappings.all()}
292 return [
293 self._unit(
294 row,
295 profile,
296 mappings[_source_text(row.get("device_class"))],
297 netbox_reader,
298 ignored,
299 duplicate_names,
300 duplicate_source_ids,
301 existing,
302 )
303 for row in rows
304 ]
306 @staticmethod
307 def _rack_rows(source_batch, profile) -> list[dict]:
308 """Return the batch rows whose class the profile maps to a rack."""
309 creates_rack = {
310 mapping.source_class
311 for mapping in profile.class_role_mappings.all()
312 if mapping.creates_rack and not mapping.ignore
313 }
314 return [row for row in source_batch.rows if _source_text(row.get("device_class")) in creates_rack]
316 @staticmethod
317 def unit_identity(row) -> str:
318 """Return the identity that survives replanning, which is never the row number."""
319 return rack_unit_identity(row)
321 def _unit(
322 self,
323 row,
324 profile,
325 mapping,
326 netbox_reader,
327 ignored,
328 duplicate_names,
329 duplicate_source_ids,
330 existing,
331 ) -> SynchronizationUnit:
332 """Return the one unit this row produces."""
333 identity = self.unit_identity(row)
334 name = rack_row_name(row)
335 source_id = _source_text(row.get("source_id"))
336 if identity_text(name) in duplicate_names or (source_id and source_id in duplicate_source_ids):
337 identity = f"{identity}:row:{row.get('_row_number')}"
338 unit_display = _unit_display(row, self.key, name, name)
339 unit_display["extra_data"].update(
340 {
341 "rack_type_id": mapping.rack_type_id or "",
342 "rack_type_name": str(mapping.rack_type) if mapping.rack_type_id else "",
343 "rack_type_set": bool(mapping.rack_type_id),
344 }
345 )
347 rejection = rack_row_rejection(row, ignored, duplicate_names, duplicate_source_ids)
348 if rejection is not None:
349 code, display = rejection
350 display = {**unit_display, **display}
351 if code == "rack.ignored":
352 return SynchronizationUnit(
353 identity=identity,
354 disposition=Disposition.EXCLUDED,
355 diagnostics=(
356 Diagnostic(code=code, severity=Severity.INFO, identities=(identity,), display=display),
357 ),
358 display=unit_display,
359 )
360 return _refused(identity, code, display)
362 height = _coerce_rack_height(row.get("u_height"))
363 serial = _source_text(row.get("serial"))
364 rack_type_id = mapping.rack_type_id
365 matches = existing.get(identity_text(name), ())
366 if len(matches) > 1:
367 return _refused(identity, "rack.ambiguous_name", unit_display)
368 rack = matches[0] if matches else None
369 if rack is None:
370 actor = netbox_reader.actor
371 candidate, validation = self._validated_candidate(
372 None,
373 name,
374 height,
375 serial,
376 rack_type_id,
377 netbox_reader,
378 profile.adapter_settings.custom_field_name,
379 source_id=_source_text(row.get("source_id")),
380 )
381 if validation is not None:
382 return _refused(identity, "rack.validation_failed", {**unit_display, "message": validation})
383 if actor is not None and not _candidate_save_is_allowed(actor, candidate):
384 return _refused(
385 identity,
386 "rack.add_permission",
387 unit_display,
388 disposition=Disposition.BLOCKED,
389 )
390 return SynchronizationUnit(
391 identity=identity,
392 disposition=Disposition.ACTIONABLE,
393 changes=(
394 self._change(
395 identity,
396 "create",
397 name,
398 height,
399 serial,
400 rack_type_id,
401 netbox_reader,
402 None,
403 _source_text(row.get("source_id")),
404 ),
405 ),
406 display=unit_display,
407 )
409 existing_display = {
410 **unit_display,
411 "netbox_url": rack.get_absolute_url(),
412 "extra_data": {
413 **unit_display["extra_data"],
414 "netbox_rack_id": rack.pk,
415 },
416 }
417 update_existing = profile.adapter_settings.update_existing
418 if not update_existing:
419 existing_display["detail"] = f"Rack '{name}' already exists (update_existing=False)"
420 return SynchronizationUnit(identity=identity, disposition=Disposition.NO_OP, display=existing_display)
421 candidate, validation = self._validated_candidate(
422 rack,
423 name,
424 height,
425 serial,
426 rack_type_id,
427 netbox_reader,
428 profile.adapter_settings.custom_field_name,
429 source_id=_source_text(row.get("source_id")),
430 )
431 if not self._differs(rack, candidate):
432 existing_display["extra_data"]["writes_nothing"] = True
433 existing_display["detail"] = f"Rack '{name}' already exists and this row changes nothing"
434 return SynchronizationUnit(identity=identity, disposition=Disposition.NO_OP, display=existing_display)
435 if validation is not None:
436 return _refused(identity, "rack.validation_failed", {**existing_display, "message": validation})
437 if netbox_reader.actor is not None and not _candidate_save_is_allowed(netbox_reader.actor, candidate):
438 return _refused(
439 identity,
440 "rack.change_permission",
441 existing_display,
442 disposition=Disposition.BLOCKED,
443 )
444 return SynchronizationUnit(
445 identity=identity,
446 disposition=Disposition.ACTIONABLE,
447 changes=(
448 self._change(
449 identity,
450 "update",
451 name,
452 height,
453 serial,
454 rack_type_id,
455 netbox_reader,
456 rack,
457 _source_text(row.get("source_id")),
458 ),
459 ),
460 display=existing_display,
461 )
463 def apply(self, planned_change: PlannedChange, execution_context) -> Any:
464 """Apply one rack change, having locked its row and rechecked its preconditions."""
465 from dcim.models import Rack
467 from .object_permissions import enforce_saved_object_permission
469 payload = planned_change.payload
470 rack_id = planned_change.preconditions.get("rack_id")
471 if rack_id is None:
472 existing = (
473 Rack.objects.filter(site_id=payload["site_id"], location_id=payload["location_id"])
474 .filter(identity_in("name", [identity_text(payload["name"])]))
475 .first()
476 )
477 if existing is not None:
478 raise PreconditionFailed(f"Rack '{payload['name']}' appeared after the plan was made.")
479 rack = Rack(site_id=payload["site_id"], location_id=payload["location_id"])
480 action = "add"
481 else:
482 # `of=("self",)` because NetBox's default Rack queryset outer-joins, which cannot be locked.
483 rack = Rack.objects.filter(pk=rack_id).select_for_update(of=("self",)).first()
484 if rack is None:
485 raise PreconditionFailed(f"Rack {rack_id} is gone, so '{payload['name']}' cannot be updated.")
486 current = self._precondition_state(rack)
487 expected = planned_change.preconditions.get("state")
488 if current != expected:
489 raise PreconditionFailed(f"Rack '{rack.name}' changed after the plan was made.")
490 rack.snapshot()
491 action = "change"
493 rack.name = payload["name"]
494 rack.u_height = payload["u_height"]
495 if payload["serial"]:
496 rack.serial = payload["serial"]
497 rack.rack_type_id = payload["rack_type_id"]
498 rack.copy_racktype_attrs()
499 if payload["location_id"] is not None:
500 rack.location_id = payload["location_id"]
501 if payload["tenant_id"] is not None:
502 rack.tenant_id = payload["tenant_id"]
503 custom_field = execution_context.profile.adapter_settings.custom_field_name
504 if action == "add" and custom_field and payload["source_id"]:
505 rack.custom_field_data[custom_field] = payload["source_id"]
506 rack.full_clean()
507 rack.save()
508 # An ObjectPermission's constraints are only evaluated against the saved row.
509 enforce_saved_object_permission(rack, execution_context.actor, action)
510 return rack
512 @staticmethod
513 def _differs(rack: _RackComparison, candidate: _RackComparison) -> bool:
514 """Return whether the stored Rack differs from the state the write will leave."""
515 if normalize_for_compare(rack.u_height) != normalize_for_compare(candidate.u_height):
516 return True
517 if rack.rack_type_id != candidate.rack_type_id:
518 return True
519 if rack.location_id != candidate.location_id:
520 return True
521 if rack.tenant_id != candidate.tenant_id:
522 return True
523 return _text(rack.serial) != _text(candidate.serial)
525 @staticmethod
526 def _change(
527 identity, operation, name, height, serial, rack_type_id, netbox_reader, rack, source_id
528 ) -> PlannedChange:
529 """Return the one write this unit performs, with the target state it assumed."""
530 payload = {
531 "name": name,
532 "u_height": height,
533 "serial": serial,
534 "rack_type_id": rack_type_id,
535 "source_id": source_id,
536 "site_id": netbox_reader.site.pk if netbox_reader.site is not None else None,
537 "location_id": netbox_reader.location.pk if netbox_reader.location is not None else None,
538 "tenant_id": netbox_reader.tenant.pk if netbox_reader.tenant is not None else None,
539 }
540 preconditions = (
541 {"rack_id": rack.pk, "state": RackModule._precondition_state(rack)}
542 if rack is not None
543 else {"rack_id": None}
544 )
545 return PlannedChange(
546 identity=f"{identity}:{operation}",
547 target_module=RackModule.key,
548 operation=operation,
549 payload=payload,
550 preconditions=preconditions,
551 )
553 @staticmethod
554 def _precondition_state(rack) -> dict:
555 """Return every rack field this module can overwrite."""
556 return {
557 "name": rack.name,
558 "u_height": rack.u_height,
559 "serial": rack.serial,
560 "rack_type_id": rack.rack_type_id,
561 "location_id": rack.location_id,
562 "tenant_id": rack.tenant_id,
563 }
565 @staticmethod
566 def _validated_candidate(rack, name, height, serial, rack_type_id, reader, custom_field, source_id):
567 """Return the prospective Rack and its model validation message, if any."""
568 from dcim.models import Rack
570 candidate = copy(rack) if rack is not None else Rack(site=reader.site, location=reader.location)
571 candidate.name = name
572 candidate.u_height = height
573 if serial:
574 candidate.serial = serial
575 candidate.rack_type_id = rack_type_id
576 candidate.copy_racktype_attrs()
577 if reader.location is not None:
578 candidate.location = reader.location
579 if reader.tenant is not None:
580 candidate.tenant = reader.tenant
581 if rack is None and custom_field and source_id:
582 candidate.custom_field_data[custom_field] = source_id
583 try:
584 candidate.full_clean()
585 except ValidationError as exc:
586 return candidate, "; ".join(exc.messages)
587 return candidate, None
590def _occupied_units(position, height):
591 """Return the half-unit slots a device of *height* fills from *position*."""
592 from decimal import Decimal
594 start = Decimal(str(position))
595 step = Decimal("0.5")
596 count = int(Decimal(str(height)) / step)
597 return [start + step * index for index in range(count)]
600def _refused(identity, code, display, *, disposition=Disposition.INVALID) -> SynchronizationUnit:
601 """Return a non-actionable unit carrying the error that refused it."""
602 return SynchronizationUnit(
603 identity=identity,
604 disposition=disposition,
605 diagnostics=(Diagnostic(code=code, severity=Severity.ERROR, identities=(identity,), display=display),),
606 display=display,
607 )
610def _with_issues(identity, issues) -> SynchronizationUnit:
611 """Return one unit carrying every problem a row has, in the order the checks found them.
613 The first problem decides the disposition and the display, so a unit reads exactly as it did
614 when only that problem was reported. The rest tell the operator what this row still needs.
615 """
616 disposition, _first_code, first_display = issues[0]
617 return SynchronizationUnit(
618 identity=identity,
619 disposition=disposition,
620 diagnostics=tuple(
621 Diagnostic(code=code, severity=Severity.ERROR, identities=(identity,), display=issue_display)
622 for _disposition, code, issue_display in issues
623 ),
624 display=first_display,
625 )
628@dataclass(frozen=True)
629class _Dependencies:
630 """What a device row needs to already exist, or the first thing that does not."""
632 device_type: Any = None
633 role: Any = None
634 rack: Any = None
635 rack_identity: str | None = None
636 planned_rack: Any = None
637 rack_change_identity: str | None = None
638 unavailable_rack_name: str = ""
639 role_slug: str = ""
640 explicit_device_type: bool = False
641 changes: tuple[PlannedChange, ...] = ()
642 missing: tuple[str, dict] | None = None
645@dataclass(frozen=True)
646class _Placement:
647 """Where a device row puts the device, or the reason it cannot go there."""
649 position: Any = None
650 face: str = ""
651 airflow: str = ""
652 status: str = "active"
653 refused: tuple[str, dict] | None = None
656@dataclass(frozen=True)
657class _PlacementClaim:
658 """The rack units one settled device row can reserve."""
660 keys: tuple[tuple[int | str, str | None, Any], ...] = ()
661 refused: tuple[str, dict] | None = None
664@dataclass(frozen=True)
665class _Match:
666 """The stored device a row reconciles, or the reason no automatic answer is safe."""
668 device: Any = None
669 ambiguous: str | None = None
670 value: str = ""
671 method: str = ""
672 inaccessible: bool = False
675@dataclass(frozen=True)
676class _PlannedRack:
677 """One Rack row's final state and whether its write can run."""
679 identity: str
680 operation: str
681 candidate: Any
682 executable: bool
684 @property
685 def change_identity(self) -> str:
686 """Return the exact Planned Change identity for this Rack write."""
687 return f"{self.identity}:{self.operation}"
690_REVIEWED_PAYLOAD_FIELDS: dict[str, tuple[str, Any]] = {
691 "serial": ("serial", lambda device: _text(device.serial)),
692 "asset_tag": ("asset_tag", lambda device: _text(device.asset_tag)),
693 "u_position": ("u_position", lambda device: source_position(device.position)),
694 "face": ("face", lambda device: _text(device.face)),
695 "airflow": ("airflow", lambda device: _text(device.airflow)),
696 "status": ("status", lambda device: _text(device.status)),
697 "device_type": ("device_type_id", lambda device: device.device_type_id),
698 "role": ("role_id", lambda device: device.role_id),
699 "rack_name": ("rack_id", lambda device: device.rack_id),
700 "tenant": ("tenant_id", lambda device: device.tenant_id),
701 "location": ("location_id", lambda device: device.location_id),
702}
705def _reviewed_payload(payload, review, device) -> dict:
706 """Return the payload with every ignored field back at the value NetBox holds.
708 The disposition and the write both read this one result, so a row whose only difference the
709 operator ignored plans as a no-op and can never be written as an update.
710 """
711 if review is None or not review.ignored:
712 return payload
713 effective = dict(payload)
714 effective["ip_fields"] = {
715 name: value for name, value in (payload.get("ip_fields") or {}).items() if name not in review.ignored
716 }
717 for target_field in review.ignored:
718 mapped = _REVIEWED_PAYLOAD_FIELDS.get(target_field)
719 if mapped is None:
720 continue
721 key, stored = mapped
722 effective[key] = stored(device)
723 if target_field == "rack_name":
724 effective["rack_name"] = None
725 return effective
728def _snapshot_saved_device(device) -> None:
729 """Snapshot the stored row of a device this write already saved, for its next change log entry."""
730 # NetBox moves counter fields such as interface_count in the database only, so the instance can be behind.
731 device.refresh_from_db()
732 device.snapshot()
735def _assign_ips(device, ip_fields, actor) -> dict:
736 """Assign each placeable address and return the fields that remain unassigned."""
737 unassigned = {}
738 changed = {}
739 for field, address in ip_fields.items():
740 try:
741 target = ip_assignment.resolve(device, field, address)
742 except ip_assignment.IPAssignmentError:
743 unassigned[field] = address
744 continue
745 if target.already_held:
746 if getattr(device, f"{field}_id", None) != target.held.pk:
747 changed[field] = target.held
748 continue
749 changed[field] = ip_assignment.apply(target, actor)
750 if changed:
751 _snapshot_saved_device(device)
752 for field, address in changed.items():
753 setattr(device, field, address)
754 device.save(update_fields=sorted(changed))
755 return unassigned
758def _apply_contact(device, contact, execution_context) -> None:
759 """Assign the reviewed primary contact to a device the write has already saved."""
760 if contact is None:
761 return
762 PrimaryContactResolver.apply(
763 device,
764 execution_context.profile,
765 ContactReview(
766 selection=ContactSelection(values=dict(contact["values"]), contact_id=contact["contact_id"]),
767 extra_columns={},
768 plan=None,
769 candidate_values={},
770 suggestion=None,
771 ),
772 execution_context.actor,
773 )
776def _provenance_is_current(device, payload, profile) -> bool:
777 """Return whether the stored provenance already records what this row would write.
779 A device that holds every field but no record is still work: the record is what lets the next
780 import reconcile this device instead of creating a second one beside it.
781 """
782 from .models import DeviceImportSource
784 source_id = payload.get("source_id") or ""
785 if (
786 source_id
787 and not profile.device_matches.filter(
788 source_id=source_id,
789 netbox_device_id=device.pk,
790 device_name=device.name,
791 source_asset_tag=payload.get("asset_tag") or "",
792 ).exists()
793 ):
794 return False
795 custom_field = profile.adapter_settings.custom_field_name
796 if custom_field and source_id and device.custom_field_data.get(custom_field) != source_id:
797 return False
798 stored = DeviceImportSource.objects.filter(device_id=device.pk).first()
799 return stored is not None and (
800 stored.profile_id == profile.pk
801 and stored.source_id == source_id
802 and stored.extra_columns == (payload.get("extra_columns") or {})
803 and not stored.unassigned_ips
804 )
807def _bind_source(profile, source_id, device, asset_tag) -> None:
808 """Bind this source row to this device, refusing a binding that already names another."""
809 existing = profile.device_matches.select_for_update().filter(source_id=source_id).first()
810 if existing is None:
811 profile.device_matches.create(
812 source_id=source_id,
813 netbox_device_id=device.pk,
814 device_name=device.name,
815 source_asset_tag=asset_tag,
816 )
817 return
818 if existing.netbox_device_id != device.pk:
819 raise PreconditionFailed(
820 f"Source ID '{source_id}' is bound to device #{existing.netbox_device_id}, not '{device.name}'."
821 )
822 existing.device_name = device.name
823 existing.source_asset_tag = asset_tag
824 existing.save(update_fields=["device_name", "source_asset_tag"])
827def _store_provenance(device, payload, unassigned, execution_context) -> None:
828 """Record which source row wrote this device, so a later import reconciles it instead of copying it."""
829 from .models import DeviceImportSource
831 profile = execution_context.profile
832 source_id = payload.get("source_id") or ""
833 asset_tag = payload.get("asset_tag") or ""
834 if source_id:
835 _bind_source(profile, source_id, device, asset_tag)
836 custom_field = profile.adapter_settings.custom_field_name
837 if custom_field and device.custom_field_data.get(custom_field) != source_id:
838 _snapshot_saved_device(device)
839 device.custom_field_data[custom_field] = source_id
840 device.save(update_fields=["custom_field_data"])
841 DeviceImportSource.objects.update_or_create(
842 device=device,
843 defaults={
844 "profile": profile,
845 "source_id": source_id,
846 "extra_columns": payload.get("extra_columns") or {},
847 "unassigned_ips": unassigned,
848 },
849 )
852def _contact_payload(review) -> dict | None:
853 """Return the reviewed contact selection in the form the plan carries and the write replays."""
854 if review is None or review.selection is None:
855 return None
856 return {"values": dict(review.selection.values), "contact_id": review.selection.contact_id}
859def _contact_writes_nothing(review) -> bool:
860 """Return whether the reviewed contact leaves the stored assignment as it stands."""
861 plan = None if review is None else review.plan
862 if plan is None:
863 return True
864 return plan["contact_action"] == "reuse" and plan["assignment_action"] == "unchanged"
867class _DeviceBatch:
868 """The batch-wide state every device row is planned against, loaded once."""
870 _CLASH_FIELDS = (
871 ("source_id", "device.duplicate_source_id", False),
872 ("serial", "device.duplicate_serial", False),
873 ("asset_tag", "device.duplicate_asset_tag", True),
874 )
876 def __init__(self, source_batch, rows, profile, netbox_reader, *, lock_plan_references: bool = False):
877 from dcim.models import Device
879 self.profile = profile
880 self.reader = netbox_reader
881 self.lock_plan_references = lock_plan_references
882 self._locked_placement: dict[tuple[type, int], Any] = {}
883 self.ignored = _ignored_source_ids(profile)
884 self._identity = DeviceTypeIdentityResolver.for_profile(profile)
885 self._reviewer = DeviceFieldReviewer.for_profile(profile)
886 self._candidate_columns = PrimaryContactResolver.candidate_source_columns(profile)
887 self._mappings = {mapping.source_class: mapping for mapping in profile.class_role_mappings.all()}
888 self._roles = {source_class: _text(mapping.role_slug) for source_class, mapping in self._mappings.items()}
889 self._device_types, self._role_objects = self._load_dependency_objects(rows)
890 self._bindings = {_text(match.source_id): match.netbox_device_id for match in profile.device_matches.all()}
891 self._bound_sources = {match.netbox_device_id: _text(match.source_id) for match in profile.device_matches.all()}
892 identity_rows = [row for row in rows if self._is_identity_writing_row(row)]
893 identity_source_ids = {
894 _source_text(row.get("source_id")) for row in identity_rows if _source_text(row.get("source_id"))
895 }
896 self._review_device_ids = {
897 source_id: self._reviewer.review_device_ids(source_id) for source_id in identity_source_ids
898 }
899 reviewed_device_ids = {
900 self._bindings[source_id] for source_id in identity_source_ids if source_id in self._bindings
901 }
902 reviewed_device_ids.update(
903 next(iter(device_ids)) for device_ids in self._review_device_ids.values() if len(device_ids) == 1
904 )
905 self._reviewed_devices = {
906 device.pk: device
907 for device in Device.objects.select_related(
908 "device_type__manufacturer",
909 "rack__location",
910 "role",
911 "tenant",
912 "location",
913 "site",
914 ).filter(pk__in=reviewed_device_ids)
915 }
916 self._visible_reviewed_device_ids = (
917 frozenset(reviewed_device_ids)
918 if self.reader.actor is None
919 else frozenset(self.reader.devices().filter(pk__in=reviewed_device_ids).values_list("pk", flat=True))
920 )
921 (
922 self._devices_by_source_id,
923 self._devices_by_serial,
924 self._devices_by_asset_tag,
925 self._devices_by_name,
926 self._visible_identity_device_ids,
927 ) = self._load_identity_objects(identity_rows)
928 self._duplicate_names = _repeated(identity_text(effective_device_name(row)) for row in identity_rows)
929 self._reserved_names = {
930 identity_text(name)
931 for name in netbox_reader.devices()
932 .filter(
933 site=netbox_reader.site,
934 **({"tenant": netbox_reader.tenant} if netbox_reader.tenant is not None else {"tenant__isnull": True}),
935 )
936 .values_list("name", flat=True)
937 if name is not None
938 }
939 self._reserved_names.update(identity_text(effective_device_name(row)) for row in identity_rows)
940 self._effective_identity = self._effective_identity_values(identity_rows)
941 self._clashes = {
942 "source_id": self._rows_by_value(source_batch.rows, "source_id", False),
943 "serial": self._rows_by_effective("serial", fold=False),
944 "asset_tag": self._rows_by_effective("asset_tag", fold=True),
945 }
946 self._racks = _racks_by_comparison_name(netbox_reader)
947 self._planned_racks = self._planned_racks_by_name(source_batch, profile)
948 self._lock_placement_references(rows)
949 self.side_map, self.airflow_map, self.status_map = translation_maps()
950 # Row order decides who keeps a slot two rows claim, so the first row planned wins it.
951 self._claimed: dict[tuple[int | str, str | None, Any], int] = {}
952 self._claimed_devices: dict[int, tuple[int | None, str]] = {}
954 def placement_reference(self, model, pk):
955 """Return one row the placement reads, from the locks this batch took before planning."""
956 if self.lock_plan_references:
957 return self._locked_placement.get((model, pk))
958 return model.objects.filter(pk=pk).first()
960 def _lock_placement_references(self, rows: list[dict[str, Any]]) -> None:
961 """Lock every placement row this batch can read, in one model and primary-key order.
963 Two profiles plan concurrently, because the policy lock covers one profile each. A lock
964 taken per row therefore follows source order, and two batches can take the same rows in
965 opposite orders. One ordered pass per model cannot.
966 """
967 if not self.lock_plan_references:
968 return
969 from dcim.models import DeviceType, Rack
971 candidates = self._placement_candidate_devices()
972 # u_height and is_full_depth decide placement here, and Device.full_clean() reads them again.
973 type_ids = {device_type.pk for device_type in self._device_types.values()}
974 type_ids.update(device.device_type_id for device in candidates)
975 rack_ids = {rack.pk for rack in self._named_rack_matches(rows)}
976 rack_ids.update(device.rack_id for device in candidates if device.rack_id)
977 locked_types = (
978 DeviceType.objects.select_related("manufacturer")
979 .filter(pk__in=sorted(type_ids))
980 .order_by("pk")
981 .select_for_update(of=("self",))
982 )
983 for device_type in locked_types:
984 self._locked_placement[(DeviceType, device_type.pk)] = device_type
985 locked_racks = Rack.objects.filter(pk__in=sorted(rack_ids)).order_by("pk").select_for_update(of=("self",))
986 for rack in locked_racks:
987 self._locked_placement[(Rack, rack.pk)] = rack
988 # A Device Type deleted between the unlocked read and the lock leaves its rows unplannable.
989 self._device_types = {
990 key: locked
991 for key, device_type in self._device_types.items()
992 if (locked := self._locked_placement.get((DeviceType, device_type.pk))) is not None
993 }
995 def _placement_candidate_devices(self) -> list[Any]:
996 """Return every stored Device a row can match, whose placement a review can retain."""
997 candidates = list(self._reviewed_devices.values())
998 for index in (
999 self._devices_by_source_id,
1000 self._devices_by_serial,
1001 self._devices_by_asset_tag,
1002 self._devices_by_name,
1003 ):
1004 for found in index.values():
1005 candidates.extend(found)
1006 return candidates
1008 def _named_rack_matches(self, rows: list[dict[str, Any]]) -> list[Any]:
1009 """Return the one existing Rack each row names, skipping the names that match several."""
1010 matched = []
1011 for row in rows:
1012 rack_name = _source_text(row.get("rack_name"))
1013 found = self._racks.get(identity_text(rack_name), ()) if rack_name else ()
1014 if len(found) == 1:
1015 matched.append(found[0])
1016 return matched
1018 def _load_dependency_objects(self, rows: list[dict[str, Any]]) -> tuple[dict[tuple[str, str], Any], dict[str, Any]]:
1019 """Load each Device Type and Device Role this batch can reference."""
1020 from dcim.models import DeviceRole, DeviceType
1022 type_keys = set()
1023 for row in rows:
1024 make = " ".join((_source_text(row.get("make")) or "Unknown").split())
1025 model = " ".join((_source_text(row.get("model")) or "Unknown").split())
1026 resolved = self._identity.resolve(make, model)
1027 if not resolved.ambiguous:
1028 type_keys.add(resolved[:2])
1030 referenced_types = DeviceType.objects.select_related("manufacturer").filter(
1031 manufacturer__slug__in={mfg_slug for mfg_slug, _dt_slug in type_keys},
1032 slug__in={dt_slug for _mfg_slug, dt_slug in type_keys},
1033 )
1034 device_types = {
1035 (device_type.manufacturer.slug, device_type.slug): device_type for device_type in referenced_types
1036 }
1037 role_slugs = {role_slug for role_slug in self._roles.values() if role_slug}
1038 roles = {role.slug: role for role in DeviceRole.objects.filter(slug__in=role_slugs)}
1039 return device_types, roles
1041 def _load_identity_objects(self, rows):
1042 """Load the global Device identity candidates and their visibility once."""
1043 from dcim.models import Device
1044 from django.db.models import F
1046 source_ids = {_source_text(row.get("source_id")) for row in rows} - {""}
1047 serials = {_source_text(row.get("serial")) for row in rows} - {""}
1048 asset_tags = {identity_text(_source_text(row.get("asset_tag"))[:50]) for row in rows} - {""}
1049 names = {identity_text(effective_device_name(row)) for row in rows} - {""}
1050 devices = Device.objects.select_related(
1051 "device_type__manufacturer",
1052 "rack__location",
1053 "role",
1054 "tenant",
1055 "location",
1056 "site",
1057 )
1058 stored_devices = list(
1059 devices.select_related("data_import_source").filter(
1060 data_import_source__profile=self.profile,
1061 data_import_source__source_id__in=source_ids,
1062 )
1063 )
1064 serial_devices = list(devices.filter(serial__in=serials))
1065 asset_tag_devices = list(
1066 devices.annotate(_identity_asset_tag=identity_expression(F("asset_tag"))).filter(
1067 _identity_asset_tag__in=asset_tags
1068 )
1069 )
1070 tenant_filter = {"tenant": self.reader.tenant} if self.reader.tenant is not None else {"tenant__isnull": True}
1071 name_devices = (
1072 list(
1073 devices.annotate(_identity_name=identity_expression(F("name"))).filter(
1074 _identity_name__in=names,
1075 site=self.reader.site,
1076 **tenant_filter,
1077 )
1078 )
1079 if self.reader.site is not None
1080 else []
1081 )
1083 def index(items, key):
1084 """Group already-loaded devices by one matching value."""
1085 found = {}
1086 for device in items:
1087 found.setdefault(key(device), []).append(device)
1088 return found
1090 candidates = {device.pk for device in (*stored_devices, *serial_devices, *asset_tag_devices, *name_devices)}
1091 visible_ids = (
1092 frozenset(candidates)
1093 if self.reader.actor is None
1094 else frozenset(self.reader.devices().filter(pk__in=candidates).values_list("pk", flat=True))
1095 )
1096 return (
1097 index(stored_devices, lambda device: _source_text(device.data_import_source.source_id)),
1098 index(serial_devices, lambda device: _text(device.serial)),
1099 index(asset_tag_devices, lambda device: device._identity_asset_tag),
1100 index(name_devices, lambda device: device._identity_name),
1101 visible_ids,
1102 )
1104 def _is_identity_writing_row(self, row) -> bool:
1105 """Return whether a row can write Device identity fields."""
1106 mapping = self._mappings.get(_source_text(row.get("device_class")))
1107 source_id = _source_text(row.get("source_id"))
1108 position = source_position(row.get("u_position"))
1109 return bool(
1110 mapping
1111 and not mapping.creates_rack
1112 and not mapping.ignore
1113 and mapping.role_slug
1114 and not (source_id and source_id in self.ignored)
1115 and not (position is not None and position < 1)
1116 )
1118 @staticmethod
1119 def _rows_by_value(rows, field, fold) -> dict[str, list[int]]:
1120 """Return the source row numbers each non-empty value of *field* appears on."""
1121 found: dict[str, list[int]] = {}
1122 for row in rows:
1123 raw = _source_text(row.get(field))[:50] if field == "asset_tag" else row.get(field)
1124 value = identity_text(raw) if fold else _source_text(raw)
1125 if value:
1126 found.setdefault(value, []).append(row.get("_row_number"))
1127 return {value: numbers for value, numbers in found.items() if len(numbers) > 1}
1129 def _effective_identity_values(self, rows) -> dict[int, dict[str, str]]:
1130 """Return review-aware serial and asset-tag writes for duplicate checks."""
1131 values = {}
1132 for row in rows:
1133 row_number = row.get("_row_number")
1134 source_id = _source_text(row.get("source_id"))
1135 proposal = {
1136 "serial": _source_text(row.get("serial")),
1137 "asset_tag": _source_text(row.get("asset_tag"))[:50],
1138 }
1139 device_id = self._bindings.get(source_id)
1140 if device_id is None and source_id:
1141 reviewed_ids = self._review_device_ids.get(source_id, frozenset())
1142 if len(reviewed_ids) == 1:
1143 device_id = next(iter(reviewed_ids))
1144 device = self._reviewed_devices.get(device_id)
1145 if device is not None:
1146 review = self._reviewer.review(source_id, device, proposal)
1147 effective = review.effective_proposal
1148 proposal = {
1149 "serial": "" if "serial" in review.ignored else _source_text(effective.get("serial")),
1150 "asset_tag": (
1151 "" if "asset_tag" in review.ignored else _source_text(effective.get("asset_tag"))[:50]
1152 ),
1153 }
1154 values[row_number] = proposal
1155 return values
1157 def _rows_by_effective(self, field, fold) -> dict[str, list[int]]:
1158 """Return duplicate row numbers from review-aware identity values."""
1159 found: dict[str, list[int]] = {}
1160 for row_number, values in self._effective_identity.items():
1161 raw = values[field]
1162 value = identity_text(raw) if fold else raw
1163 if value:
1164 found.setdefault(value, []).append(row_number)
1165 return {value: numbers for value, numbers in found.items() if len(numbers) > 1}
1167 def _planned_racks_by_name(self, source_batch, profile) -> dict[str, _PlannedRack]:
1168 """Return final Rack candidates in this batch, keyed by comparison name."""
1169 rows = RackModule._rack_rows(source_batch, profile)
1170 ignored = _ignored_source_ids(profile)
1171 duplicate_names, duplicate_source_ids = rack_duplicate_keys(rows)
1172 mappings = {mapping.source_class: mapping for mapping in profile.class_role_mappings.all()}
1173 planned = {}
1174 for row in rows:
1175 if rack_row_rejection(row, ignored, duplicate_names, duplicate_source_ids) is not None:
1176 continue
1177 name_key = identity_text(rack_row_name(row))
1178 matches = self._racks.get(name_key, ())
1179 if len(matches) > 1:
1180 continue
1181 rack = matches[0] if matches else None
1182 mapping = mappings[_source_text(row.get("device_class"))]
1183 height = _coerce_rack_height(row.get("u_height"))
1184 serial = _source_text(row.get("serial"))
1185 rack_type_id = mapping.rack_type_id
1186 if rack is not None and not profile.adapter_settings.update_existing:
1187 continue
1188 candidate, validation = RackModule._validated_candidate(
1189 rack,
1190 rack_row_name(row),
1191 height,
1192 serial,
1193 rack_type_id,
1194 self.reader,
1195 profile.adapter_settings.custom_field_name,
1196 _source_text(row.get("source_id")),
1197 )
1198 if rack is not None and not RackModule._differs(rack, candidate):
1199 continue
1200 actor = self.reader.actor
1201 executable = validation is None and (actor is None or _candidate_save_is_allowed(actor, candidate))
1202 planned[name_key] = _PlannedRack(
1203 rack_unit_identity(row),
1204 "create" if rack is None else "update",
1205 candidate,
1206 executable,
1207 )
1208 return planned
1210 def clash(self, row) -> tuple[str, str, list[int]] | None:
1211 """Return the first identity another row in this batch also claims."""
1212 for field, code, fold in self._CLASH_FIELDS:
1213 raw = (
1214 row.get(field)
1215 if field == "source_id"
1216 else self._effective_identity.get(row.get("_row_number"), {}).get(field, "")
1217 )
1218 value = _source_text(raw)
1219 key = identity_text(value) if fold else value
1220 numbers = self._clashes[field].get(key) if key else None
1221 if numbers:
1222 return code, value, numbers
1223 return None
1225 def dependencies(self, row) -> _Dependencies:
1226 """Return existing relation objects and planned roles, or the first unmet dependency."""
1227 make = " ".join((_source_text(row.get("make")) or "Unknown").split())
1228 model = " ".join((_source_text(row.get("model")) or "Unknown").split())
1229 mfg_slug, dt_slug, explicit, ambiguous = self._identity.resolve(make, model)
1230 if ambiguous:
1231 return _Dependencies(
1232 missing=(_AMBIGUOUS_MAPPING_CODES[ambiguous], {"source_make": make, "source_model": model})
1233 )
1234 changes = []
1235 actor = self.reader.actor
1236 device_type = self._device_types.get((mfg_slug, dt_slug))
1237 if device_type is None:
1238 return _Dependencies(
1239 missing=(
1240 "device.device_type_missing",
1241 {
1242 "mfg_slug": mfg_slug,
1243 "dt_slug": dt_slug,
1244 "source_make": make,
1245 "source_model": model,
1246 },
1247 )
1248 )
1249 if not explicit and identity_text(device_type.model) != identity_text(model):
1250 return _Dependencies(
1251 missing=(
1252 "device.device_type_slug_collision",
1253 {"dt_slug": dt_slug, "source_model": model, "stored_model": device_type.model},
1254 )
1255 )
1257 role_slug = self._roles.get(_source_text(row.get("device_class")), "")
1258 if not role_slug:
1259 return _Dependencies(missing=("device.role_unconfigured", {"source_class": row.get("device_class")}))
1260 role = self._role_objects.get(role_slug)
1261 if role is None:
1262 from dcim.models import DeviceRole
1264 role_change = self._role_change(role_slug)
1265 role = DeviceRole(**role_change.payload)
1266 if actor is not None and not _candidate_save_is_allowed(actor, role):
1267 return _Dependencies(missing=("device.role_permission", {"role_slug": role_slug}))
1268 changes.append(role_change)
1270 rack_name = _source_text(row.get("rack_name"))
1271 rack_key = identity_text(rack_name)
1272 rack_matches = self._racks.get(rack_key, ()) if rack_name else ()
1273 if len(rack_matches) > 1:
1274 return _Dependencies(missing=("device.rack_ambiguous", {"rack_name": rack_name}))
1275 rack = rack_matches[0] if rack_matches else None
1276 if rack is not None:
1277 from dcim.models import Rack
1279 # The name scan above cannot lock, and this rack decides the placement claim.
1280 rack = self.placement_reference(Rack, rack.pk)
1281 planned_rack = self._planned_racks.get(rack_key) if rack_name else None
1282 if rack_name and rack is None and planned_rack is None:
1283 return _Dependencies(missing=("device.rack_missing", {"rack_name": rack_name}))
1284 executable_rack = planned_rack if planned_rack is not None and planned_rack.executable else None
1285 return _Dependencies(
1286 device_type=device_type,
1287 role=role,
1288 rack=rack,
1289 rack_identity=(
1290 planned_rack.identity if planned_rack is not None and planned_rack.operation == "create" else None
1291 ),
1292 planned_rack=executable_rack.candidate if executable_rack is not None else None,
1293 rack_change_identity=executable_rack.change_identity if executable_rack is not None else None,
1294 unavailable_rack_name=(
1295 rack_name if rack is None and planned_rack is not None and not planned_rack.executable else ""
1296 ),
1297 role_slug=role_slug,
1298 explicit_device_type=explicit,
1299 changes=tuple(changes),
1300 )
1302 @staticmethod
1303 def _role_change(role_slug) -> PlannedChange:
1304 return PlannedChange(
1305 identity=f"device_role:{role_slug}:create",
1306 target_module=DeviceModule.key,
1307 operation="create_role",
1308 payload={"slug": role_slug, "name": role_slug.replace("-", " ").title(), "color": "9e9e9e"},
1309 preconditions={"role_id": None},
1310 )
1312 def placement(self, row, device_type, rack, rack_identity) -> _Placement:
1313 """Return where this row puts the device, or the first reason it cannot go there."""
1314 zero_u = device_type.u_height == 0
1315 position = None if zero_u else source_position(row.get("u_position"))
1316 face = "" if zero_u else _translate(row.get("face"), self.side_map)
1317 airflow = _translate(row.get("airflow"), self.airflow_map)
1318 status = _translate(row.get("status"), self.status_map) or "active"
1319 has_rack = rack is not None or rack_identity is not None
1321 if position is None:
1322 # NetBox refuses a rack face on a device in no rack, so the plan never asks for one.
1323 return _Placement(position=None, face=face if has_rack else "", airflow=airflow, status=status)
1324 if not has_rack:
1325 return _Placement(refused=("device.rack_required", {"u_position": position}))
1326 if not face:
1327 return _Placement(refused=("device.face_required", {"u_position": position}))
1328 return _Placement(position=position, face=face, airflow=airflow, status=status)
1330 def prepare_claim(self, rack, rack_identity, placement, device_type, matched) -> _PlacementClaim:
1331 """Return the available units this row can claim without reserving them yet."""
1332 rack_key = rack.pk if rack is not None and rack.pk is not None else rack_identity
1333 if rack_key is None or placement.position is None or device_type.u_height == 0:
1334 return _PlacementClaim()
1335 rack_face = None if device_type.is_full_depth else placement.face
1336 claim_faces = tuple(dict.fromkeys(self.side_map.values())) if device_type.is_full_depth else (placement.face,)
1337 keys = tuple(
1338 (rack_key, claim_face, unit)
1339 for claim_face in claim_faces
1340 for unit in _occupied_units(placement.position, device_type.u_height)
1341 )
1342 for key in keys:
1343 claimed_by = self._claimed.get(key)
1344 if claimed_by is not None:
1345 return _PlacementClaim(
1346 refused=(
1347 "device.rack_position_claimed",
1348 {"u_position": placement.position, "claimed_by_row": claimed_by},
1349 )
1350 )
1352 if rack is not None:
1353 if rack.pk is None:
1354 position_available = set(_occupied_units(placement.position, device_type.u_height)).issubset(rack.units)
1355 else:
1356 available = rack.get_available_units(
1357 u_height=device_type.u_height,
1358 rack_face=rack_face,
1359 exclude=[matched.pk] if matched is not None else [],
1360 )
1361 position_available = placement.position in available
1362 if not position_available:
1363 return _PlacementClaim(
1364 refused=(
1365 "device.rack_position_occupied",
1366 {"u_position": placement.position, "rack_name": rack.name},
1367 )
1368 )
1369 return _PlacementClaim(keys=keys)
1371 def commit_claim(self, row, claim: _PlacementClaim) -> None:
1372 """Reserve a checked placement after its row has settled."""
1373 self._claimed.update({key: row.get("_row_number") for key in claim.keys})
1375 def match(self, row, name) -> _Match:
1376 """Return the stored device this row reconciles, strongest identifier first.
1378 Only for a row `_is_identity_writing_row` accepts: the collation keys are loaded for those.
1379 """
1380 source_id = _source_text(row.get("source_id"))
1382 bound_id = self._bindings.get(source_id) if source_id else None
1383 if bound_id is not None:
1384 bound = self._reviewed_devices.get(bound_id)
1385 if bound is not None:
1386 return self._visible_match(bound, "source ID link")
1388 stored = self._devices_by_source_id.get(source_id, ()) if source_id else ()
1389 if len(stored) > 1:
1390 return _Match(ambiguous="device.ambiguous_stored_source_id", value=source_id)
1391 if stored:
1392 return self._visible_match(stored[0], "stored source ID")
1394 reviewed_ids = self._review_device_ids.get(source_id, frozenset())
1395 if len(reviewed_ids) > 1:
1396 return _Match(ambiguous="device.ambiguous_field_review", value=source_id)
1397 if reviewed_ids:
1398 reviewed = self._reviewed_devices.get(next(iter(reviewed_ids)))
1399 if reviewed is not None:
1400 return self._visible_match(reviewed, "field review")
1402 for field, matches, code in (
1403 ("serial", self._devices_by_serial, "device.ambiguous_serial"),
1404 ("asset_tag", self._devices_by_asset_tag, "device.ambiguous_asset_tag"),
1405 ):
1406 value = _source_text(row.get(field))[:50] if field == "asset_tag" else _source_text(row.get(field))
1407 if not value:
1408 continue
1409 key = identity_text(value) if field == "asset_tag" else value
1410 found = matches.get(key, ())
1411 if len(found) > 1:
1412 return _Match(ambiguous=code, value=value)
1413 if found:
1414 return self._visible_match(found[0], field.replace("_", " "))
1416 if identity_text(name) in self._duplicate_names or self.reader.site is None:
1417 return _Match()
1418 name_value = _source_text(name)
1419 by_name = self._devices_by_name.get(identity_text(name_value), ()) if name_value else ()
1420 if len(by_name) > 1:
1421 return _Match(ambiguous="device.ambiguous_name", value=name)
1422 return self._visible_match(by_name[0], "name") if by_name else _Match()
1424 def _visible_match(self, device, method) -> _Match:
1425 """Return a global match with its scope and site safety attached."""
1426 inaccessible = (
1427 device.pk not in self._visible_reviewed_device_ids
1428 if device.pk in self._reviewed_devices
1429 else device.pk not in self._visible_identity_device_ids
1430 )
1431 return _Match(device=device, method=method, inaccessible=inaccessible)
1433 def binding_conflict(self, row, match) -> str:
1434 """Return the source identity that already claims a matched device."""
1435 source_id = _source_text(row.get("source_id"))
1436 bound_source = self._bound_sources.get(match.device.pk)
1437 if bound_source and bound_source != source_id:
1438 return bound_source
1439 previous = self._claimed_devices.get(match.device.pk)
1440 if previous is not None and previous[0] != row.get("_row_number"):
1441 return previous[1] or f"row {previous[0]}"
1442 return ""
1444 def commit_device_claim(self, row, match) -> None:
1445 """Reserve a matched Device for one accepted unit."""
1446 self._claimed_devices[match.device.pk] = (
1447 row.get("_row_number"),
1448 _source_text(row.get("source_id")),
1449 )
1451 def suggest_name(self, row) -> str:
1452 """Return and reserve one deterministic name for a duplicate source row."""
1453 name = effective_device_name(row)
1454 rack_name = _source_text(row.get("rack_name")) or "NO-RACK"
1455 position = normalize_for_compare(row.get("u_position")) or "NO-U"
1456 base = f"{name}-{rack_name}-U{position}" if position != "NO-U" else f"{name}-{rack_name}"
1457 candidate = base[:64]
1458 if identity_text(candidate) in self._reserved_names:
1459 source_suffix = _source_text(row.get("source_id")) or str(row.get("_row_number") or "ROW")
1460 source_suffix = re.sub(r"[^A-Za-z0-9_.-]+", "-", source_suffix).strip("-") or "ROW"
1461 candidate = f"{base[: max(1, 63 - len(source_suffix))]}-{source_suffix}"[:64]
1462 unique = candidate
1463 counter = 2
1464 while identity_text(unique) in self._reserved_names:
1465 suffix = f"-{counter}"
1466 unique = f"{candidate[: 64 - len(suffix)]}{suffix}"
1467 counter += 1
1468 self._reserved_names.add(identity_text(unique))
1469 return unique
1471 def review(self, row, device, dependencies, placement, payload):
1472 """Return the operator's saved review for one matched device, in the reviewer's terms."""
1473 device_type = dependencies.device_type
1474 manufacturer = device_type.manufacturer
1475 device_type_identity = (
1476 manufacturer.slug,
1477 device_type.slug,
1478 manufacturer.name,
1479 device_type.model,
1480 )
1481 ip_fields = payload.get("ip_fields") or {}
1482 proposal = {
1483 "device_name": payload["name"],
1484 "serial": payload["serial"],
1485 "asset_tag": payload["asset_tag"],
1486 "u_position": placement.position,
1487 "face": placement.face,
1488 "airflow": placement.airflow,
1489 "status": placement.status,
1490 "rack_name": _source_text(row.get("rack_name")),
1491 "_rack_location_id": self.reader.location.pk if self.reader.location is not None else None,
1492 "device_type": device_type_identity,
1493 "role": _text(dependencies.role.slug),
1494 "tenant": self.reader.tenant,
1495 "location": self.reader.location,
1496 **{name: ip_fields.get(name, "") for name in ip_assignment.IP_FIELD_FAMILY},
1497 }
1498 return self._reviewer.review(_source_text(row.get("source_id")), device, proposal)
1500 def contact_review(self, row, device):
1501 """Return the reviewed primary contact for one row, before anything is written."""
1502 return PrimaryContactResolver.review(
1503 device,
1504 row,
1505 self.profile,
1506 self.reader.actor,
1507 candidate_source_columns=self._candidate_columns,
1508 )
1511class DeviceModule:
1512 """Plans the Devices a flat source batch describes.
1514 Section 4.4 makes a Device Type, a Device Role and a Rack dependencies of a device, not part of
1515 it, so a device row that names one NetBox does not hold is blocked rather than invalid. Creating
1516 them is another unit's work.
1517 """
1519 key = "device"
1520 consumes = frozenset({OutputKind.DEVICE_SOURCE_ROW})
1522 def plan(
1523 self,
1524 source_batch,
1525 profile,
1526 catalog,
1527 netbox_reader,
1528 *,
1529 lock_plan_references: bool = False,
1530 ) -> list[SynchronizationUnit]:
1531 """Return one Synchronization Unit per device row, with the disposition its state earns."""
1532 rows = self._device_rows(source_batch, profile)
1533 if not rows:
1534 return []
1535 if netbox_reader.site is None:
1536 raise PlanningTargetUnavailable("Device planning needs an import target site.")
1537 batch = _DeviceBatch(source_batch, rows, profile, netbox_reader, lock_plan_references=lock_plan_references)
1538 return [self._unit(row, batch) for row in rows]
1540 @staticmethod
1541 def _device_rows(source_batch, profile) -> list[dict]:
1542 """Return every non-rack row, including rows whose policy must be reviewed."""
1543 mappings = {mapping.source_class: mapping for mapping in profile.class_role_mappings.all()}
1544 return [
1545 row
1546 for row in source_batch.rows
1547 if not (
1548 (mapping := mappings.get(_source_text(row.get("device_class")))) is not None and mapping.creates_rack
1549 )
1550 ]
1552 @staticmethod
1553 def unit_identity(row) -> str:
1554 """Return the identity that survives replanning, which is never the row number."""
1555 source_id = _source_text(row.get("source_id"))
1556 if source_id:
1557 return f"device:source:{source_id}"
1558 return f"device:name:{identity_text(effective_device_name(row))}"
1560 def _unit(self, row, batch) -> SynchronizationUnit: # noqa: C901
1561 """Return the one unit this row produces, naming every problem it can already prove.
1563 The checks run in one fixed order and record what they find instead of returning, so a row
1564 held up by its identity still reports the mapping, placement and Contact work it needs. The
1565 first problem stays the authoritative one, and an operator fixing them in any order sees the
1566 rest of the list shrink. Only the checks whose inputs are available run, and every one of
1567 them reads: the helpers that reserve a name, a rack unit or a device stay on the paths that
1568 settle a row.
1569 """
1570 identity = self.unit_identity(row)
1571 name = effective_device_name(row)
1572 source_id = _source_text(row.get("source_id"))
1573 if source_id in batch._clashes["source_id"] or (
1574 not source_id and identity_text(name) in batch._duplicate_names
1575 ):
1576 identity = f"{identity}:row:{row.get('_row_number')}"
1577 display = _unit_display(row, self.key, name, _source_text(row.get("rack_name")))
1578 display["extra_data"].update(_class_mapping_display(batch._mappings.get(_source_text(row.get("device_class")))))
1579 display["device_name"] = name
1580 display["source_id"] = source_id
1582 issues: list[tuple[str, str, dict]] = []
1584 def problem(disposition, code, extra=None) -> None:
1585 """Record one problem this row has, so the checks after it still run."""
1586 issues.append((disposition, code, {**display, **(extra or {})}))
1588 clash = batch.clash(row)
1589 if clash is not None:
1590 code, value, numbers = clash
1591 other_rows = [number for number in numbers if number != row.get("_row_number")]
1592 label = {
1593 "device.duplicate_source_id": "source ID",
1594 "device.duplicate_serial": "serial",
1595 "device.duplicate_asset_tag": "asset tag",
1596 }[code]
1597 conflict_display = {
1598 "message": _duplicate_value_detail(label, value, other_rows),
1599 "value": value,
1600 "rows": numbers,
1601 }
1602 if code == "device.duplicate_serial":
1603 conflict_display["duplicate_serial"] = value
1604 problem(Disposition.INVALID, code, conflict_display)
1606 # An excluded or below-rack row is an answer rather than a problem, so it reports no list.
1607 if not issues and source_id and source_id in batch.ignored:
1608 ignored_display = {
1609 **display,
1610 "extra_data": {**display["extra_data"], "ignore_kind": "individual"},
1611 }
1612 return SynchronizationUnit(
1613 identity=identity,
1614 disposition=Disposition.EXCLUDED,
1615 diagnostics=(
1616 Diagnostic(
1617 code="device.ignored",
1618 severity=Severity.INFO,
1619 identities=(identity,),
1620 display=ignored_display,
1621 ),
1622 ),
1623 display=ignored_display,
1624 )
1626 position = source_position(row.get("u_position"))
1627 if not issues and position is not None and position < 1:
1628 return SynchronizationUnit(
1629 identity=identity,
1630 disposition=Disposition.NO_OP,
1631 diagnostics=(
1632 Diagnostic(
1633 code="device.below_rack",
1634 severity=Severity.INFO,
1635 identities=(identity,),
1636 display={**display, "u_position": position},
1637 ),
1638 ),
1639 display=display,
1640 )
1641 if not name:
1642 problem(Disposition.INVALID, "device.missing_name")
1644 mapping = batch._mappings.get(_source_text(row.get("device_class")))
1645 if mapping is None:
1646 problem(
1647 Disposition.INVALID,
1648 "device.class_unmapped",
1649 {"source_class": _source_text(row.get("device_class"))},
1650 )
1651 elif mapping.ignore:
1652 if not issues:
1653 return SynchronizationUnit(
1654 identity=identity,
1655 disposition=Disposition.EXCLUDED,
1656 diagnostics=(
1657 Diagnostic(
1658 code="device.class_ignored",
1659 severity=Severity.INFO,
1660 identities=(identity,),
1661 display={
1662 **display,
1663 "extra_data": {**display["extra_data"], "ignore_kind": "class"},
1664 },
1665 ),
1666 ),
1667 display={**display, "extra_data": {**display["extra_data"], "ignore_kind": "class"}},
1668 )
1669 # The class says to skip this row, so nothing after it has anything to plan.
1670 mapping = None
1672 dependencies = batch.dependencies(row) if mapping is not None else None
1673 if dependencies is not None and dependencies.missing is not None:
1674 code, missing_display = dependencies.missing
1675 problem(Disposition.BLOCKED, code, missing_display)
1677 # Only an identity-writing row has its name and asset tag in the batch's collation keys.
1678 match = batch.match(row, name) if name and batch._is_identity_writing_row(row) else _Match()
1679 if match.ambiguous is not None:
1680 problem(Disposition.INVALID, match.ambiguous, {"value": match.value})
1681 elif match.inaccessible:
1682 problem(Disposition.INVALID, "device.inaccessible_match")
1683 elif match.device is not None and match.device.site_id != batch.reader.site.pk:
1684 problem(
1685 Disposition.INVALID,
1686 "device.cross_site_match",
1687 {"netbox_device_id": match.device.pk, "match_method": match.method},
1688 )
1689 elif identity_text(name) in batch._duplicate_names and match.device is None:
1690 problem(
1691 Disposition.INVALID,
1692 "device.duplicate_name",
1693 {"extra_data": {**display["extra_data"], "suggested_name": batch.suggest_name(row)}},
1694 )
1695 elif match.device is not None and (bound_source := batch.binding_conflict(row, match)):
1696 problem(
1697 Disposition.INVALID,
1698 "device.already_bound",
1699 {"bound_source_id": bound_source, "netbox_device_id": match.device.pk},
1700 )
1702 # Only a row that has settled every check so far may claim the device it matched.
1703 if not issues and match.device is not None:
1704 name_note = (
1705 f"; name stays '{match.device.name}' (source: '{name}')"
1706 if match.device.name != name
1707 else "; name unchanged"
1708 )
1709 display = {
1710 **display,
1711 "detail": f"Will update '{match.device.name}' (matched by {match.method}{name_note})",
1712 "netbox_url": match.device.get_absolute_url(),
1713 "extra_data": {
1714 **display["extra_data"],
1715 "netbox_device_id": match.device.pk,
1716 "netbox_face": match.device.face or "",
1717 "netbox_position": normalize_for_compare(match.device.position),
1718 "netbox_rack_name": match.device.rack.name if match.device.rack_id else "",
1719 # A row refused for an identity conflict states no change, so it needs this here.
1720 "_placement_state": {
1721 # `_placement_differs` reads location as placement, so the baseline states it.
1722 "location_id": match.device.location_id,
1723 "rack_id": match.device.rack_id,
1724 "position": normalize_for_compare(match.device.position),
1725 "face": match.device.face or "",
1726 },
1727 },
1728 }
1729 if not batch.profile.adapter_settings.update_existing:
1730 display["detail"] = (
1731 f"Matched to '{match.device.name}' (by {match.method}{name_note}, skip: update_existing off)"
1732 )
1733 batch.commit_device_claim(row, match)
1734 return SynchronizationUnit(
1735 identity=identity,
1736 disposition=Disposition.NO_OP,
1737 display=display,
1738 )
1740 placement = None
1741 placement_rack = None
1742 if dependencies is not None and dependencies.missing is None:
1743 placement_rack = dependencies.planned_rack if dependencies.planned_rack is not None else dependencies.rack
1744 placement = batch.placement(row, dependencies.device_type, placement_rack, dependencies.rack_identity)
1745 if placement.refused is not None:
1746 code, placement_display = placement.refused
1747 problem(Disposition.INVALID, code, placement_display)
1748 placement = None
1749 else:
1750 display = {
1751 **display,
1752 "extra_data": {
1753 **display["extra_data"],
1754 "airflow": placement.airflow,
1755 "dt_slug": dependencies.device_type.slug,
1756 "face": placement.face,
1757 "is_explicit_mapping": dependencies.explicit_device_type,
1758 "mfg_slug": dependencies.device_type.manufacturer.slug,
1759 "status": placement.status,
1760 "u_height": _display_value(dependencies.device_type.u_height),
1761 "u_position": placement.position,
1762 **({"zero_u": True} if dependencies.device_type.u_height == 0 else {}),
1763 },
1764 }
1766 contact = None
1767 try:
1768 contact = batch.contact_review(row, match.device)
1769 except ObjectPermissionDenied as exc:
1770 problem(Disposition.BLOCKED, "device.contact_permission", {"message": str(exc)})
1771 except DanglingProfileReference as exc:
1772 problem(Disposition.BLOCKED, "profile.dangling_reference", {"message": "; ".join(exc.messages)})
1773 except ContactResolutionRequired as exc:
1774 problem(
1775 Disposition.INVALID,
1776 "device.contact_resolution_required",
1777 {
1778 "extra_data": {
1779 **display["extra_data"],
1780 "candidate_values": {"contact": exc.candidate_values},
1781 "contact_suggestion": exc.suggestion or {},
1782 },
1783 },
1784 )
1785 except ValidationError as exc:
1786 problem(Disposition.INVALID, "device.contact_invalid", {"error": "; ".join(exc.messages)})
1787 else:
1788 display = {
1789 **display,
1790 "extra_data": {
1791 **display["extra_data"],
1792 "candidate_values": {"contact": contact.candidate_values} if contact.candidate_values else {},
1793 "contact_suggestion": contact.suggestion or {},
1794 "extra_columns": contact.extra_columns,
1795 },
1796 }
1798 # Everything below reads all of these, and a row that could not settle one recorded why.
1799 if dependencies is None or dependencies.missing is not None or placement is None or contact is None:
1800 return _with_issues(identity, issues)
1802 payload = self._payload(row, name, dependencies, placement, batch)
1803 ip_fields, ip_diagnostics = self._ip_fields(row, identity, display)
1804 payload = {
1805 **payload,
1806 "contact": _contact_payload(contact),
1807 "source_id": source_id,
1808 "extra_columns": contact.extra_columns,
1809 "ip_fields": ip_fields,
1810 }
1811 if match.device is None:
1812 if dependencies.unavailable_rack_name:
1813 problem(
1814 Disposition.BLOCKED,
1815 "device.rack_missing",
1816 {"rack_name": dependencies.unavailable_rack_name},
1817 )
1818 return _with_issues(identity, issues)
1819 claim = batch.prepare_claim(
1820 placement_rack,
1821 dependencies.rack_identity,
1822 placement,
1823 dependencies.device_type,
1824 None,
1825 )
1826 if claim.refused is not None:
1827 code, taken_display = claim.refused
1828 problem(Disposition.INVALID, code, taken_display)
1829 actor = batch.reader.actor
1830 candidate, validation = self._validated_candidate(None, payload, batch.profile)
1831 prospective_relations = _planned_device_relations(payload, dependencies)
1832 prospective_relations.update(_prospective_ip_relations(candidate, payload["ip_fields"]))
1833 if validation:
1834 problem(Disposition.INVALID, "device.validation_failed", {"message": validation})
1835 if actor is not None and not _candidate_save_is_allowed(
1836 actor,
1837 candidate,
1838 prospective_relations,
1839 ):
1840 problem(Disposition.BLOCKED, "device.add_permission")
1841 if issues:
1842 return _with_issues(identity, issues)
1843 batch.commit_claim(row, claim)
1844 device_change = self._change(
1845 identity,
1846 "create",
1847 payload,
1848 None,
1849 dependencies.rack_change_identity if "rack" in prospective_relations else None,
1850 dependencies.changes,
1851 batch.profile,
1852 )
1853 display = {
1854 **display,
1855 "extra_data": {
1856 **display["extra_data"],
1857 "pending_write_contact": not _contact_writes_nothing(contact),
1858 # A create stores its provenance record, so there is never a current one to reuse.
1859 "pending_write_provenance": True,
1860 },
1861 }
1862 return SynchronizationUnit(
1863 identity=identity,
1864 disposition=Disposition.ACTIONABLE,
1865 changes=(*dependencies.changes, device_change),
1866 diagnostics=ip_diagnostics,
1867 display=display,
1868 )
1869 review = batch.review(row, match.device, dependencies, placement, payload)
1870 display = self._review_display(display, review)
1871 payload = _reviewed_payload(payload, review, match.device)
1872 if dependencies.unavailable_rack_name and payload["rack_id"] is None and payload["rack_name"]:
1873 problem(
1874 Disposition.BLOCKED,
1875 "device.rack_missing",
1876 {"rack_name": dependencies.unavailable_rack_name},
1877 )
1878 return _with_issues(identity, issues)
1879 relation_changes = () if "role" in review.ignored else dependencies.changes
1880 effective_type = dependencies.device_type
1881 if payload["device_type_id"] is not None and payload["device_type_id"] != dependencies.device_type.pk:
1882 from dcim.models import DeviceType
1884 # A retained review value sizes the placement too, so the replan must hold it.
1885 effective_type = batch.placement_reference(DeviceType, payload["device_type_id"])
1886 if effective_type is None:
1887 # Every check below reads the device type this row would write.
1888 problem(Disposition.INVALID, "device.device_type_missing")
1889 return _with_issues(identity, issues)
1890 if effective_type.u_height == 0:
1891 zero_u_conflicts = []
1892 if "u_position" in review.ignored and payload["u_position"] is not None:
1893 zero_u_conflicts.append("u_position")
1894 else:
1895 payload["u_position"] = None
1896 if "face" in review.ignored and payload["face"]:
1897 zero_u_conflicts.append("face")
1898 else:
1899 payload["face"] = ""
1900 if zero_u_conflicts:
1901 problem(Disposition.INVALID, "device.zero_u_review_conflict", {"fields": zero_u_conflicts})
1902 effective_rack = placement_rack
1903 effective_rack_identity = dependencies.rack_identity
1904 if payload["rack_name"] is None:
1905 effective_rack_identity = None
1906 if payload["rack_id"] != (dependencies.rack.pk if dependencies.rack is not None else None):
1907 from dcim.models import Rack
1909 effective_rack = batch.placement_reference(Rack, payload["rack_id"]) if payload["rack_id"] else None
1910 effective_placement = _Placement(
1911 position=payload["u_position"],
1912 face=payload["face"],
1913 airflow=payload["airflow"],
1914 status=payload["status"],
1915 )
1916 claim = batch.prepare_claim(
1917 effective_rack,
1918 effective_rack_identity,
1919 effective_placement,
1920 effective_type,
1921 match.device,
1922 )
1923 if claim.refused is not None:
1924 code, taken_display = claim.refused
1925 problem(Disposition.INVALID, code, taken_display)
1926 if not self._placement_differs(match.device, payload):
1927 display = {
1928 **display,
1929 "extra_data": {**display["extra_data"], "placement_sync_writes_nothing": True},
1930 }
1931 if match.method == "name" and self._placement_differs(match.device, payload):
1932 # The preview offers the rename for both refusals, so both state the name it would use.
1933 rename = {"extra_data": {**display["extra_data"], "suggested_name": batch.suggest_name(row)}}
1934 # A stored Device with no placement has none to sit at, so it reads as a different refusal.
1935 if self._device_is_unplaced(match.device):
1936 problem(Disposition.INVALID, "device.name_unplaced_match", rename)
1937 else:
1938 problem(Disposition.INVALID, "device.name_placement_conflict", rename)
1939 contact_writes_nothing = _contact_writes_nothing(contact)
1940 provenance_is_current = _provenance_is_current(match.device, payload, batch.profile)
1941 if not issues and not self._differs(match.device, payload) and contact_writes_nothing and provenance_is_current:
1942 batch.commit_claim(row, claim)
1943 batch.commit_device_claim(row, match)
1944 display = {
1945 **display,
1946 "detail": f"Device '{match.device.name}' matches this row, which writes nothing",
1947 "extra_data": {
1948 **display["extra_data"],
1949 "placement_sync_writes_nothing": not self._placement_differs(match.device, payload),
1950 "writes_nothing": True,
1951 },
1952 }
1953 return SynchronizationUnit(
1954 identity=identity,
1955 disposition=Disposition.NO_OP,
1956 diagnostics=ip_diagnostics,
1957 display=display,
1958 )
1959 actor = batch.reader.actor
1960 candidate, validation = self._validated_candidate(match.device, payload, batch.profile)
1961 prospective_relations = _planned_device_relations(payload, dependencies)
1962 prospective_relations.update(_prospective_ip_relations(candidate, payload["ip_fields"]))
1963 if validation:
1964 problem(Disposition.INVALID, "device.validation_failed", {"message": validation})
1965 if actor is not None and not _candidate_save_is_allowed(actor, candidate, prospective_relations):
1966 problem(Disposition.BLOCKED, "device.change_permission")
1967 if issues:
1968 return _with_issues(identity, issues)
1969 batch.commit_claim(row, claim)
1970 batch.commit_device_claim(row, match)
1971 display = {
1972 **display,
1973 "extra_data": {
1974 **display["extra_data"],
1975 "pending_write_contact": not contact_writes_nothing,
1976 "pending_write_provenance": not provenance_is_current,
1977 },
1978 }
1979 device_change = self._change(
1980 identity,
1981 "update",
1982 payload,
1983 match.device,
1984 dependencies.rack_change_identity if "rack" in prospective_relations else None,
1985 relation_changes,
1986 batch.profile,
1987 )
1988 return SynchronizationUnit(
1989 identity=identity,
1990 disposition=Disposition.ACTIONABLE,
1991 changes=(*relation_changes, device_change),
1992 diagnostics=ip_diagnostics,
1993 display=display,
1994 )
1996 @staticmethod
1997 def _placement_differs(device, payload) -> bool:
1998 """Return whether a name-only match would move the stored Device."""
1999 return (
2000 payload["rack_name"] is not None
2001 or device.location_id != payload["location_id"]
2002 or device.rack_id != payload["rack_id"]
2003 or normalize_for_compare(device.position) != normalize_for_compare(payload["u_position"])
2004 or (device.face or "") != payload["face"]
2005 )
2007 @staticmethod
2008 def _device_is_unplaced(device) -> bool:
2009 """Return whether the stored Device records no placement for the source to move it from."""
2010 return device.location_id is None and device.rack_id is None and device.position is None and not device.face
2012 @staticmethod
2013 def _validated_candidate(device, payload, profile):
2014 """Return the prospective Device and its model validation message, if resolvable."""
2015 from dcim.models import Device
2017 candidate = copy(device) if device is not None else Device(name=payload["name"])
2018 candidate.device_type_id = payload["device_type_id"]
2019 candidate.role_id = payload["role_id"]
2020 candidate.site_id = payload["site_id"]
2021 candidate.location_id = payload["location_id"]
2022 candidate.rack_id = payload["rack_id"]
2023 candidate.position = payload["u_position"]
2024 candidate.face = payload["face"]
2025 candidate.status = payload["status"]
2026 candidate.tenant_id = payload["tenant_id"]
2027 if payload["airflow"]:
2028 candidate.airflow = payload["airflow"]
2029 for field in ("serial", "asset_tag"):
2030 if payload[field]:
2031 setattr(candidate, field, payload[field])
2032 custom_field = profile.adapter_settings.custom_field_name
2033 source_id = payload.get("source_id") or ""
2034 if custom_field and source_id:
2035 candidate.custom_field_data = {
2036 **candidate.custom_field_data,
2037 custom_field: source_id,
2038 }
2039 unresolved = set()
2040 if payload["role_id"] is None:
2041 unresolved.add("role")
2042 if payload["rack_name"] is not None and payload["rack_id"] is None:
2043 unresolved.update(("rack", "position", "face"))
2044 try:
2045 if unresolved:
2046 # Model validation reads the unresolved relations, so only the row's own fields can run.
2047 candidate.clean_fields(exclude=unresolved)
2048 else:
2049 candidate.full_clean()
2050 except ValidationError as exc:
2051 if hasattr(exc, "message_dict"):
2052 return candidate, "; ".join(
2053 f"{field}: {', '.join(errors)}" for field, errors in exc.message_dict.items()
2054 )
2055 return candidate, "; ".join(exc.messages)
2056 return candidate, ""
2058 @staticmethod
2059 def _review_display(display, review) -> dict:
2060 """Attach the matched Device review state as display-only plan data."""
2061 snapshots = {
2062 field: {"file": file_snapshot, "netbox": netbox_snapshot}
2063 for field, (file_snapshot, netbox_snapshot) in review.snapshots.items()
2064 }
2065 return {
2066 **display,
2067 "extra_data": {
2068 **display["extra_data"],
2069 "field_diff": review.differing,
2070 "field_ignored": review.ignored,
2071 "field_informational": review.informational,
2072 "field_matching": review.matching,
2073 "field_review_snapshots": snapshots,
2074 "field_non_writable": sorted(DeviceFieldReviewer.non_writable_fields()),
2075 },
2076 }
2078 def apply(self, planned_change: PlannedChange, execution_context) -> Any:
2079 """Apply one device change, having locked its row and rechecked its preconditions."""
2080 from dcim.models import Device, DeviceRole, Rack
2082 from .object_permissions import enforce_saved_object_permission
2084 payload = planned_change.payload
2085 if planned_change.operation == "create_role":
2086 if DeviceRole.objects.filter(slug=payload["slug"]).exists():
2087 raise PreconditionFailed(f"Device Role slug '{payload['slug']}' appeared after planning.")
2088 role = DeviceRole(name=payload["name"], slug=payload["slug"], color=payload["color"])
2089 role.full_clean()
2090 role.save()
2091 enforce_saved_object_permission(role, execution_context.actor, "add")
2092 return role
2094 device_id = planned_change.preconditions.get("device_id")
2095 if device_id is None:
2096 conflict = self._create_identity_conflict(payload, execution_context.profile)
2097 if conflict:
2098 raise PreconditionFailed(f"{conflict} appeared after planning.")
2099 device = Device()
2100 action = "add"
2101 else:
2102 # `of=("self",)` because NetBox's default Device queryset outer-joins, which cannot be locked.
2103 device = Device.objects.filter(pk=device_id).select_for_update(of=("self",)).first()
2104 if device is None:
2105 raise PreconditionFailed(f"Device {device_id} is gone, so '{payload['name']}' cannot be updated.")
2106 current = self._precondition_state(
2107 device,
2108 execution_context.profile,
2109 payload.get("source_id") or "",
2110 )
2111 if current != planned_change.preconditions.get("state"):
2112 raise PreconditionFailed(f"Device '{device.name}' changed after the plan was made.")
2113 device.snapshot()
2114 action = "change"
2116 role_id = payload["role_id"]
2117 if role_id is None:
2118 role = DeviceRole.objects.filter(slug=payload["role_slug"]).first()
2119 if role is None:
2120 raise PreconditionFailed("The planned Device Role dependency is still absent.")
2121 role_id = role.pk
2123 rack_id = payload["rack_id"]
2124 rack_name = payload["rack_name"]
2125 if rack_id is None and rack_name:
2126 rack = (
2127 Rack.objects.filter(site_id=payload["site_id"], location_id=payload["location_id"])
2128 .filter(identity_in("name", [identity_text(rack_name)]))
2129 .select_for_update(of=("self",))
2130 .first()
2131 )
2132 if rack is None:
2133 raise PreconditionFailed(
2134 f"Rack '{rack_name}' is still absent, so '{payload['name']}' cannot be placed."
2135 )
2136 rack_id = rack.pk
2138 if action == "add":
2139 device.name = payload["name"]
2140 device.device_type_id = payload["device_type_id"]
2141 device.role_id = role_id
2142 device.site_id = payload["site_id"]
2143 device.location_id = payload["location_id"]
2144 device.rack_id = rack_id
2145 device.position = payload["u_position"]
2146 device.face = payload["face"]
2147 device.status = payload["status"]
2148 device.tenant_id = payload["tenant_id"]
2149 if payload["airflow"]:
2150 device.airflow = payload["airflow"]
2151 for field in ("serial", "asset_tag"):
2152 if payload[field]:
2153 setattr(device, field, payload[field])
2154 device.full_clean()
2155 device.save()
2156 unassigned = _assign_ips(device, payload.get("ip_fields") or {}, execution_context.actor)
2157 _apply_contact(device, payload.get("contact"), execution_context)
2158 _store_provenance(device, payload, unassigned, execution_context)
2159 # Constraints are only evaluated against the saved row, so this reads the state the row leaves.
2160 enforce_saved_object_permission(device, execution_context.actor, action)
2161 return device
2163 @staticmethod
2164 def _ip_fields(row, identity, display) -> tuple[dict, tuple[Diagnostic, ...]]:
2165 """Return addresses in their declared families and warn about invalid values."""
2166 fields = {}
2167 diagnostics = []
2168 for name in ip_assignment.IP_FIELD_FAMILY:
2169 raw = _source_text(row.get(name))
2170 if not raw:
2171 continue
2172 try:
2173 address = ip_assignment.normalized_address(name, raw)
2174 except ip_assignment.IPAssignmentError:
2175 pass
2176 else:
2177 fields[name] = address
2178 continue
2179 diagnostics.append(
2180 Diagnostic(
2181 code="device.unparseable_ip",
2182 severity=Severity.WARNING,
2183 identities=(identity,),
2184 display={**display, "field": name, "value": raw},
2185 )
2186 )
2187 return fields, tuple(diagnostics)
2189 @staticmethod
2190 def _payload(row, name, dependencies, placement, batch) -> dict:
2191 """Return the device state this row asks for, resolved to what a write needs."""
2192 return {
2193 "name": name,
2194 "serial": _source_text(row.get("serial")),
2195 "asset_tag": _source_text(row.get("asset_tag"))[:50],
2196 "u_position": placement.position,
2197 "face": placement.face,
2198 "airflow": placement.airflow,
2199 "status": placement.status,
2200 "device_type_id": dependencies.device_type.pk,
2201 "role_id": dependencies.role.pk,
2202 "role_slug": dependencies.role_slug,
2203 "rack_id": dependencies.rack.pk if dependencies.rack is not None else None,
2204 "rack_name": _source_text(row.get("rack_name")) if dependencies.rack_identity is not None else None,
2205 "site_id": batch.reader.site.pk if batch.reader.site is not None else None,
2206 "location_id": batch.reader.location.pk if batch.reader.location is not None else None,
2207 "tenant_id": batch.reader.tenant.pk if batch.reader.tenant is not None else None,
2208 }
2210 @staticmethod
2211 def _differs(device, payload) -> bool:
2212 """Return whether the stored device already holds what the row asks for.
2214 The name is absent on purpose. It is how a row finds a device, and an import reconciles
2215 the device it matched rather than retitling it.
2216 """
2217 if device.device_type_id != payload["device_type_id"] or device.role_id != payload["role_id"]:
2218 return True
2219 if payload["rack_name"] is not None:
2220 return True
2221 if device.rack_id != payload["rack_id"]:
2222 return True
2223 # The target context assigns both fields, including a blank value that clears one.
2224 if device.location_id != payload["location_id"]:
2225 return True
2226 if device.tenant_id != payload["tenant_id"]:
2227 return True
2228 if normalize_for_compare(device.position) != normalize_for_compare(payload["u_position"]):
2229 return True
2230 if _text(device.status) != payload["status"]:
2231 return True
2232 if _text(device.face) != payload["face"]:
2233 return True
2234 if payload["airflow"] and _text(device.airflow) != payload["airflow"]:
2235 return True
2236 for field in ("serial", "asset_tag"):
2237 if payload[field] and _text(getattr(device, field)) != payload[field]:
2238 return True
2239 return bool(
2240 any(
2241 not ip_assignment.already_assigned(device, field, address)
2242 for field, address in (payload.get("ip_fields") or {}).items()
2243 )
2244 )
2246 @staticmethod
2247 def _change(
2248 identity,
2249 operation,
2250 payload,
2251 device,
2252 rack_change_identity=None,
2253 relation_changes=(),
2254 profile=None,
2255 ) -> PlannedChange:
2256 """Return the one write this unit performs, with the target state it assumed."""
2257 if device is None:
2258 preconditions: dict = {"device_id": None}
2259 else:
2260 preconditions = {
2261 "device_id": device.pk,
2262 "state": DeviceModule._precondition_state(device, profile, payload.get("source_id") or ""),
2263 }
2264 dependencies = [change.identity for change in relation_changes]
2265 if rack_change_identity is not None:
2266 dependencies.append(rack_change_identity)
2267 return PlannedChange(
2268 identity=f"{identity}:{operation}",
2269 target_module=DeviceModule.key,
2270 operation=operation,
2271 payload=payload,
2272 dependencies=tuple(dependencies),
2273 preconditions=preconditions,
2274 )
2276 @staticmethod
2277 def _precondition_state(device, profile=None, source_id="") -> dict:
2278 """Return every Device field and relation this module can overwrite."""
2279 state = {
2280 "name": device.name,
2281 "device_type_id": device.device_type_id,
2282 "role_id": device.role_id,
2283 "site_id": device.site_id,
2284 "location_id": device.location_id,
2285 "rack_id": device.rack_id,
2286 "position": normalize_for_compare(device.position),
2287 "face": device.face or "",
2288 "status": device.status,
2289 "tenant_id": device.tenant_id,
2290 "airflow": device.airflow or "",
2291 "serial": device.serial or "",
2292 "asset_tag": device.asset_tag or "",
2293 "primary_ip4_id": device.primary_ip4_id,
2294 "primary_ip6_id": device.primary_ip6_id,
2295 "oob_ip_id": device.oob_ip_id,
2296 }
2297 if profile is None:
2298 return state
2299 custom_field = profile.adapter_settings.custom_field_name
2300 state["source_id_custom_field"] = (
2301 custom_field,
2302 device.custom_field_data.get(custom_field) if custom_field else None,
2303 )
2304 from .models import DeviceImportSource
2306 stored = DeviceImportSource.objects.filter(device_id=device.pk).first()
2307 state["provenance"] = (
2308 {
2309 "profile_id": stored.profile_id,
2310 "source_id": stored.source_id,
2311 "extra_columns": stored.extra_columns,
2312 "unassigned_ips": stored.unassigned_ips,
2313 }
2314 if stored is not None
2315 else None
2316 )
2317 state["source_binding"] = tuple(
2318 tuple(values)
2319 for values in profile.device_matches.filter(source_id=source_id).values_list(
2320 "source_id", "netbox_device_id", "device_name", "source_asset_tag"
2321 )
2322 )
2323 state["device_bindings"] = tuple(
2324 tuple(values)
2325 for values in profile.device_matches.filter(netbox_device_id=device.pk).values_list(
2326 "source_id", "netbox_device_id", "device_name", "source_asset_tag"
2327 )
2328 )
2329 return state
2331 @staticmethod
2332 def _create_identity_conflict(payload, profile) -> str:
2333 """Return the first target identity that would turn a planned create into a duplicate."""
2334 from dcim.models import Device
2336 source_id = payload.get("source_id") or ""
2337 if source_id and profile.device_matches.filter(source_id=source_id).exists():
2338 return f"A Device link for source ID '{source_id}'"
2339 if (
2340 source_id
2341 and Device.objects.filter(
2342 data_import_source__profile=profile,
2343 data_import_source__source_id=source_id,
2344 ).exists()
2345 ):
2346 return f"A Device with stored source ID '{source_id}'"
2347 serial = payload.get("serial") or ""
2348 if serial and Device.objects.filter(serial=serial).exists():
2349 return f"A Device with serial '{serial}'"
2350 asset_tag = payload.get("asset_tag") or ""
2351 if asset_tag and Device.objects.filter(identity_in("asset_tag", [identity_text(asset_tag)])).exists():
2352 return f"A Device with asset tag '{asset_tag}'"
2353 tenant_filter = (
2354 {"tenant_id": payload["tenant_id"]} if payload.get("tenant_id") is not None else {"tenant__isnull": True}
2355 )
2356 if (
2357 Device.objects.filter(site_id=payload["site_id"], **tenant_filter)
2358 .filter(identity_in("name", [identity_text(payload["name"])]))
2359 .exists()
2360 ):
2361 return f"A Device named '{payload['name']}' at the target site and tenant"
2362 return ""
2365MODULE_RUNTIMES: dict[str, Any] = {
2366 RackModule.key: RackModule(),
2367 DeviceModule.key: DeviceModule(),
2368 CableModule.key: CableModule(),
2369}
2372def runtime_for(key: str) -> Any | None:
2373 """Return the Target Module runtime registered under *key*, or None."""
2374 return MODULE_RUNTIMES.get(key)
2377__all__ = (
2378 "DEFAULT_RACK_HEIGHT",
2379 "MODULE_RUNTIMES",
2380 "CableModule",
2381 "DeviceModule",
2382 "ExecutionContext",
2383 "PreconditionFailed",
2384 "RackModule",
2385 "TargetModuleRuntime",
2386 "rack_duplicate_keys",
2387 "rack_row_name",
2388 "rack_row_rejection",
2389 "rack_unit_identity",
2390 "runtime_for",
2391)