Coverage for netbox_data_import/api/serializers.py: 96.20%
158 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# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3"""DRF serializers for the data-import plugin models."""
5import json
7from django.core.exceptions import ValidationError
8from netbox.api.serializers import NetBoxModelSerializer, ValidatedModelSerializer
9from rest_framework import serializers
11from ..adapters import DEFAULT_ADAPTER_KEY, get_adapter, selectable_adapter_choices
12from ..catalog import CATALOG
13from ..forms import InferenceBackendForm
14from ..models import (
15 ImportProfile,
16 CableClassMapping,
17 CableImportSource,
18 ColumnMapping,
19 ClassRoleMapping,
20 DeviceTypeMapping,
21 IgnoredDevice,
22 ColumnTransformRule,
23 SourceResolution,
24 ImportExecution,
25 InferenceBackend,
26 PortClassMapping,
27 ResolutionProposal,
28 validate_adapter_target_module,
29 validate_contact_candidate_resolution,
30 validate_section_applicability,
31)
34def _validate_candidate_values(candidate_values):
35 """Require a JSON object at the Contact candidate boundary."""
36 if not isinstance(candidate_values, dict):
37 raise TypeError("Contact candidate values must be a JSON object.")
40class PolicySectionApplicabilityMixin:
41 """Apply the shared policy-section applicability rule on the REST write path."""
43 def validate_policy_section(self, attrs):
44 """Reject a row whose section does not apply to the profile's Source Adapter."""
45 profile = attrs.get("profile", getattr(self.instance, "profile", None))
46 try:
47 validate_section_applicability(profile, self.Meta.model.POLICY_SECTION)
48 except ValidationError as exc:
49 raise serializers.ValidationError({"profile": exc.messages}) from exc
52class PolicySectionSerializer(PolicySectionApplicabilityMixin, ValidatedModelSerializer):
53 """Base for the policy-section serializers, which are backed by plain Django models."""
55 # No Meta.fields below lists display_url: these rows have no UI detail route to reverse.
57 def validate(self, attrs):
58 """Run the policy checks, then the NetBox model validation."""
59 # A nested serializer is handed a resolved instance, so no field mapping exists to check.
60 if getattr(self, "nested", False):
61 return super().validate(attrs)
62 # Ahead of super(), whose full_clean() reports the same rules as non-field errors.
63 self.validate_policy_section(attrs)
64 self.validate_policy_row(attrs)
65 return super().validate(attrs)
67 def validate_policy_row(self, attrs):
68 """Check the fields this model resolves through the catalog. Subclasses override."""
70 def model_cleaned_values(self):
71 """Return request values after applying the model's canonical normalization."""
72 values = dict(self.validated_data)
73 if self.instance is None:
74 instance = self.Meta.model(**values)
75 instance.full_clean(validate_unique=False)
76 else:
77 # ValidatedModelSerializer.validate() applies request values and full_clean() to this instance.
78 instance = self.instance
79 return {name: getattr(instance, name) for name in values}
82def _validate_target_keys(instance, attrs, names, *, allow_candidates=True, required=False):
83 """Reject a target key the profile's Source Adapter cannot supply."""
84 profile = attrs.get("profile", getattr(instance, "profile", None))
85 output_kinds = profile.output_kinds if profile is not None else None
86 for name in names:
87 value = attrs.get(name, getattr(instance, name, None)) or ""
88 if not value and not required:
89 continue
90 if not CATALOG.is_valid(value, output_kinds=output_kinds, allow_candidates=allow_candidates):
91 raise serializers.ValidationError({name: CATALOG.invalid_key_message(value)})
94class ImportProfileSerializer(NetBoxModelSerializer):
95 """Full serializer for ImportProfile (NetBoxModel)."""
97 url = serializers.HyperlinkedIdentityField(
98 view_name="plugins-api:netbox_data_import-api:importprofile-detail",
99 )
101 class Meta:
102 model = ImportProfile
103 brief_fields = ["id", "url", "display", "name", "description"]
104 fields = [
105 "id",
106 "url",
107 "display",
108 "name",
109 "description",
110 "source_adapter",
111 "adapter_config",
112 "tags",
113 "custom_fields",
114 "created",
115 "last_updated",
116 ]
118 def __init__(self, *args, **kwargs):
119 """Offer only the runnable adapters on create, so the schema states what REST accepts."""
120 super().__init__(*args, **kwargs)
121 # An update keeps the full registry, so an existing trace profile still round-trips.
122 if self.instance is None:
123 self.fields["source_adapter"].choices = selectable_adapter_choices()
125 def validate(self, attrs):
126 """Validate the adapter configuration and keep the Source Adapter immutable."""
127 instance = self.instance
128 adapter_key = attrs.get("source_adapter") or getattr(instance, "source_adapter", DEFAULT_ADAPTER_KEY)
129 if instance is not None and "source_adapter" in attrs and attrs["source_adapter"] != instance.source_adapter:
130 raise serializers.ValidationError(
131 {"source_adapter": "The source adapter cannot change after the profile is created."}
132 )
133 adapter = get_adapter(adapter_key)
134 if adapter is None:
135 raise serializers.ValidationError({"source_adapter": f"Unknown source adapter '{adapter_key}'."})
136 if instance is None:
137 try:
138 validate_adapter_target_module(adapter_key)
139 except ValidationError as exc:
140 raise serializers.ValidationError(exc.message_dict) from exc
141 # Normalize unconditionally: this serializer never calls Model.full_clean, so an absent key
142 # would otherwise persist {} while the form path persists the full mapping.
143 raw_config = attrs.get("adapter_config", getattr(instance, "adapter_config", None))
144 try:
145 attrs["adapter_config"] = adapter.config_form_class().validate_config(raw_config)
146 except ValidationError as exc:
147 raise serializers.ValidationError({"adapter_config": exc.messages}) from exc
148 return attrs
151class ColumnMappingSerializer(PolicySectionSerializer):
152 """Serializer for ColumnMapping (plain model)."""
154 class Meta:
155 model = ColumnMapping
156 fields = ["id", "url", "display", "profile", "source_column", "target_field"]
158 def validate_policy_row(self, attrs):
159 """Resolve the target field through the catalog."""
160 _validate_target_keys(self.instance, attrs, ("target_field",), required=True)
163class CableClassMappingSerializer(PolicySectionSerializer):
164 """Serializer for CableClassMapping policy decisions."""
166 class Meta:
167 model = CableClassMapping
168 fields = [
169 "id",
170 "url",
171 "display",
172 "profile",
173 "cable_class",
174 "cable_type_resolved",
175 "cable_type",
176 "cable_profile_resolved",
177 "cable_profile",
178 ]
181class PortClassMappingSerializer(PolicySectionSerializer):
182 """Serializer for PortClassMapping policy decisions."""
184 class Meta:
185 model = PortClassMapping
186 fields = ["id", "url", "display", "profile", "source_label", "port_class"]
189class _RackTypeSlugField(serializers.SlugRelatedField):
190 """SlugRelatedField for RackType that defers the queryset import."""
192 def get_queryset(self):
193 from dcim.models import RackType
195 return RackType.objects.all()
198class ClassRoleMappingSerializer(PolicySectionSerializer):
199 """Serializer for ClassRoleMapping (plain model)."""
201 rack_type = _RackTypeSlugField(slug_field="slug", allow_null=True, required=False)
203 class Meta:
204 model = ClassRoleMapping
205 fields = ["id", "url", "display", "profile", "source_class", "creates_rack", "rack_type", "role_slug", "ignore"]
208class DeviceTypeMappingSerializer(PolicySectionSerializer):
209 """Serializer for DeviceTypeMapping (plain model)."""
211 class Meta:
212 model = DeviceTypeMapping
213 fields = [
214 "id",
215 "url",
216 "display",
217 "profile",
218 "source_make",
219 "source_model",
220 "netbox_manufacturer_slug",
221 "netbox_device_type_slug",
222 ]
225class IgnoredDeviceSerializer(PolicySectionSerializer):
226 """Serializer for IgnoredDevice (plain model)."""
228 class Meta:
229 model = IgnoredDevice
230 fields = ["id", "url", "display", "profile", "source_id", "device_name"]
233class ColumnTransformRuleSerializer(PolicySectionSerializer):
234 """Serializer for ColumnTransformRule (plain model)."""
236 class Meta:
237 model = ColumnTransformRule
238 fields = [
239 "id",
240 "url",
241 "display",
242 "profile",
243 "source_column",
244 "pattern",
245 "group_1_target",
246 "group_2_target",
247 ]
249 def validate_policy_row(self, attrs):
250 """Resolve both group targets through the catalog, excluding the candidate targets."""
251 _validate_target_keys(self.instance, attrs, ("group_1_target", "group_2_target"), allow_candidates=False)
254class SourceResolutionSerializer(PolicySectionSerializer):
255 """Serializer for SourceResolution (rerere, plain model)."""
257 def validate_profile(self, value):
258 """Refuse to move a saved row: its source ID and column only mean anything in one profile."""
259 # Field validators run before ValidatedModelSerializer.validate() writes onto the instance.
260 if self.instance is not None and value.pk != self.instance.profile_id:
261 raise serializers.ValidationError("A saved resolution cannot move to another profile.")
262 return value
264 def validate_policy_row(self, attrs):
265 """Reject Contact candidate resolutions that the importer cannot apply."""
266 instance = self.instance
267 source_column = attrs.get("source_column", getattr(instance, "source_column", None))
268 if source_column == "candidate:contact":
269 profile = attrs.get("profile", getattr(instance, "profile", None))
270 original_value = attrs.get("original_value", getattr(instance, "original_value", None))
271 resolved_fields = attrs.get("resolved_fields", getattr(instance, "resolved_fields", None))
272 try:
273 candidate_values = json.loads(original_value)
274 _validate_candidate_values(candidate_values)
275 configured_sources = profile.column_mappings.filter(target_field="candidate:contact").values_list(
276 "source_column", flat=True
277 )
278 validate_contact_candidate_resolution(
279 resolved_fields,
280 profile.adapter_settings.primary_contact_lookup_field,
281 set(candidate_values) & set(configured_sources),
282 )
283 except (TypeError, ValueError, json.JSONDecodeError):
284 raise serializers.ValidationError(
285 {"original_value": "Enter the Contact candidate values as a JSON object."}
286 ) from None
287 except ValidationError as exc:
288 raise serializers.ValidationError({"resolved_fields": exc.messages}) from exc
290 class Meta:
291 model = SourceResolution
292 fields = [
293 "id",
294 "url",
295 "display",
296 "profile",
297 "source_id",
298 "source_column",
299 "original_value",
300 "resolved_fields",
301 ]
304class ImportExecutionSerializer(serializers.ModelSerializer):
305 """Read-only serializer for the Import Execution audit record."""
307 class Meta:
308 model = ImportExecution
309 fields = [
310 "id",
311 "profile",
312 "created",
313 "input_filename",
314 "site_name",
315 "result_counts",
316 "outcome",
317 "idempotency_key",
318 "accepted_plan_fingerprint",
319 "selected_units",
320 "applied_changes",
321 "failure_detail",
322 ]
323 read_only_fields = fields
326class CableImportSourceSerializer(serializers.ModelSerializer):
327 """Read-only serializer for one Cable provenance record."""
329 class Meta:
330 model = CableImportSource
331 fields = [
332 "id",
333 "cable",
334 "profile",
335 "trace_identity",
336 "segment_index",
337 "from_text",
338 "to_text",
339 "direction",
340 "workbook_fingerprint",
341 "sheet",
342 "block_ordinal",
343 "row_start",
344 "row_end",
345 "export_timestamp",
346 ]
347 read_only_fields = fields
350class ResolutionProposalSerializer(serializers.ModelSerializer):
351 """Read-only serializer for the Resolution Proposal audit record."""
353 class Meta:
354 model = ResolutionProposal
355 fields = [
356 "id",
357 "profile",
358 "task_type",
359 "field_key",
360 "status",
361 "source_evidence",
362 "resolved_device_type",
363 "resolved_device_id",
364 "prompt_version",
365 "response_schema_version",
366 "candidate_snapshot",
367 "requested_by",
368 "created",
369 "last_updated",
370 "outcome",
371 "selected_candidate_id",
372 "selected_object_type",
373 "selected_object_id",
374 "explanation",
375 "backend_metadata",
376 "response_diagnostic",
377 "failure_reason",
378 "decision",
379 "decided_by",
380 "decided_at",
381 "written_resolution",
382 ]
383 read_only_fields = fields
386class ResolutionProposalHistorySerializer(serializers.ModelSerializer):
387 """Serialize one attempt summary for profile-scoped field history."""
389 class Meta:
390 model = ResolutionProposal
391 fields = ["id", "created", "status", "outcome", "decision", "failure_reason"]
392 read_only_fields = fields
395class InferenceBackendSerializer(NetBoxModelSerializer):
396 """Serialize one Inference Backend row with a write-only credential reference.
398 NetBox resolves this class by model name to freeze a delete event payload.
399 """
401 credential_reference = serializers.JSONField(write_only=True)
403 class Meta:
404 model = InferenceBackend
405 fields = [
406 "id",
407 "url",
408 "display",
409 *InferenceBackendForm.Meta.fields,
410 "custom_fields",
411 "created",
412 "last_updated",
413 ]
414 brief_fields = ["id", "url", "display", "backend_key", "display_name", "enabled"]