Coverage for netbox_data_import/adapters.py: 100.00%

1# SPDX-License-Identifier: Apache-2.0 

2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com> 

3"""Source Adapter registry. 

4 

5The registry is a static in-plugin mapping from a stable adapter key to the adapter class. Forms, 

6REST, GraphQL, and YAML derive their choices from it. There is no third-party extension point. An 

7adapter is offered as a choice only once the catalog declares a Target Module that consumes it. 

8 

9An adapter declares source interpretation only, so this module imports no NetBox model and no Target 

10Module implementation. The configuration form is imported lazily because it validates references at 

11the NetBox boundary. 

12""" 

13 

14from __future__ import annotations 

15 

16from dataclasses import dataclass, field, replace 

17from typing import TYPE_CHECKING 

18 

19from .catalog import OutputKind, has_implemented_module 

20from .identity import identity_text 

21from .public_refusal import PublicRefusal 

22 

23if TYPE_CHECKING: 

24 from .source_trace import SourceTrace 

25 

26 

27class UnknownSourceAdapter(PublicRefusal): 

28 """A profile names a Source Adapter key this release does not register.""" 

29 

30 

31class SourceUnreadable(Exception): 

32 """The source document, or the configuration naming its shape, cannot produce rows.""" 

33 

34 

35@dataclass(frozen=True) 

36class SourceDiagnostic: 

37 """One thing the adapter could not interpret, reported without failing the batch.""" 

38 

39 code: str 

40 message: str 

41 row_number: int | None = None 

42 

43 

44def _foreign_row_number(row) -> int | None: 

45 """Return the source row a foreign row states, so its diagnostic can name a location.""" 

46 provenance = getattr(row, "provenance", ()) 

47 if provenance: 

48 return provenance[0].row_start 

49 if isinstance(row, dict): 

50 return row.get("_row_number") 

51 return None 

52 

53 

54@dataclass(frozen=True) 

55class SourceBatch: 

56 """The typed source items and source diagnostics from one file (section 1).""" 

57 

58 output_kinds: frozenset[str] 

59 rows: tuple[dict | SourceTrace, ...] = () 

60 diagnostics: tuple[SourceDiagnostic, ...] = () 

61 unused_columns: dict[str, dict] = field(default_factory=dict) 

62 

63 def __post_init__(self): 

64 """Exclude a row the output kinds cannot carry, and mark a Trace with empty Device references.""" 

65 from .source_trace import SourceTrace 

66 

67 trace_batch = OutputKind.SOURCE_TRACE in self.output_kinds 

68 expected = SourceTrace if trace_batch else dict 

69 carries = "Source Traces" if trace_batch else "source rows" 

70 rows = [] 

71 diagnostics = list(self.diagnostics) 

72 for row in self.rows: 

73 if not isinstance(row, expected): 

74 # Excluded here, because every reader below reads the row the output kinds declare. 

75 diagnostics.append( 

76 SourceDiagnostic( 

77 code="source.row_type_unexpected", 

78 message=f"This batch carries {carries} only; got {type(row).__name__}.", 

79 row_number=_foreign_row_number(row), 

80 ) 

81 ) 

82 continue 

83 if not trace_batch: 

84 rows.append(row) 

85 continue 

86 summary = row.endpoint_summary 

87 references = [summary.from_termination, summary.to_termination, *row.corroboration] 

88 for segment in row.segments: 

89 references.extend((segment.left, segment.right)) 

90 if all(identity_text(reference.device) for reference in references): 

91 rows.append(row) 

92 continue 

93 diagnostic = SourceDiagnostic( 

94 code="trace.device_required", 

95 message="A Source Trace Termination Reference must name a Device.", 

96 row_number=row.provenance[0].row_start, 

97 ) 

98 diagnostics.append(diagnostic) 

99 rows.append(replace(row, errors=(*row.errors, diagnostic))) 

100 object.__setattr__(self, "rows", tuple(rows)) 

101 object.__setattr__(self, "diagnostics", tuple(diagnostics)) 

102 

103 

104class SourceAdapter: 

105 """Base class for a Source Adapter declaration.""" 

106 

107 key: str = "" 

108 label: str = "" 

109 output_kinds: frozenset[str] = frozenset() 

110 

111 @classmethod 

112 def config_form_class(cls): 

