Coverage for netbox_data_import/catalog.py: 100%
127 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"""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
56TARGET_MODULES: tuple[TargetModule, ...] = (
57 TargetModule(
58 key=TargetModuleKey.DEVICE,
59 label="Device",
60 consumes=frozenset({OutputKind.DEVICE_SOURCE_ROW}),
61 implemented=True,
62 ),
63 TargetModule(
64 key=TargetModuleKey.RACK,
65 label="Rack",
66 consumes=frozenset({OutputKind.RACK_SOURCE_ROW}),
67 implemented=True,
68 ),
69 TargetModule(
70 key=TargetModuleKey.CABLE,
71 label="Cable",
72 consumes=frozenset({OutputKind.SOURCE_TRACE}),
73 implemented=True,
74 ),
75)
77_MODULES_BY_KEY = {module.key: module for module in TARGET_MODULES}
80@dataclass(frozen=True)
81class TargetField:
82 """One catalog entry with a fixed key."""
84 key: str
85 label: str
86 value_kind: str
87 output_kinds: frozenset[str]
88 # A candidate target supplies review candidates rather than a written value. ColumnTransformRule
89 # excludes these because a regex capture group cannot produce a candidate bundle.
90 candidate_target: bool = False
92 @property
93 def target_modules(self) -> frozenset[str]:
94 """Return the Target Modules that consume this field, derived from its output kinds."""
95 return frozenset(m.key for m in TARGET_MODULES if m.consumes & self.output_kinds)
98@dataclass(frozen=True)
99class KeyFamily:
100 """A catalog entry whose exact key is data, not a fixed choice.
102 The key is the prefix plus a name. Every surface resolves a family key through this validator,
103 so no surface reimplements the prefix rule.
104 """
106 prefix: str
107 label: str
108 value_kind: str
109 output_kinds: frozenset[str]
110 name_label: str = "Custom field"
112 def matches(self, key: str) -> bool:
113 """Return True when *key* carries this family's prefix."""
114 return key.startswith(self.prefix)
116 def name_of(self, key: str) -> str:
117 """Return the name part of *key*, without the prefix."""
118 return key[len(self.prefix) :]
120 def is_valid(self, key: str) -> bool:
121 """Return True when *key* is a well-formed member: the prefix plus a non-empty name."""
122 return self.matches(key) and bool(self.name_of(key).strip())
124 def display(self, key: str) -> str:
125 """Return the human-readable name for a member key."""
126 return f"{self.name_label}: {self.name_of(key)}"
128 @property
129 def target_modules(self) -> frozenset[str]:
130 """Return the Target Modules that consume this family, derived from its output kinds."""
131 return frozenset(m.key for m in TARGET_MODULES if m.consumes & self.output_kinds)
134_FLAT_ROW_KINDS = frozenset({OutputKind.DEVICE_SOURCE_ROW, OutputKind.RACK_SOURCE_ROW})
135_DEVICE_ONLY = frozenset({OutputKind.DEVICE_SOURCE_ROW})
136_TRACE_ONLY = frozenset({OutputKind.SOURCE_TRACE})
138EXTRA_JSON_PREFIX = "extra_json:"
139CANDIDATE_TARGET_PREFIX = "candidate:"
142@dataclass(frozen=True)
143class TargetFieldCatalog:
144 """The static registry of Target Fields and key families."""
146 fields: tuple[TargetField, ...]
147 families: tuple[KeyFamily, ...]
148 _by_key: dict[str, TargetField] = field(init=False, repr=False, compare=False, default_factory=dict)
150 def __post_init__(self):
151 object.__setattr__(self, "_by_key", {entry.key: entry for entry in self.fields})
153 def entry(self, key: str) -> TargetField | None:
154 """Return the fixed-key entry for *key*, or None."""
155 return self._by_key.get(key)
157 def family(self, key: str) -> KeyFamily | None:
158 """Return the key family that claims *key*, or None."""
159 for candidate in self.families:
160 if candidate.matches(key):
161 return candidate
162 return None
164 def is_valid(self, key: str, *, output_kinds: frozenset[str] | None = None, allow_candidates: bool = True) -> bool:
165 """Return True when *key* is a Target Field the given output kinds can supply."""
166 entry = self.entry(key)
167 if entry is not None:
168 if entry.candidate_target and not allow_candidates:
169 return False
170 return output_kinds is None or bool(entry.output_kinds & output_kinds)
171 family = self.family(key)
172 if family is None or not family.is_valid(key):
173 return False
174 return output_kinds is None or bool(family.output_kinds & output_kinds)
176 def display(self, key: str) -> str:
177 """Return the human-readable name for any valid key, or the key itself."""
178 entry = self.entry(key)
179 if entry is not None:
180 return entry.label
181 family = self.family(key)
182 if family is not None and family.is_valid(key):
183 return family.display(key)
184 return key
186 def choices(self, *, output_kinds: frozenset[str] | None = None, allow_candidates: bool = True):
187 """Return Django choice pairs for the fixed-key entries the given output kinds can supply."""
188 return [
189 (entry.key, entry.label)
190 for entry in self.fields
191 if (allow_candidates or not entry.candidate_target)
192 and (output_kinds is None or entry.output_kinds & output_kinds)
193 ]
195 def invalid_key_message(self, key: str) -> str:
196 """Return the validation message for a rejected key."""
197 prefixes = ", ".join(f"'{f.prefix}'" for f in self.families)
198 return f"Value '{key}' is not a valid choice. Must be one of the standard field names or start with {prefixes}."
201CATALOG = TargetFieldCatalog(
202 fields=(
203 TargetField("rack_name", "Rack name", ValueKind.TEXT, _FLAT_ROW_KINDS),
204 TargetField("device_name", "Device name", ValueKind.TEXT, _FLAT_ROW_KINDS),
205 TargetField("device_class", "Device class (maps to role/rack)", ValueKind.TEXT, _FLAT_ROW_KINDS),
206 TargetField("face", "Face (Front/Back)", ValueKind.CHOICE, _DEVICE_ONLY),
207 TargetField("airflow", "Airflow", ValueKind.CHOICE, _DEVICE_ONLY),
208 TargetField("u_position", "U position", ValueKind.INTEGER, _DEVICE_ONLY),
209 TargetField("status", "Status", ValueKind.CHOICE, _DEVICE_ONLY),
210 TargetField("make", "Make (manufacturer)", ValueKind.TEXT, _DEVICE_ONLY),
211 TargetField("model", "Model (device type)", ValueKind.TEXT, _DEVICE_ONLY),
212 TargetField("u_height", "U height", ValueKind.DECIMAL, _FLAT_ROW_KINDS),
213 TargetField("serial", "Serial number", ValueKind.TEXT, _FLAT_ROW_KINDS),
214 TargetField("asset_tag", "Asset tag", ValueKind.TEXT, _DEVICE_ONLY),
215 TargetField("primary_ip4", "Primary IPv4", ValueKind.IP_ADDRESS, _DEVICE_ONLY),
216 TargetField("primary_ip6", "Primary IPv6", ValueKind.IP_ADDRESS, _DEVICE_ONLY),
217 TargetField("oob_ip", "Out-of-band IP", ValueKind.IP_ADDRESS, _DEVICE_ONLY),
218 TargetField("primary_contact", "Primary contact", ValueKind.TEXT, _DEVICE_ONLY),
219 TargetField("source_id", "Source ID (stored in custom field)", ValueKind.TEXT, _FLAT_ROW_KINDS),
220 TargetField(
221 f"{CANDIDATE_TARGET_PREFIX}contact",
222 "Candidate values: Contact fields",
223 ValueKind.CANDIDATE,
224 _DEVICE_ONLY,
225 candidate_target=True,
226 ),
227 ),
228 families=(
229 KeyFamily(
230 prefix=EXTRA_JSON_PREFIX,
231 label="Custom field",
232 value_kind=ValueKind.TEXT,
233 output_kinds=_DEVICE_ONLY,
234 ),
235 ),
236)
239@dataclass(frozen=True)
240class PolicySection:
241 """A profile policy table and the adapter output kinds it applies to."""
243 key: str
244 label: str
245 output_kinds: frozenset[str]
247 def applies_to(self, output_kinds: frozenset[str]) -> bool:
248 """Return True when an adapter emitting *output_kinds* can use this section."""
249 return bool(self.output_kinds & output_kinds)
252POLICY_SECTIONS: tuple[PolicySection, ...] = (
253 PolicySection("column_mappings", "Column Mappings", _FLAT_ROW_KINDS),
254 PolicySection("column_transform_rules", "Column Transform Rules", _FLAT_ROW_KINDS),
255 PolicySection("class_role_mappings", "Class/Role Mappings", _FLAT_ROW_KINDS),
256 PolicySection("device_type_mappings", "Device Type Mappings", _FLAT_ROW_KINDS),
257 PolicySection("manufacturer_mappings", "Manufacturer Mappings", _FLAT_ROW_KINDS),
258 PolicySection("ignored_devices", "Ignored Devices", _FLAT_ROW_KINDS),
259 PolicySection("source_resolutions", "Source Resolutions", _DEVICE_ONLY),
260 PolicySection("device_existing_matches", "Device Existing Matches", _DEVICE_ONLY),
261 PolicySection("ignored_field_differences", "Ignored Field Differences", _DEVICE_ONLY),
262 PolicySection("termination_resolutions", "Termination Resolutions", _TRACE_ONLY),
263 PolicySection("cable_class_mappings", "CableClass Mappings", _TRACE_ONLY),
264)
266_SECTIONS_BY_KEY = {section.key: section for section in POLICY_SECTIONS}
269def target_module(key: str) -> TargetModule | None:
270 """Return the Target Module declared under *key*."""
271 return _MODULES_BY_KEY.get(key)
274def policy_section(key: str) -> PolicySection | None:
275 """Return the policy section declared under *key*."""
276 return _SECTIONS_BY_KEY.get(key)
279_declared_override: ContextVar[tuple[TargetModule, ...] | None] = ContextVar("declared_modules", default=None)
282def declared_modules() -> tuple[TargetModule, ...]:
283 """Return the Target Module declarations in force."""
284 override = _declared_override.get()
285 return TARGET_MODULES if override is None else override
288@contextmanager
289def declared_modules_override(modules: Sequence[TargetModule]) -> Iterator[None]:
290 """Run the block against *modules*, so a caller can state a table this release does not ship.
292 Every entry point reaches the runtime gate through `ImportProfile.clean`, far below its own
293 signature, so a declaration table cannot be passed down to it as an argument.
294 """
295 token = _declared_override.set(tuple(modules))
296 try:
297 yield
298 finally:
299 _declared_override.reset(token)
302def consuming_modules(output_kinds: frozenset[str]) -> tuple[TargetModule, ...]:
303 """Return the Target Modules that consume any of *output_kinds*."""
304 return tuple(module for module in declared_modules() if module.consumes & output_kinds)
307def has_implemented_module(output_kinds: frozenset[str]) -> bool:
308 """Return True when a Target Module this release implements consumes any of *output_kinds*."""
309 return any(module.implemented for module in consuming_modules(output_kinds))
312__all__ = (
313 "CANDIDATE_TARGET_PREFIX",
314 "CATALOG",
315 "EXTRA_JSON_PREFIX",
316 "POLICY_SECTIONS",
317 "TARGET_MODULES",
318 "KeyFamily",
319 "OutputKind",
320 "PolicySection",
321 "TargetField",
322 "TargetFieldCatalog",
323 "TargetModule",
324 "TargetModuleKey",
325 "ValueKind",
326 "consuming_modules",
327 "declared_modules",
328 "declared_modules_override",
329 "has_implemented_module",
330 "policy_section",
331 "target_module",
332)