Coverage for netbox_data_import/jobs.py: 98%
90 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# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3"""Native NetBox background jobs for data imports."""
5import logging
7from typing import NoReturn
9from django.core.exceptions import ValidationError
10from django.db import DatabaseError
11from rq import get_current_job
13from core.exceptions import JobFailed
14from netbox.jobs import JobRunner, system_job
16from .adapters import SourceUnreadable, UnknownSourceAdapter
17from .import_engine import (
18 EngineConfigurationError,
19 ImportEngine,
20 PreconditionFailed,
21 SelectionError,
22 StalePlan,
23 StaleSourceDocument,
24 operator_failure_message,
25)
26from .models import ExecutionOutcome, ImportExecution, ImportProfile, SourceDocument, validate_registered_adapter
27from .netbox_reader import PlanningTargetUnavailable
28from .object_permissions import ObjectPermissionDenied
29from .plan import PlanError
32_PROGRESS_REPORT_INTERVAL = 25
33logger = logging.getLogger(__name__)
36class ImportJobRunner(JobRunner):
37 """Validate and execute one import while publishing row progress to RQ."""
39 job_type = "netbox_data_import.import"
41 class Meta:
42 name = "Data Import"
44 def _save_data(self, **values):
45 """Merge values into the native Job data."""
46 self.job.data = {**(self.job.data or {}), **values}
47 self.job.save(update_fields=["data"])
49 def _fail(self, message) -> NoReturn:
50 """Record a recoverable failure and stop the native Job."""
51 values = {"phase": "failed", "message": message}
52 execution = ImportExecution.objects.filter(job=self.job).first()
53 if execution is not None:
54 values["import_execution_id"] = execution.pk
55 self._save_data(**values)
56 raise JobFailed
58 @staticmethod
59 def _publish_progress(processed, total):
60 """Publish progress outside the database transaction through RQ metadata."""
61 if processed not in (0, total) and processed % _PROGRESS_REPORT_INTERVAL:
62 return
63 rq_job = get_current_job()
64 if rq_job is None:
65 return
66 rq_job.meta.update({"processed": processed, "total": total, "phase": "importing"})
67 rq_job.save_meta()
69 def run(self, profile_id, source_document_id, accepted_plan, selection, idempotency_key):
70 """Execute one accepted Import Plan as the Job's actor."""
71 user = self.job.user
72 if user is None:
73 self._fail("The user who started this import is no longer available.")
75 profile = ImportProfile.objects.restrict(user, "change").filter(pk=profile_id).first()
76 if profile is None:
77 self._fail("The import profile is no longer available.")
78 try:
79 validate_registered_adapter(profile)
80 except ValidationError as exc:
81 self._fail(operator_failure_message(exc))
82 source_document = SourceDocument.objects.filter(pk=source_document_id, profile=profile).first()
83 if source_document is None:
84 self._fail("The stored source is no longer available. Upload it again.")
86 self._save_data(phase="validating")
87 progress = {"processed": 0, "total": 0}
89 def publish_progress(processed, total):
90 """Remember final progress and publish the bounded RQ updates."""
91 progress.update(processed=processed, total=total)
92 self._publish_progress(processed, total)
94 try:
95 execution = ImportEngine.execute(
96 profile,
97 source_document,
98 accepted_plan,
99 selection,
100 idempotency_key,
101 user,
102 job=self.job,
103 progress_callback=publish_progress,
104 )
105 except ImportProfile.DoesNotExist:
106 self._fail("The import profile is no longer available.")
107 except DatabaseError as exc:
108 logger.exception("Import execution failed with a database error")
109 self._fail(operator_failure_message(exc))
110 except (
111 EngineConfigurationError,
112 ObjectPermissionDenied,
113 PlanError,
114 PlanningTargetUnavailable,
115 PreconditionFailed,
116 SelectionError,
117 SourceUnreadable,
118 StalePlan,
119 StaleSourceDocument,
120 UnknownSourceAdapter,
121 ValidationError,
122 ) as exc:
123 self._fail(operator_failure_message(exc))
124 if execution.outcome != ExecutionOutcome.SUCCEEDED:
125 reason = (execution.failure_detail or {}).get("reason") or execution.outcome or "unknown"
126 self._fail(f"The accepted import execution did not succeed ({reason}).")
127 self._save_data(
128 phase="completed",
129 processed=progress["processed"],
130 total=progress["total"],
131 import_execution_id=execution.pk,
132 )
135@system_job(interval=60 * 24)
136class SourceDocumentRetentionJob(JobRunner):
137 """Reclaim stored uploads no Import Execution references (section 9.1)."""
139 class Meta:
140 name = "Data Import source document retention"
142 @staticmethod
143 def purge() -> int:
144 """Apply the retention rules and return the number of deleted documents."""
145 return SourceDocument.purge_unreferenced()
147 def run(self, *args, **kwargs):
148 """Run one retention pass."""
149 return self.purge()
152class InferenceBackendConnectionTestJob(JobRunner):
153 """Resolve the named backend's credential on the worker, so the web process never holds one.
155 The row ID selects the backend to test, enabled or not: an operator tests a backend to decide whether to
156 enable it (specification 8.6).
157 """
159 job_type = "netbox_data_import.inference_connection_test"
161 class Meta:
162 name = "AI backend connection test"
164 def run(self, pk, backend_key, *args, **kwargs):
165 """Select by pk alone to prevent redirection to a recreated row; backend_key is display text only."""
166 from .inference_connection_test import run_connection_test
168 result = run_connection_test(pk, backend_key)
169 self.job.data = {**(self.job.data or {}), **result.as_dict()}
170 self.job.save(update_fields=["data"])
171 return result.category
174__all__ = ("ImportJobRunner", "InferenceBackendConnectionTestJob", "SourceDocumentRetentionJob")