113 """Return the Django form that validates this adapter's ``adapter_config``.""" 

114 raise NotImplementedError 

115 

116 @classmethod 

117 def interpret(cls, content, adapter_config, *, collect_unused: bool = False) -> SourceBatch: 

118 """Return the Source Batch the content carries under *adapter_config*.""" 

119 raise NotImplementedError 

120 

121 

122class FlatWorkbookAdapter(SourceAdapter): 

123 """One flat worksheet whose rows describe devices and racks.""" 

124 

125 key = "flat_workbook" 

126 label = "Flat workbook" 

127 output_kinds = frozenset({OutputKind.DEVICE_SOURCE_ROW, OutputKind.RACK_SOURCE_ROW}) 

128 

129 @classmethod 

130 def config_form_class(cls): 

131 """Return the flat-workbook configuration form.""" 

132 from .adapter_forms import FlatWorkbookConfigForm 

133 

134 return FlatWorkbookConfigForm 

135 

136 @classmethod 

137 def interpret(cls, content, adapter_config, *, collect_unused: bool = False) -> SourceBatch: 

138 """Interpret workbook bytes under a `FlatWorkbookConfig`.""" 

139 from . import flat_workbook 

140 

141 rows, unused = flat_workbook.interpret(content, adapter_config, collect_unused=collect_unused) 

142 return SourceBatch(output_kinds=cls.output_kinds, rows=tuple(rows), unused_columns=unused) 

143 

144 

145class TraceWorkbookAdapter(SourceAdapter): 

146 """A cable-trace workbook whose sheet names are fixed by the Source Trace model.""" 

147 

148 key = "trace_workbook" 

149 label = "Trace workbook" 

150 output_kinds = frozenset({OutputKind.SOURCE_TRACE}) 

151 

152 @classmethod 

153 def config_form_class(cls): 

154 """Return the trace-workbook configuration form, which declares no settings.""" 

155 from .adapter_forms import TraceWorkbookConfigForm 

156 

157 return TraceWorkbookConfigForm 

158 

159 @classmethod 

160 def interpret(cls, content, adapter_config, *, collect_unused: bool = False) -> SourceBatch: 

161 """Interpret workbook bytes under a `TraceWorkbookConfig`.""" 

162 from . import trace_workbook 

163 

164 rows, diagnostics = trace_workbook.interpret(content, adapter_config) 

165 return SourceBatch(output_kinds=cls.output_kinds, rows=rows, diagnostics=diagnostics) 

166 

167 

168ADAPTERS: tuple[type[SourceAdapter], ...] = (FlatWorkbookAdapter, TraceWorkbookAdapter) 

169 

170_ADAPTERS_BY_KEY = {adapter.key: adapter for adapter in ADAPTERS} 

171 

172DEFAULT_ADAPTER_KEY = FlatWorkbookAdapter.key 

173 

174 

175def get_adapter(key: str) -> type[SourceAdapter] | None: 

176 """Return the adapter class registered under *key*, or None.""" 

177 return _ADAPTERS_BY_KEY.get(key) 

178 

179 

180def adapter_choices(): 

181 """Return Django choice pairs for every registered adapter.""" 

182 return [(adapter.key, adapter.label) for adapter in ADAPTERS] 

183 

184 

185def selectable_adapter_choices(): 

186 """Return choice pairs for the adapters a Target Module in this release can consume.""" 

187 return [(adapter.key, adapter.label) for adapter in ADAPTERS if has_implemented_module(adapter.output_kinds)] 

188 

189 

190def output_kinds_for(key: str) -> frozenset[str]: 

191 """Return the output kinds the adapter registered under *key* emits.""" 

192 adapter = get_adapter(key) 

193 return adapter.output_kinds if adapter is not None else frozenset() 

194 

195 

196__all__ = ( 

197 "ADAPTERS", 

198 "DEFAULT_ADAPTER_KEY", 

199 "FlatWorkbookAdapter", 

200 "SourceAdapter", 

201 "SourceBatch", 

202 "SourceDiagnostic", 

203 "SourceUnreadable", 

204 "TraceWorkbookAdapter", 

205 "UnknownSourceAdapter", 

206 "adapter_choices", 

207 "get_adapter", 

208 "output_kinds_for", 

209 "selectable_adapter_choices", 

210)