Coverage for netbox_data_import/catalog.py: 100.00%
130 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"""Static target-field catalog and policy applicability.
5This module is the one source of Target Field keys. Forms, REST, GraphQL, YAML, and validation
6derive their choices from it and keep no local field list. It imports nothing from the plugin and
7nothing from NetBox, so every other layer may depend on it.
8"""
10from __future__ import annotations
12from collections.abc import Iterator, Sequence
13from contextlib import contextmanager
14from contextvars import ContextVar
15from dataclasses import dataclass, field
18class OutputKind:
19 """The typed source items a Source Adapter can emit."""
21 DEVICE_SOURCE_ROW = "device_source_row"
22 RACK_SOURCE_ROW = "rack_source_row"
23 SOURCE_TRACE = "source_trace"
26class TargetModuleKey:
27 """The target domains a Target Module writes to."""
29 DEVICE = "device"
30 RACK = "rack"
31 CABLE = "cable"
34class ValueKind:
35 """How a Target Field's value is interpreted once a source supplies it."""
37 TEXT = "text"
38 INTEGER = "integer"
39 DECIMAL = "decimal"
40 CHOICE = "choice"
41 IP_ADDRESS = "ip_address"
42 CANDIDATE = "candidate"
45@dataclass(frozen=True)
46class TargetModule:
47 """A Target Module declares the adapter output kinds it consumes."""
49 key: str
50 label: str
51 consumes: frozenset[str]
52 # A declared module the release does not implement yet cannot make its adapters selectable.
53 implemented: bool
54 # A module that writes into the import Location needs it; every other module reads it as evidence.
55 writes_placement: bool
58TARGET_MODULES: tuple[TargetModule, ...] = (
59 TargetModule(
60 key=TargetModuleKey.DEVICE,
61 label="Device",
62 consumes=frozenset({OutputKind.DEVICE_SOURCE_ROW}),
63 implemented=True,
64 writes_placement=True,
65 ),
66 TargetModule(
67 key=TargetModuleKey.RACK,
68 label="Rack",
69 consumes=frozenset({OutputKind.RACK_SOURCE_ROW}),
70 implemented=True,
71 writes_placement=True,
72 ),
73 TargetModule(
74 key=TargetModuleKey.CABLE,
75 label="Cable",
76 consumes=frozenset({OutputKind.SOURCE_TRACE}),
77 implemented=True,
78 writes_placement=False,
79 ),
80)
82_MODULES_BY_KEY = {module.key: module for module in TARGET_MODULES}
85@dataclass(frozen=True)
86class TargetField:
87 """One catalog entry with a fixed key."""
89 key: str
90 label: str
91 value_kind: str
92 output_kinds: frozenset[str]
93 # A candidate target supplies review candidates rather than a written value. ColumnTransformRule
94 # excludes these because a regex capture group cannot produce a candidate bundle.
95 candidate_target: bool = False
97 @property
98 def target_modules(self) -> frozenset[str]:
99 """Return the Target Modules that consume this field, derived from its output kinds."""
100 return frozenset(m.key for m in TARGET_MODULES if m.consumes & self.output_kinds)
103@dataclass(frozen=True)
104class KeyFamily:
105 """A catalog entry whose exact key is data, not a fixed choice.
107 The key is the prefix plus a name. Every surface resolves a family key through this validator,
108 so no surface reimplements the prefix rule.
109 """
111 prefix: str
112 label: str
113 value_kind: str
114 output_kinds: frozenset[str]
115 name_label: str = "Custom field"
117 def matches(self, key: str) -> bool:
118 """Return True when *key* carries this family's prefix."""
119 return key.startswith(self.prefix)
121 def name_of(self, key: str) -> str:
122 """Return the name part of *key*, without the prefix."""
123 return key[len(self.prefix) :]
125 def is_valid(self, key: str) -> bool:
126 """Return True when *key* is a well-formed member: the prefix plus a non-empty name."""
127 return self.matches(key) and bool(self.name_of(key).strip())
129 def display(self, key: str) -> str:
130 """Return the human-readable name for a member key."""
131 return f"{self.name_label}: {self.name_of(key)}"
133 @property
134 def target_modules(self) -> frozenset[str]:
135 """Return the Target Modules that consume this family, derived from its output kinds."""
136 return frozenset(m.key for m in TARGET_MODULES if m.consumes & self.output_kinds)
139_FLAT_ROW_KINDS = frozenset({OutputKind.DEVICE_SOURCE_ROW, OutputKind.RACK_SOURCE_ROW})
140_DEVICE_ONLY = frozenset({OutputKind.DEVICE_SOURCE_ROW})
141_TRACE_ONLY = frozenset({OutputKind.SOURCE_TRACE})
143EXTRA_JSON_PREFIX = "extra_json:"
144CANDIDATE_TARGET_PREFIX = "candidate:"
147@dataclass(frozen=True)
148class TargetFieldCatalog:
149 """The static registry of Target Fields and key families."""
151 fields: tuple[TargetField, ...]
152 families: tuple[KeyFamily, ...]
153 _by_key: dict[str, TargetField] = field(init=False, repr=False, compare=False, default_factory=dict)
155 def __post_init__(self):
156 object.__setattr__(self, "_by_key", {entry.key: entry for entry in self.fields})
158 def entry(self, key: str) -> TargetField | None:
159 """Return the fixed-key entry for *key*, or None."""
160 return self._by_key.get(key)
162 def family(self, key: str) -> KeyFamily | None:
163 """Return the key family that claims *key*, or None."""
164 for candidate in self.families:
165 if candidate.matches(key):
166 return candidate
167 return None
169 def is_valid(self, key: str, *, output_kinds: frozenset[str] | None = None, allow_candidates: bool = True) -> bool:
170 """Return True when *key* is a Target Field the given output kinds can supply."""
171 entry = self.entry(key)
172 if entry is not None:
173 if entry.candidate_target and not allow_candidates:
174 return False
175 return output_kinds is None or bool(entry.output_kinds & output_kinds)
176 family = self.family(key)
177 if family is None or not family.is_valid(key):
178 return False
179 return output_kinds is None or bool(family.output_kinds & output_kinds)
181 def display(self, key: str) -> str:
182 """Return the human-readable name for any valid key, or the key itself."""
183 entry = self.entry(key)
184 if entry is not None:
185 return entry.label
186 family = self.family(key)
187 if family is not None and family.is_valid(key):
188 return family.display(key)
189 return key
191 def choices(self, *, output_kinds: frozenset[str] | None = None, allow_candidates: bool = True):
192 """Return Django choice pairs for the fixed-key entries the given output kinds can supply."""
193 return [
194 (entry.key, entry.label)
195 for entry in self.fields
196 if (allow_candidates or not entry.candidate_target)
197 and (output_kinds is None or entry.output_kinds & output_kinds)
198 ]
200 def invalid_key_message(self, key: str) -> str:
201 """Return the validation message for a rejected key."""
202 prefixes = ", ".join(f"'{f.prefix}'" for f in self.families)
203 return f"Value '{key}' is not a valid choice. Must be one of the standard field names or start with {prefixes}."
206CATALOG = TargetFieldCatalog(
207 fields=(
208 TargetField("rack_name", "Rack name", ValueKind.TEXT, _FLAT_ROW_KINDS),
209 TargetField("device_name", "Device name", ValueKind.TEXT, _FLAT_ROW_KINDS),
210 TargetField("device_class", "Device class (maps to role/rack)", ValueKind.TEXT, _FLAT_ROW_KINDS),
211 TargetField("face", "Face (Front/Back)", ValueKind.CHOICE, _DEVICE_ONLY),
212 TargetField("airflow", "Airflow", ValueKind.CHOICE, _DEVICE_ONLY),
213 TargetField("u_position", "U position", ValueKind.INTEGER, _DEVICE_ONLY),
214 TargetField("status", "Status", ValueKind.CHOICE, _DEVICE_ONLY),
215 TargetField("make", "Make (manufacturer)", ValueKind.TEXT, _DEVICE_ONLY),
216 TargetField("model", "Model (device type)", ValueKind.TEXT, _DEVICE_ONLY),
217 TargetField("u_height", "U height", ValueKind.DECIMAL, _FLAT_ROW_KINDS),
218 TargetField("serial", "Serial number", ValueKind.TEXT, _FLAT_ROW_KINDS),
219 TargetField("asset_tag", "Asset tag", ValueKind.TEXT, _DEVICE_ONLY),
220 TargetField("primary_ip4", "Primary IPv4", ValueKind.IP_ADDRESS, _DEVICE_ONLY),
221 TargetField("primary_ip6", "Primary IPv6", ValueKind.IP_ADDRESS, _DEVICE_ONLY),
222 TargetField("oob_ip", "Out-of-band IP", ValueKind.IP_ADDRESS, _DEVICE_ONLY),
223 TargetField("primary_contact", "Primary contact", ValueKind.TEXT, _DEVICE_ONLY),
224 TargetField("source_id", "Source ID (stored in custom field)", ValueKind.TEXT, _FLAT_ROW_KINDS),
225 TargetField(
226 f"{CANDIDATE_TARGET_PREFIX}contact",
227 "Candidate values: Contact fields",
228 ValueKind.CANDIDATE,
229 _DEVICE_ONLY,
230 candidate_target=True,
231 ),
232 ),
233 families=(
234 KeyFamily(
235 prefix=EXTRA_JSON_PREFIX,
236 label="Custom field",
237 value_kind=ValueKind.TEXT,
238 output_kinds=_DEVICE_ONLY,
239 ),
240 ),
241)
244@dataclass(frozen=True)
245class PolicySection:
246 """A profile policy table and the adapter output kinds it applies to."""
248 key: str
249 label: str
250 output_kinds: frozenset[str]
252 def applies_to(self, output_kinds: frozenset[str]) -> bool:
253 """Return True when an adapter emitting *output_kinds* can use this section."""
254 return bool(self.output_kinds & output_kinds)
257POLICY_SECTIONS: tuple[PolicySection, ...] = (
258 PolicySection("column_mappings", "Column Mappings", _FLAT_ROW_KINDS),
259 PolicySection("column_transform_rules", "Column Transform Rules", _FLAT_ROW_KINDS),
260 PolicySection("class_role_mappings", "Class/Role Mappings", _FLAT_ROW_KINDS),
261 PolicySection("device_type_mappings", "Device Type Mappings", _FLAT_ROW_KINDS),
262 PolicySection("manufacturer_mappings", "Manufacturer Mappings", _FLAT_ROW_KINDS),
263 PolicySection("ignored_devices", "Ignored Devices", _FLAT_ROW_KINDS),
264 PolicySection("source_resolutions", "Source Resolutions", _DEVICE_ONLY),
265 PolicySection("device_existing_matches", "Device Existing Matches", _DEVICE_ONLY),
266 PolicySection("ignored_field_differences", "Ignored Field Differences", _DEVICE_ONLY),
267 PolicySection("trace_device_resolutions", "Trace Device Resolutions", _TRACE_ONLY),
268 PolicySection("trace_location_resolutions", "Trace Location Resolutions", _TRACE_ONLY),
269 PolicySection("termination_resolutions", "Termination Resolutions", _TRACE_ONLY),
270 PolicySection("cable_class_mappings", "CableClass Mappings", _TRACE_ONLY),
271 PolicySection("port_class_mappings", "PortClass Mappings", _TRACE_ONLY),
272 PolicySection("cable_segment_overrides", "Cable Segment Overrides", _TRACE_ONLY),
273)
275_SECTIONS_BY_KEY = {section.key: section for section in POLICY_SECTIONS}
278def target_module(key: str) -> TargetModule | None:
279 """Return the Target Module declared under *key*."""
280 return _MODULES_BY_KEY.get(key)
283def policy_section(key: str) -> PolicySection | None:
284 """Return the policy section declared under *key*."""
285 return _SECTIONS_BY_KEY.get(key)
288_declared_override: ContextVar[tuple[TargetModule, ...] | None] = ContextVar("declared_modules", default=None)
291def declared_modules() -> tuple[TargetModule, ...]:
292 """Return the Target Module declarations in force."""
293 override = _declared_override.get()
294 return TARGET_MODULES if override is None else override
297@contextmanager
298def declared_modules_override(modules: Sequence[TargetModule]) -> Iterator[None]:
299 """Run the block against *modules*, so a caller can state a table this release does not ship.
301 Every entry point reaches the runtime gate through `ImportProfile.clean`, far below its own
302 signature, so a declaration table cannot be passed down to it as an argument.
303 """
304 token = _declared_override.set(tuple(modules))
305 try:
306 yield
307 finally:
308 _declared_override.reset(token)
311def consuming_modules(output_kinds: frozenset[str]) -> tuple[TargetModule, ...]:
312 """Return the Target Modules that consume any of *output_kinds*."""
313 return tuple(module for module in declared_modules() if module.consumes & output_kinds)
316def has_implemented_module(output_kinds: frozenset[str]) -> bool:
317 """Return True when a Target Module this release implements consumes any of *output_kinds*."""
318 return any(module.implemented for module in consuming_modules(output_kinds))
321def import_location_required(output_kinds: frozenset[str]) -> bool:
322 """Return True when a Target Module that consumes *output_kinds* writes into the import Location."""
323 return any(module.writes_placement for module in consuming_modules(output_kinds))
326__all__ = (
327 "CANDIDATE_TARGET_PREFIX",
328 "CATALOG",
329 "EXTRA_JSON_PREFIX",
330 "POLICY_SECTIONS",
331 "TARGET_MODULES",
332 "KeyFamily",
333 "OutputKind",
334 "PolicySection",
335 "TargetField",
336 "TargetFieldCatalog",
337 "TargetModule",
338 "TargetModuleKey",
339 "ValueKind",
340 "consuming_modules",
341 "declared_modules",
342 "declared_modules_override",
343 "has_implemented_module",
344 "import_location_required",
345 "policy_section",
346 "target_module",
347)