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