Coverage for netbox_data_import/import_engine.py: 99.22%
258 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 target-neutral Import Engine coordinator.
5Section 2.3 gives the coordinator the merged dependency graph, transaction scope, idempotency and
6audit. It reaches Target Modules and Source Adapters through their registries alone, so it names no
7workbook, no column and no NetBox object type.
8"""
10from __future__ import annotations
12from contextlib import suppress
13from collections.abc import Callable
14import logging
15from typing import cast
17import psycopg
18from django.core.exceptions import ValidationError
19from django.db import DatabaseError, transaction
21from . import adapter_config, adapters, branching, catalog, target_modules
22from .models import FailureReason, ImportExecution, SourceDocument, locked_profile_policy
23from .netbox_reader import NetBoxReader, PlanningTargetUnavailable
24from .object_permissions import ObjectPermissionDenied, clear_user_permission_caches
25from .plan import (
26 Diagnostic,
27 Disposition,
28 ImportPlan,
29 PlanError,
30 PlanInvalid,
31 Severity,
32 executable_units,
33 merge_changes,
34 missing_companions,
35)
36from .source_resolution import derive_effective_rows
37from .target_runtime import DeletedObject, ExecutionContext, PreconditionFailed
38from .public_refusal import PublicRefusal
40logger = logging.getLogger(__name__)
42_RESOLUTION_SECTION = "source_resolutions"
45class EngineConfigurationError(Exception):
46 """The release lacks a catalog or Target Module runtime needed by an accepted plan."""
49def _resolution_section():
50 """Return the policy section that declares where a saved Source Resolution applies."""
51 section = catalog.policy_section(_RESOLUTION_SECTION)
52 if section is None:
53 raise EngineConfigurationError(f"The catalog declares no '{_RESOLUTION_SECTION}' policy section.")
54 return section
57# An Import Plan error names plan identities built from source data, so the operator reads a fixed sentence.
58UNREADABLE_PLAN = "The accepted Import Plan cannot be read. Re-read the preview, then sync again."
59UNMERGEABLE_SELECTION = (
60 "The selected rows cannot run on their own: they need changes from rows that are not selected, "
61 "or their changes conflict. Select those rows too, or sync the whole import. The server log names the changes."
62)
65CONCURRENT_CHANGE = "NetBox changed these objects during the sync. Sync again to retry."
66# PostgreSQL deadlock (40P01) and lock timeout (55P03): a concurrent writer, not a fault in the import.
67_LOCK_CONFLICTS = (psycopg.errors.DeadlockDetected, psycopg.errors.LockNotAvailable)
70def operator_failure_message(exc) -> str:
71 """Return one execution failure as the text the operator reads.
73 Database, plan and unexpected failure details stay in the server log. Public refusals carry
74 an authored message, while Django validation errors expose only their field messages.
75 """
76 if isinstance(exc, DatabaseError):
77 return "The import could not be written. Check the NetBox logs and try again."
78 if isinstance(exc, PlanError):
79 return UNREADABLE_PLAN
80 if isinstance(exc, ObjectPermissionDenied):
81 return "Permission denied: this action is outside your NetBox object permissions."
82 if isinstance(exc, ValidationError):
83 return "; ".join(exc.messages)
84 if isinstance(exc, PublicRefusal):
85 return exc.operator_message
86 if isinstance(exc, adapters.SourceUnreadable):
87 return "The source file cannot be read. Check the file and the import profile."
88 return "An unexpected error occurred. See server logs."
91class StaleSourceDocument(PublicRefusal):
92 """The referenced stored source no longer exists, so the operator has to upload it again."""
95class StalePlan(PublicRefusal):
96 """A selected unit no longer has the decision inputs the operator accepted."""
99class SelectionError(PublicRefusal):
100 """The requested Synchronization Unit selection is not executable as stated."""
103class _ExecutionFailed(Exception):
104 """Carry one apply failure out of the transaction, with what rolled back behind it."""
106 def __init__(self, *, cause, failed_change, rolled_back, not_attempted):
107 super().__init__(str(cause))
108 self.cause = cause
109 self.failed_change = failed_change
110 self.rolled_back = list(rolled_back)
111 self.not_attempted = list(not_attempted)
114def _unit_index(plan: ImportPlan) -> dict:
115 return {unit.identity: unit for unit in plan.units}
118def _quoted(identities) -> str:
119 return ", ".join(f"'{identity}'" for identity in identities)
122class ImportEngine:
123 """Coordinate source interpretation and Target Module planning."""
125 @classmethod
126 def plan(
127 cls, profile, source_document, actor, planning_context, *, lock_plan_references: bool = False
128 ) -> ImportPlan:
129 """Return the plan and optionally lock read-only target references for execution.
131 The caller must open a transaction before it requests locks.
132 """
133 document = cls._stored_source(profile, source_document)
134 adapter = adapters.get_adapter(profile.source_adapter)
135 if adapter is None:
136 raise adapters.UnknownSourceAdapter(
137 f"This release does not register the source adapter '{profile.source_adapter}'."
138 )
139 if not catalog.has_implemented_module(adapter.output_kinds):
140 raise adapters.UnknownSourceAdapter(
141 f"This release has no Target Module for source adapter '{profile.source_adapter}'."
142 )
143 source_batch = adapter.interpret(
144 bytes(document.content),
145 adapter_config.interpreter_config_for(profile, planning_context),
146 collect_unused=True,
147 )
148 # The catalog already declares which output kinds the resolution policy applies to.
149 if _resolution_section().applies_to(source_batch.output_kinds):
150 # The section applies only to the flat output kinds, whose rows are always dictionaries.
151 source_rows = cast(tuple[dict, ...], source_batch.rows)
152 source_batch = adapters.SourceBatch(
153 output_kinds=source_batch.output_kinds,
154 rows=tuple(derive_effective_rows(list(source_rows), profile)),
155 diagnostics=source_batch.diagnostics,
156 unused_columns=source_batch.unused_columns,
157 )
158 reader = NetBoxReader.for_actor(actor).for_planning_context(planning_context, output_kinds=profile.output_kinds)
160 units = []
161 for declaration in catalog.TARGET_MODULES:
162 if not declaration.consumes & source_batch.output_kinds:
163 continue
164 runtime = target_modules.runtime_for(declaration.key)
165 if runtime is not None:
166 units.extend(
167 runtime.plan(
168 source_batch,
169 profile,
170 catalog.CATALOG,
171 reader,
172 lock_plan_references=lock_plan_references,
173 )
174 )
176 # Section 4.4 makes the merged graph the coordinator's, so a bad one fails here, not at a write.
177 merge_changes(executable_units(units))
178 return ImportPlan(
179 units=tuple(units),
180 diagnostics=(
181 *(cls._source_diagnostic(item) for item in source_batch.diagnostics),
182 *(
183 cls._unused_column_diagnostic(adapter.key, name, stats)
184 for name, stats in source_batch.unused_columns.items()
185 ),
186 ),
187 source_fingerprint=document.content_fingerprint,
188 profile_fingerprint=profile.planning_fingerprint,
189 actor=str(actor.pk),
190 planning_context=planning_context,
191 )
193 @classmethod
194 def execute(
195 cls,
196 profile,
197 source_document,
198 accepted_plan,
199 selection,
200 idempotency_key,
201 actor,
202 *,
203 job=None,
204 progress_callback=None,
205 ) -> ImportExecution:
206 """Apply selected units from one accepted serialized plan and return their audit row."""
207 execution, _replanned = cls._execute(
208 profile,
209 source_document,
210 accepted_plan,
211 selection,
212 idempotency_key,
213 actor,
214 job=job,
215 progress_callback=progress_callback,
216 )
217 return execution
219 @classmethod
220 def execute_and_replan(
221 cls,
222 profile,
223 source_document,
224 accepted_plan,
225 selection,
226 idempotency_key,
227 actor,
228 *,
229 validate_replan: Callable[[ImportPlan], object],
230 ) -> tuple[ImportExecution, ImportPlan]:
231 """Apply a selection and plan the source again in one savepoint of the caller's transaction.
233 The audit reservation and a failure record stay outside the savepoint, so a failed write or a
234 failed replan rolls back the NetBox changes and the replan together, and the caller commits
235 the failed audit row. A duplicate key returns that row and no plan.
236 """
237 if not transaction.get_connection().in_atomic_block:
238 raise RuntimeError("A selective execution with a replan runs inside its caller's transaction.")
239 execution, replanned = cls._execute(
240 profile,
241 source_document,
242 accepted_plan,
243 selection,
244 idempotency_key,
245 actor,
246 replan=True,
247 validate_replan=validate_replan,
248 )
249 if replanned is None:
250 raise SelectionError("This execution was already requested, so it does not run again.")
251 return execution, replanned
253 @classmethod
254 def _execute(
255 cls,
256 profile,
257 source_document,
258 accepted_plan,
259 selection,
260 idempotency_key,
261 actor,
262 *,
263 job=None,
264 progress_callback=None,
265 replan=False,
266 validate_replan=None,
267 ) -> tuple[ImportExecution, ImportPlan | None]:
268 """Reserve, apply and audit one selection, optionally replanning inside the write savepoint."""
269 branching.refuse_branch()
270 accepted = ImportPlan.from_dict(accepted_plan)
271 selected_identities = tuple(selection)
272 if not selected_identities:
273 raise SelectionError("An execution needs at least one Synchronization Unit.")
274 try:
275 execution, created = ImportExecution.reserve(
276 profile=profile,
277 source_document=source_document,
278 actor=actor,
279 idempotency_key=idempotency_key,
280 plan_schema_version=accepted.schema_version,
281 accepted_plan_fingerprint=accepted.fingerprint,
282 selected_units=list(selected_identities),
283 input_filename=source_document.filename,
284 committed_by_caller=replan,
285 )
286 except DatabaseError as exc:
287 # The insert checks its profile key, which another execution's policy lock can hold.
288 refusal = cls._concurrency_refusal(exc)
289 if refusal is not None:
290 raise refusal from exc
291 raise
292 if not created:
293 return execution, None
295 replanned = None
296 try:
297 if job is not None:
298 execution.link_job(job)
299 target = NetBoxReader.for_actor(actor).for_planning_context(
300 accepted.planning_context, output_kinds=profile.output_kinds
301 )
302 execution.site_name = str(target.site)
303 execution.save(update_fields=["site_name"])
304 # One lock over the replan, the comparison and the writes: policy cannot move between them.
305 with locked_profile_policy(profile.pk):
306 profile.refresh_from_db()
307 completed = cls._write_selection(
308 execution,
309 profile,
310 source_document,
311 accepted,
312 selected_identities,
313 actor,
314 progress_callback,
315 )
316 if replan:
317 replanned = cls._replan_after_writes(
318 profile, source_document, actor, accepted, completed, validate_replan
319 )
320 except _ExecutionFailed as failure:
321 # The savepoint has already unwound, so the audit row can still be written.
322 cause = cls._concurrency_refusal(failure.cause) or failure.cause
323 cls._mark_failed(
324 execution,
325 reason=cls._failure_reason(cause),
326 failed_change=failure.failed_change,
327 rolled_back=failure.rolled_back,
328 not_attempted=failure.not_attempted,
329 )
330 raise cause from failure
331 except Exception as exc:
332 # Every other failure after the reservation, so the row can never stay pending.
333 refusal = cls._concurrency_refusal(exc)
334 cls._mark_failed(
335 execution,
336 reason=cls._failure_reason(refusal or exc),
337 not_attempted=cls._selected_change_identities(accepted, selected_identities),
338 )
339 if refusal is not None:
340 raise refusal from exc
341 raise
342 return execution, replanned
344 @staticmethod
345 def _concurrency_refusal(exc) -> StalePlan | None:
346 """Return the retryable refusal for a deadlock or lock timeout, or None for every other failure."""
347 if isinstance(exc, DatabaseError) and isinstance(exc.__cause__, _LOCK_CONFLICTS):
348 logger.warning("A concurrent NetBox write stopped the import execution.", exc_info=exc)
349 return StalePlan(CONCURRENT_CHANGE)
350 return None
352 @classmethod
353 def _replan_after_writes(cls, profile, source_document, actor, accepted, completed, validate_replan) -> ImportPlan:
354 """Plan the source against the writes just made; a failure rolls those writes back with it."""
355 clear_user_permission_caches(actor)
356 try:
357 plan = cls.plan(profile, source_document, actor, accepted.planning_context)
358 validate_replan(plan)
359 except Exception as exc:
360 raise _ExecutionFailed(
361 cause=exc,
362 failed_change=None,
363 rolled_back=completed,
364 not_attempted=[],
365 ) from exc
366 return plan
368 @classmethod
369 def _write_selection(
370 cls,
371 execution,
372 profile,
373 source_document,
374 accepted,
375 selected_identities,
376 actor,
377 progress_callback,
378 ) -> list[str]:
379 """Compare the selection against a fresh plan and apply it, inside the caller's transaction."""
380 clear_user_permission_caches(actor)
381 current = cls.plan(
382 profile,
383 source_document,
384 actor,
385 accepted.planning_context,
386 lock_plan_references=True,
387 )
388 units = cls._selected_units(accepted, current, selected_identities)
389 try:
390 changes = merge_changes(units)
391 except PlanInvalid as exc:
392 logger.warning("The selected units cannot merge into one execution order.", exc_info=True)
393 raise SelectionError(UNMERGEABLE_SELECTION) from exc
394 total = len(units) + len(changes)
395 if progress_callback is not None:
396 progress_callback(0, total)
397 progress_callback(len(units), total)
398 context = ExecutionContext(
399 actor=actor,
400 reader=NetBoxReader.for_actor(actor).for_planning_context(
401 accepted.planning_context, output_kinds=profile.output_kinds
402 ),
403 profile=profile,
404 )
405 completed: list[str] = []
406 deleted: list[dict] = []
407 for index, change in enumerate(changes):
408 runtime = target_modules.runtime_for(change.target_module)
409 if runtime is None:
410 raise EngineConfigurationError(f"No Target Module runtime is registered for '{change.target_module}'.")
411 try:
412 clear_user_permission_caches(actor)
413 applied = runtime.apply(change, context)
414 except (PreconditionFailed, ObjectPermissionDenied, ValidationError, DatabaseError) as exc:
415 raise _ExecutionFailed(
416 cause=exc,
417 failed_change=change.identity,
418 rolled_back=completed,
419 not_attempted=[later.identity for later in changes[index + 1 :]],
420 ) from exc
421 completed.append(change.identity)
422 if isinstance(applied, DeletedObject):
423 deleted.append(applied.to_dict())
424 if progress_callback is not None:
425 progress_callback(len(units) + index + 1, total)
426 execution.mark_succeeded(
427 applied_changes={"changes": completed, "deleted": deleted},
428 result_counts=cls._result_counts(changes),
429 )
430 return completed
432 @staticmethod
433 def _result_counts(changes) -> dict:
434 """Count successful primary creates by Target Module for audit display."""
435 created: dict[str, int] = {}
436 for change in changes:
437 if change.operation == "create":
438 created[change.target_module] = created.get(change.target_module, 0) + 1
439 return {"created": created, "errors": 0}
441 @staticmethod
442 def _failure_reason(exc) -> str:
443 """Return the typed audit reason for one failure a Target Module can raise."""
444 if isinstance(exc, StalePlan):
445 return FailureReason.STALE_PLAN
446 if isinstance(exc, SelectionError):
447 return FailureReason.SELECTION
448 if isinstance(exc, (StaleSourceDocument, PlanningTargetUnavailable)):
449 return FailureReason.PLANNING
450 if isinstance(exc, PreconditionFailed):
451 return FailureReason.PRECONDITION
452 if isinstance(exc, ObjectPermissionDenied):
453 return FailureReason.PERMISSION
454 if isinstance(exc, ValidationError):
455 return FailureReason.VALIDATION
456 if isinstance(exc, DatabaseError):
457 return FailureReason.DATABASE
458 return FailureReason.PLANNING
460 @staticmethod
461 def _selected_units(accepted, current, selected_identities):
462 """Return the current actionable units whose accepted fingerprints still match.
464 Each plan's `selected_with` must be honoured by the selection as submitted; it is never expanded.
465 """
466 if len(set(selected_identities)) != len(selected_identities):
467 raise SelectionError("A Synchronization Unit can be selected only once.")
468 for identity in selected_identities:
469 accepted_unit = accepted.unit(identity)
470 if accepted_unit is None or current.unit(identity) is None:
471 raise SelectionError(f"The current Import Plan does not carry selected unit '{identity}'.")
472 if accepted_unit.disposition != Disposition.ACTIONABLE:
473 raise SelectionError(f"Synchronization Unit '{identity}' was not actionable when selected.")
474 for identity, members in missing_companions(_unit_index(accepted), selected_identities).items():
475 raise SelectionError(
476 f"Synchronization Unit '{identity}' must be selected together with {_quoted(members)}."
477 )
478 for identity, members in missing_companions(_unit_index(current), selected_identities).items():
479 raise StalePlan(
480 f"Synchronization Unit '{identity}' changed after the plan was accepted: "
481 f"it now needs {_quoted(members)} selected and actionable."
482 )
483 selected = []
484 for identity in selected_identities:
485 current_unit = current.unit(identity)
486 if current_unit.disposition != Disposition.ACTIONABLE:
487 raise StalePlan(f"Synchronization Unit '{identity}' is no longer actionable.")
488 if accepted.unit_fingerprint(identity) != current.unit_fingerprint(identity):
489 raise StalePlan(f"Synchronization Unit '{identity}' changed after the plan was accepted.")
490 selected.append(current_unit)
491 return tuple(selected)
493 @staticmethod
494 def _selected_change_identities(plan, selected_identities) -> list[str]:
495 """Return unique Planned Change identities carried by the selected units."""
496 identities = []
497 for unit_identity in selected_identities:
498 unit = plan.unit(unit_identity)
499 if unit is None:
500 continue
501 for change in unit.changes:
502 if change.identity not in identities:
503 identities.append(change.identity)
504 return identities
506 @staticmethod
507 def _mark_failed(execution, **detail) -> None:
508 """Finish a failed row unless a concurrent finisher already chose its outcome."""
509 with suppress(ValueError):
510 execution.mark_failed(**detail)
512 @staticmethod
513 def _stored_source(profile, source_document) -> SourceDocument:
514 """Return the stored bytes this profile may plan, refusing a reference it does not own."""
515 document = SourceDocument.objects.filter(pk=source_document.pk).first()
516 if document is None:
517 raise StaleSourceDocument("The stored source no longer exists. Upload it again.")
518 if document.profile_id != profile.pk:
519 raise StaleSourceDocument(
520 f"Source document {document.pk} belongs to another Import Profile, so this one cannot plan it."
521 )
522 return document
524 @staticmethod
525 def _source_diagnostic(diagnostic) -> Diagnostic:
526 """Return one adapter diagnostic in the plan vocabulary, under the adapter's own namespace.
528 The code passes through: a Source Adapter names its own domain, and prefixing a second one
529 would push a namespaced code past what a plan diagnostic code accepts.
530 """
531 display = {"message": diagnostic.message}
532 if diagnostic.row_number is not None:
533 display["row_number"] = diagnostic.row_number
534 return Diagnostic(code=diagnostic.code, severity=Severity.WARNING, display=display)
536 @staticmethod
537 def _unused_column_diagnostic(adapter_key, name, stats) -> Diagnostic:
538 """Carry one unmapped source column as display-only review information."""
539 return Diagnostic(
540 code=f"{adapter_key}.unused_column",
541 severity=Severity.INFO,
542 display={
543 "name": str(name),
544 "count": int((stats or {}).get("count", 0)),
545 "samples": [str(value) for value in (stats or {}).get("samples", ())],
546 },
547 )
550__all__ = (
551 "EngineConfigurationError",
552 "ImportEngine",
553 "PreconditionFailed",
554 "SelectionError",
555 "StalePlan",
556 "StaleSourceDocument",
557)