Coverage for netbox_data_import/adapters.py: 100%
66 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"""Source Adapter registry.
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.
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"""
14from __future__ import annotations
16from dataclasses import dataclass, field
17from typing import TYPE_CHECKING
19from .catalog import OutputKind, has_implemented_module
21if TYPE_CHECKING:
22 from .trace_workbook import SourceTrace
25class UnknownSourceAdapter(Exception):
26 """A profile names a Source Adapter key this release does not register."""
29class SourceUnreadable(Exception):
30 """The source document, or the configuration naming its shape, cannot produce rows."""
33@dataclass(frozen=True)
34class SourceDiagnostic:
35 """One thing the adapter could not interpret, reported without failing the batch."""
37 code: str
38 message: str
39 row_number: int | None = None
42@dataclass(frozen=True)
43class SourceBatch:
44 """The typed source items and source diagnostics from one file (section 1)."""
46 output_kinds: frozenset[str]
47 rows: tuple[dict | SourceTrace, ...] = ()
48 diagnostics: tuple[SourceDiagnostic, ...] = ()
49 unused_columns: dict[str, dict] = field(default_factory=dict)
52class SourceAdapter:
53 """Base class for a Source Adapter declaration."""
55 key: str = ""
56 label: str = ""
57 output_kinds: frozenset[str] = frozenset()
59 @classmethod
60 def config_form_class(cls):
61 """Return the Django form that validates this adapter's ``adapter_config``."""
62 raise NotImplementedError
64 @classmethod
65 def interpret(cls, content, adapter_config, *, collect_unused: bool = False) -> SourceBatch:
66 """Return the Source Batch the content carries under *adapter_config*."""
67 raise NotImplementedError
70class FlatWorkbookAdapter(SourceAdapter):
71 """One flat worksheet whose rows describe devices and racks."""
73 key = "flat_workbook"
74 label = "Flat workbook"
75 output_kinds = frozenset({OutputKind.DEVICE_SOURCE_ROW, OutputKind.RACK_SOURCE_ROW})
77 @classmethod
78 def config_form_class(cls):
79 """Return the flat-workbook configuration form."""
80 from .adapter_forms import FlatWorkbookConfigForm
82 return FlatWorkbookConfigForm
84 @classmethod
85 def interpret(cls, content, adapter_config, *, collect_unused: bool = False) -> SourceBatch:
86 """Interpret workbook bytes under a `FlatWorkbookConfig`."""
87 from . import flat_workbook
89 rows, unused = flat_workbook.interpret(content, adapter_config, collect_unused=collect_unused)
90 return SourceBatch(output_kinds=cls.output_kinds, rows=tuple(rows), unused_columns=unused)
93class TraceWorkbookAdapter(SourceAdapter):
94 """A cable-trace workbook whose sheet names are fixed by the Source Trace model."""
96 key = "trace_workbook"
97 label = "Trace workbook"
98 output_kinds = frozenset({OutputKind.SOURCE_TRACE})
100 @classmethod
101 def config_form_class(cls):
102 """Return the trace-workbook configuration form, which declares no settings."""
103 from .adapter_forms import TraceWorkbookConfigForm
105 return TraceWorkbookConfigForm
107 @classmethod
108 def interpret(cls, content, adapter_config, *, collect_unused: bool = False) -> SourceBatch:
109 """Interpret workbook bytes under the fixed trace-workbook format."""
110 from . import trace_workbook
112 rows, diagnostics = trace_workbook.interpret(content)
113 return SourceBatch(output_kinds=cls.output_kinds, rows=rows, diagnostics=diagnostics)
116ADAPTERS: tuple[type[SourceAdapter], ...] = (FlatWorkbookAdapter, TraceWorkbookAdapter)
118_ADAPTERS_BY_KEY = {adapter.key: adapter for adapter in ADAPTERS}
120DEFAULT_ADAPTER_KEY = FlatWorkbookAdapter.key
123def get_adapter(key: str) -> type[SourceAdapter] | None:
124 """Return the adapter class registered under *key*, or None."""
125 return _ADAPTERS_BY_KEY.get(key)
128def adapter_choices():
129 """Return Django choice pairs for every registered adapter."""
130 return [(adapter.key, adapter.label) for adapter in ADAPTERS]
133def selectable_adapter_choices():
134 """Return choice pairs for the adapters a Target Module in this release can consume."""
135 return [(adapter.key, adapter.label) for adapter in ADAPTERS if has_implemented_module(adapter.output_kinds)]
138def output_kinds_for(key: str) -> frozenset[str]:
139 """Return the output kinds the adapter registered under *key* emits."""
140 adapter = get_adapter(key)
141 return adapter.output_kinds if adapter is not None else frozenset()
144__all__ = (
145 "ADAPTERS",
146 "DEFAULT_ADAPTER_KEY",
147 "FlatWorkbookAdapter",
148 "SourceAdapter",
149 "SourceBatch",
150 "SourceDiagnostic",
151 "SourceUnreadable",
152 "TraceWorkbookAdapter",
153 "UnknownSourceAdapter",
154 "adapter_choices",
155 "get_adapter",
156 "output_kinds_for",
157 "selectable_adapter_choices",
158)