Coverage for netbox_data_import/api/serializers.py: 94%
121 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"""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 ..models import (
14 ImportProfile,
15 ColumnMapping,
16 ClassRoleMapping,
17 DeviceTypeMapping,
18 IgnoredDevice,
19 ColumnTransformRule,
20 SourceResolution,
21 ImportExecution,
22 validate_adapter_target_module,
23 validate_contact_candidate_resolution,
24 validate_section_applicability,
25)
28def _validate_candidate_values(candidate_values):
29 """Require a JSON object at the Contact candidate boundary."""
30 if not isinstance(candidate_values, dict):
31 raise TypeError("Contact candidate values must be a JSON object.")
34class PolicySectionApplicabilityMixin:
35 """Apply the shared policy-section applicability rule on the REST write path."""
37 def validate_policy_section(self, attrs):
38 """Reject a row whose section does not apply to the profile's Source Adapter."""
39 profile = attrs.get("profile", getattr(self.instance, "profile", None))
40 try:
41 validate_section_applicability(profile, self.Meta.model.POLICY_SECTION)
42 except ValidationError as exc:
43 raise serializers.ValidationError({"profile": exc.messages}) from exc
46class PolicySectionSerializer(PolicySectionApplicabilityMixin, ValidatedModelSerializer):
47 """Base for the policy-section serializers, which are backed by plain Django models."""
49 # No Meta.fields below lists display_url: these rows have no UI detail route to reverse.
51 def validate(self, attrs):
52 """Run the policy checks, then the NetBox model validation."""
53 # A nested serializer is handed a resolved instance, so no field mapping exists to check.
54 if getattr(self, "nested", False):
55 return super().validate(attrs)
56 # Ahead of super(), whose full_clean() reports the same rules as non-field errors.
57 self.validate_policy_section(attrs)
58 self.validate_policy_row(attrs)
59 return super().validate(attrs)
61 def validate_policy_row(self, attrs):
62 """Check the fields this model resolves through the catalog. Subclasses override."""
65def _validate_target_keys(instance, attrs, names, *, allow_candidates=True, required=False):
66 """Reject a target key the profile's Source Adapter cannot supply."""
67 profile = attrs.get("profile", getattr(instance, "profile", None))
68 output_kinds = profile.output_kinds if profile is not None else None
69 for name in names:
70 value = attrs.get(name, getattr(instance, name, None)) or ""
71 if not value and not required:
72 continue
73 if not CATALOG.is_valid(value, output_kinds=output_kinds, allow_candidates=allow_candidates):
74 raise serializers.ValidationError({name: CATALOG.invalid_key_message(value)})
77class ImportProfileSerializer(NetBoxModelSerializer):
78 """Full serializer for ImportProfile (NetBoxModel)."""
80 url = serializers.HyperlinkedIdentityField(
81 view_name="plugins-api:netbox_data_import-api:importprofile-detail",
82 )
84 class Meta:
85 model = ImportProfile
86 brief_fields = ["id", "url", "display", "name", "description"]
87 fields = [
88 "id",
89 "url",
90 "display",
91 "name",
92 "description",
93 "source_adapter",
94 "adapter_config",
95 "tags",
96 "custom_fields",
97 "created",
98 "last_updated",
99 ]
101 def __init__(self, *args, **kwargs):
102 """Offer only the runnable adapters on create, so the schema states what REST accepts."""
103 super().__init__(*args, **kwargs)
104 # An update keeps the full registry, so an existing trace profile still round-trips.
105 if self.instance is None:
106 self.fields["source_adapter"].choices = selectable_adapter_choices()
108 def validate(self, attrs):
109 """Validate the adapter configuration and keep the Source Adapter immutable."""
110 instance = self.instance
111 adapter_key = attrs.get("source_adapter") or getattr(instance, "source_adapter", DEFAULT_ADAPTER_KEY)
112 if instance is not None and "source_adapter" in attrs and attrs["source_adapter"] != instance.source_adapter:
113 raise serializers.ValidationError(
114 {"source_adapter": "The source adapter cannot change after the profile is created."}
115 )
116 adapter = get_adapter(adapter_key)
117 if adapter is None:
118 raise serializers.ValidationError({"source_adapter": f"Unknown source adapter '{adapter_key}'."})
119 if instance is None:
120 try:
121 validate_adapter_target_module(adapter_key)
122 except ValidationError as exc:
123 raise serializers.ValidationError(exc.message_dict) from exc
124 # Normalize unconditionally: this serializer never calls Model.full_clean, so an absent key
125 # would otherwise persist {} while the form path persists the full mapping.
126 raw_config = attrs.get("adapter_config", getattr(instance, "adapter_config", None))
127 try:
128 attrs["adapter_config"] = adapter.config_form_class().validate_config(raw_config)
129 except ValidationError as exc:
130 raise serializers.ValidationError({"adapter_config": exc.messages}) from exc
131 return attrs
134class ColumnMappingSerializer(PolicySectionSerializer):
135 """Serializer for ColumnMapping (plain model)."""
137 class Meta:
138 model = ColumnMapping
139 fields = ["id", "url", "display", "profile", "source_column", "target_field"]
141 def validate_policy_row(self, attrs):
142 """Resolve the target field through the catalog."""
143 _validate_target_keys(self.instance, attrs, ("target_field",), required=True)
146class _RackTypeSlugField(serializers.SlugRelatedField):
147 """SlugRelatedField for RackType that defers the queryset import."""
149 def get_queryset(self):
150 from dcim.models import RackType
152 return RackType.objects.all()
155class ClassRoleMappingSerializer(PolicySectionSerializer):
156 """Serializer for ClassRoleMapping (plain model)."""
158 rack_type = _RackTypeSlugField(slug_field="slug", allow_null=True, required=False)
160 class Meta:
161 model = ClassRoleMapping
162 fields = ["id", "url", "display", "profile", "source_class", "creates_rack", "rack_type", "role_slug", "ignore"]
165class DeviceTypeMappingSerializer(PolicySectionSerializer):
166 """Serializer for DeviceTypeMapping (plain model)."""
168 class Meta:
169 model = DeviceTypeMapping
170 fields = [
171 "id",
172 "url",
173 "display",
174 "profile",
175 "source_make",
176 "source_model",
177 "netbox_manufacturer_slug",
178 "netbox_device_type_slug",
179 ]
182class IgnoredDeviceSerializer(PolicySectionSerializer):
183 """Serializer for IgnoredDevice (plain model)."""
185 class Meta:
186 model = IgnoredDevice
187 fields = ["id", "url", "display", "profile", "source_id", "device_name"]
190class ColumnTransformRuleSerializer(PolicySectionSerializer):
191 """Serializer for ColumnTransformRule (plain model)."""
193 class Meta:
194 model = ColumnTransformRule
195 fields = [
196 "id",
197 "url",
198 "display",
199 "profile",
200 "source_column",
201 "pattern",
202 "group_1_target",
203 "group_2_target",
204 ]
206 def validate_policy_row(self, attrs):
207 """Resolve both group targets through the catalog, excluding the candidate targets."""
208 _validate_target_keys(self.instance, attrs, ("group_1_target", "group_2_target"), allow_candidates=False)
211class SourceResolutionSerializer(PolicySectionSerializer):
212 """Serializer for SourceResolution (rerere, plain model)."""
214 def validate_profile(self, value):
215 """Refuse to move a saved row: its source ID and column only mean anything in one profile."""
216 # Field validators run before ValidatedModelSerializer.validate() writes onto the instance.
217 if self.instance is not None and value.pk != self.instance.profile_id:
218 raise serializers.ValidationError("A saved resolution cannot move to another profile.")
219 return value
221 def validate_policy_row(self, attrs):
222 """Reject Contact candidate resolutions that the importer cannot apply."""
223 instance = self.instance
224 source_column = attrs.get("source_column", getattr(instance, "source_column", None))
225 if source_column == "candidate:contact":
226 profile = attrs.get("profile", getattr(instance, "profile", None))
227 original_value = attrs.get("original_value", getattr(instance, "original_value", None))
228 resolved_fields = attrs.get("resolved_fields", getattr(instance, "resolved_fields", None))
229 try:
230 candidate_values = json.loads(original_value)
231 _validate_candidate_values(candidate_values)
232 configured_sources = profile.column_mappings.filter(target_field="candidate:contact").values_list(
233 "source_column", flat=True
234 )
235 validate_contact_candidate_resolution(
236 resolved_fields,
237 profile.adapter_settings.primary_contact_lookup_field,
238 set(candidate_values) & set(configured_sources),
239 )
240 except (TypeError, ValueError, json.JSONDecodeError):
241 raise serializers.ValidationError(
242 {"original_value": "Enter the Contact candidate values as a JSON object."}
243 ) from None
244 except ValidationError as exc:
245 raise serializers.ValidationError({"resolved_fields": exc.messages}) from exc
247 class Meta:
248 model = SourceResolution
249 fields = [
250 "id",
251 "url",
252 "display",
253 "profile",
254 "source_id",
255 "source_column",
256 "original_value",
257 "resolved_fields",
258 ]
261class ImportExecutionSerializer(serializers.ModelSerializer):
262 """Read-only serializer for the Import Execution audit record."""
264 class Meta:
265 model = ImportExecution
266 fields = [
267 "id",
268 "profile",
269 "created",
270 "input_filename",
271 "site_name",
272 "result_counts",
273 "outcome",
274 "idempotency_key",
275 "accepted_plan_fingerprint",
276 "selected_units",
277 "applied_changes",
278 "failure_detail",
279 ]
280 read_only_fields = fields