Coverage for netbox_data_import/import_engine.py: 100%
186 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 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 typing import cast
15from django.core.exceptions import ValidationError
16from django.db import DatabaseError
18from . import adapter_config, adapters, catalog, target_modules
19from .models import FailureReason, ImportExecution, SourceDocument, locked_profile_policy
20from .netbox_reader import NetBoxReader, PlanningTargetUnavailable
21from .object_permissions import ObjectPermissionDenied
22from .plan import Diagnostic, Disposition, ImportPlan, PlanInvalid, Severity, executable_units, merge_changes
23from .source_resolution import derive_effective_rows
24from .target_runtime import DeletedObject, ExecutionContext, PreconditionFailed
27_RESOLUTION_SECTION = "source_resolutions"
30class EngineConfigurationError(Exception):
31 """The release lacks a catalog or Target Module runtime needed by an accepted plan."""
34def _resolution_section():
35 """Return the policy section that declares where a saved Source Resolution applies."""
36 section = catalog.policy_section(_RESOLUTION_SECTION)
37 if section is None:
38 raise EngineConfigurationError(f"The catalog declares no '{_RESOLUTION_SECTION}' policy section.")
39 return section
42def operator_failure_message(exc) -> str:
43 """Return one execution failure as the text the operator reads.
45 A database message names the table, the column and the constraint that refused the write, so it
46 stays in the log. Every other exception here carries a message this plugin or NetBox wrote.
47 """
48 if isinstance(exc, DatabaseError):
49 return "The import could not be written. Check the NetBox logs and try again."
50 return "; ".join(exc.messages) if isinstance(exc, ValidationError) else str(exc)
53class StaleSourceDocument(Exception):
54 """The referenced stored source no longer exists, so the operator has to upload it again."""
57class StalePlan(Exception):
58 """A selected unit no longer has the decision inputs the operator accepted."""
61class SelectionError(Exception):
62 """The requested Synchronization Unit selection is not executable as stated."""
65class _ExecutionFailed(Exception):
66 """Carry one apply failure out of the transaction, with what rolled back behind it."""
68 def __init__(self, *, cause, reason, failed_change, rolled_back, not_attempted):
69 super().__init__(str(cause))
70 self.cause = cause
71 self.reason = reason
72 self.failed_change = failed_change
73 self.rolled_back = list(rolled_back)
74 self.not_attempted = list(not_attempted)
77class ImportEngine:
78 """Coordinate source interpretation and Target Module planning."""
80 @classmethod
81 def plan(
82 cls, profile, source_document, actor, planning_context, *, lock_plan_references: bool = False
83 ) -> ImportPlan:
84 """Return the plan and optionally lock read-only target references for execution.
86 The caller must open a transaction before it requests locks.
87 """
88 document = cls._stored_source(profile, source_document)
89 adapter = adapters.get_adapter(profile.source_adapter)
90 if adapter is None:
91 raise adapters.UnknownSourceAdapter(
92 f"This release does not register the source adapter '{profile.source_adapter}'."
93 )
94 if not catalog.has_implemented_module(adapter.output_kinds):
95 raise adapters.UnknownSourceAdapter(
96 f"This release has no Target Module for source adapter '{profile.source_adapter}'."
97 )
98 source_batch = adapter.interpret(
99 bytes(document.content),
100 adapter_config.interpreter_config_for(profile),
101 collect_unused=True,
102 )
103 # The catalog already declares which output kinds the resolution policy applies to.
104 if _resolution_section().applies_to(source_batch.output_kinds):
105 # The section applies only to the flat output kinds, whose rows are always dictionaries.
106 source_rows = cast(tuple[dict, ...], source_batch.rows)
107 source_batch = adapters.SourceBatch(
108 output_kinds=source_batch.output_kinds,
109 rows=tuple(derive_effective_rows(list(source_rows), profile)),
110 diagnostics=source_batch.diagnostics,
111 unused_columns=source_batch.unused_columns,
112 )
113 reader = NetBoxReader.for_actor(actor).for_planning_context(planning_context)
115 units = []
116 for declaration in catalog.TARGET_MODULES:
117 if not declaration.consumes & source_batch.output_kinds:
118 continue
119 runtime = target_modules.runtime_for(declaration.key)
120 if runtime is not None:
121 units.extend(
122 runtime.plan(
123 source_batch,
124 profile,
125 catalog.CATALOG,
126 reader,
127 lock_plan_references=lock_plan_references,
128 )
129 )
131 # Section 4.4 makes the merged graph the coordinator's, so a bad one fails here, not at a write.
132 merge_changes(executable_units(units))
133 return ImportPlan(
134 units=tuple(units),
135 diagnostics=(
136 *(cls._source_diagnostic(item) for item in source_batch.diagnostics),
137 *(
138 cls._unused_column_diagnostic(adapter.key, name, stats)
139 for name, stats in source_batch.unused_columns.items()
140 ),
141 ),
142 source_fingerprint=document.content_fingerprint,
143 profile_fingerprint=profile.planning_fingerprint,
144 actor=str(actor.pk),
145 planning_context=planning_context,
146 )
148 @classmethod
149 def execute(
150 cls,
151 profile,
152 source_document,
153 accepted_plan,
154 selection,
155 idempotency_key,
156 actor,
157 *,
158 job=None,
159 progress_callback=None,
160 ) -> ImportExecution:
161 """Apply selected units from one accepted serialized plan and return their audit row."""
162 accepted = ImportPlan.from_dict(accepted_plan)
163 selected_identities = tuple(selection)
164 if not selected_identities:
165 raise SelectionError("An execution needs at least one Synchronization Unit.")
166 execution, created = ImportExecution.reserve(
167 profile=profile,
168 source_document=source_document,
169 actor=actor,
170 idempotency_key=idempotency_key,
171 plan_schema_version=accepted.schema_version,
172 accepted_plan_fingerprint=accepted.fingerprint,
173 selected_units=list(selected_identities),
174 input_filename=source_document.filename,
175 )
176 if not created:
177 return execution
179 try:
180 if job is not None:
181 execution.link_job(job)
182 target = NetBoxReader.for_actor(actor).for_planning_context(accepted.planning_context)
183 execution.site_name = str(target.site)
184 execution.save(update_fields=["site_name"])
185 # One lock over the replan, the comparison and the writes: policy cannot move between them.
186 with locked_profile_policy(profile.pk):
187 profile.refresh_from_db()
188 cls._write_selection(
189 execution,
190 profile,
191 source_document,
192 accepted,
193 selected_identities,
194 actor,
195 progress_callback,
196 )
197 except _ExecutionFailed as failure:
198 cls._mark_failed(
199 execution,
200 reason=failure.reason,
201 failed_change=failure.failed_change,
202 rolled_back=failure.rolled_back,
203 not_attempted=failure.not_attempted,
204 )
205 raise failure.cause from failure
206 except Exception as exc:
207 # Every other failure after the reservation, so the row can never stay pending.
208 cls._mark_failed(
209 execution,
210 reason=cls._failure_reason(exc),
211 not_attempted=cls._selected_change_identities(accepted, selected_identities),
212 )
213 raise
214 return execution
216 @classmethod
217 def _write_selection(
218 cls,
219 execution,
220 profile,
221 source_document,
222 accepted,
223 selected_identities,
224 actor,
225 progress_callback,
226 ) -> None:
227 """Compare the selection against a fresh plan and apply it, inside the caller's transaction."""
228 current = cls.plan(
229 profile,
230 source_document,
231 actor,
232 accepted.planning_context,
233 lock_plan_references=True,
234 )
235 units = cls._selected_units(accepted, current, selected_identities)
236 try:
237 changes = merge_changes(units)
238 except PlanInvalid as exc:
239 raise SelectionError(str(exc)) from exc
240 total = len(units) + len(changes)
241 if progress_callback is not None:
242 progress_callback(0, total)
243 progress_callback(len(units), total)
244 context = ExecutionContext(
245 actor=actor,
246 reader=NetBoxReader.for_actor(actor).for_planning_context(accepted.planning_context),
247 profile=profile,
248 )
249 completed: list[str] = []
250 deleted: list[dict] = []
251 for index, change in enumerate(changes):
252 runtime = target_modules.runtime_for(change.target_module)
253 if runtime is None:
254 raise EngineConfigurationError(f"No Target Module runtime is registered for '{change.target_module}'.")
255 try:
256 applied = runtime.apply(change, context)
257 except (PreconditionFailed, ObjectPermissionDenied, ValidationError, DatabaseError) as exc:
258 raise _ExecutionFailed(
259 cause=exc,
260 reason=cls._failure_reason(exc),
261 failed_change=change.identity,
262 rolled_back=completed,
263 not_attempted=[later.identity for later in changes[index + 1 :]],
264 ) from exc
265 completed.append(change.identity)
266 if isinstance(applied, DeletedObject):
267 deleted.append(applied.to_dict())
268 if progress_callback is not None:
269 progress_callback(len(units) + index + 1, total)
270 execution.mark_succeeded(
271 applied_changes={"changes": completed, "deleted": deleted},
272 result_counts=cls._result_counts(changes),
273 )
275 @staticmethod
276 def _result_counts(changes) -> dict:
277 """Count successful primary creates by Target Module for audit display."""
278 created: dict[str, int] = {}
279 for change in changes:
280 if change.operation == "create":
281 created[change.target_module] = created.get(change.target_module, 0) + 1
282 return {"created": created, "errors": 0}
284 @staticmethod
285 def _failure_reason(exc) -> str:
286 """Return the typed audit reason for one failure a Target Module can raise."""
287 if isinstance(exc, StalePlan):
288 return FailureReason.STALE_PLAN
289 if isinstance(exc, SelectionError):
290 return FailureReason.SELECTION
291 if isinstance(exc, (StaleSourceDocument, PlanningTargetUnavailable)):
292 return FailureReason.PLANNING
293 if isinstance(exc, PreconditionFailed):
294 return FailureReason.PRECONDITION
295 if isinstance(exc, ObjectPermissionDenied):
296 return FailureReason.PERMISSION
297 if isinstance(exc, ValidationError):
298 return FailureReason.VALIDATION
299 if isinstance(exc, DatabaseError):
300 return FailureReason.DATABASE
301 return FailureReason.PLANNING
303 @staticmethod
304 def _selected_units(accepted, current, selected_identities):
305 """Return the current actionable units whose accepted fingerprints still match."""
306 if len(set(selected_identities)) != len(selected_identities):
307 raise SelectionError("A Synchronization Unit can be selected only once.")
308 selected = []
309 for identity in selected_identities:
310 accepted_unit = accepted.unit(identity)
311 current_unit = current.unit(identity)
312 if accepted_unit is None or current_unit is None:
313 raise SelectionError(f"The current Import Plan does not carry selected unit '{identity}'.")
314 if accepted_unit.disposition != Disposition.ACTIONABLE:
315 raise SelectionError(f"Synchronization Unit '{identity}' was not actionable when selected.")
316 if current_unit.disposition != Disposition.ACTIONABLE:
317 raise StalePlan(f"Synchronization Unit '{identity}' is no longer actionable.")
318 if accepted.unit_fingerprint(identity) != current.unit_fingerprint(identity):
319 raise StalePlan(f"Synchronization Unit '{identity}' changed after the plan was accepted.")
320 selected.append(current_unit)
321 return tuple(selected)
323 @staticmethod
324 def _selected_change_identities(plan, selected_identities) -> list[str]:
325 """Return unique Planned Change identities carried by the selected units."""
326 identities = []
327 for unit_identity in selected_identities:
328 unit = plan.unit(unit_identity)
329 if unit is None:
330 continue
331 for change in unit.changes:
332 if change.identity not in identities:
333 identities.append(change.identity)
334 return identities
336 @staticmethod
337 def _mark_failed(execution, **detail) -> None:
338 """Finish a failed row unless a concurrent finisher already chose its outcome."""
339 with suppress(ValueError):
340 execution.mark_failed(**detail)
342 @staticmethod
343 def _stored_source(profile, source_document) -> SourceDocument:
344 """Return the stored bytes this profile may plan, refusing a reference it does not own."""
345 document = SourceDocument.objects.filter(pk=source_document.pk).first()
346 if document is None:
347 raise StaleSourceDocument("The stored source no longer exists. Upload it again.")
348 if document.profile_id != profile.pk:
349 raise StaleSourceDocument(
350 f"Source document {document.pk} belongs to another Import Profile, so this one cannot plan it."
351 )
352 return document
354 @staticmethod
355 def _source_diagnostic(diagnostic) -> Diagnostic:
356 """Return one adapter diagnostic in the plan vocabulary, under the adapter's own namespace.
358 The code passes through: a Source Adapter names its own domain, and prefixing a second one
359 would push a namespaced code past what a plan diagnostic code accepts.
360 """
361 display = {"message": diagnostic.message}
362 if diagnostic.row_number is not None:
363 display["row_number"] = diagnostic.row_number
364 return Diagnostic(code=diagnostic.code, severity=Severity.WARNING, display=display)
366 @staticmethod
367 def _unused_column_diagnostic(adapter_key, name, stats) -> Diagnostic:
368 """Carry one unmapped source column as display-only review information."""
369 return Diagnostic(
370 code=f"{adapter_key}.unused_column",
371 severity=Severity.INFO,
372 display={
373 "name": str(name),
374 "count": int((stats or {}).get("count", 0)),
375 "samples": [str(value) for value in (stats or {}).get("samples", ())],
376 },
377 )
380__all__ = (
381 "EngineConfigurationError",
382 "ImportEngine",
383 "PreconditionFailed",
384 "SelectionError",
385 "StalePlan",
386 "StaleSourceDocument",
387)