Coverage for netbox_data_import/adapter_config.py: 92.31%
26 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
« 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"""Project stored Import Profile policy and preview decisions into plain Source Adapter configuration."""
5from collections.abc import Mapping
7from .adapters import FlatWorkbookAdapter, TraceWorkbookAdapter, UnknownSourceAdapter
8from .field_keys import port_class_mapping_errors
9from .flat_workbook import FlatWorkbookConfig, TransformRule
10from .plan import PlanError
11from .trace_workbook import TraceWorkbookConfig
13#: The planning-context key of the PortClass map one preview decides for itself.
14PREVIEW_PORT_CLASS_MAP = "port_class_map"
17def preview_port_class_map(planning_context: Mapping) -> dict[str, str]:
18 """Return the PortClass map a planning context carries, refusing one that maps a label it may not."""
19 stated = planning_context.get(PREVIEW_PORT_CLASS_MAP, {})
20 if not isinstance(stated, Mapping) or any(
21 not isinstance(label, str) or not isinstance(port_class, str) or port_class_mapping_errors(label, port_class)
22 for label, port_class in stated.items()
23 ):
24 raise PlanError("The preview PortClass map is invalid.")
25 return dict(stated)
28def _flat_workbook_config(profile, planning_context) -> FlatWorkbookConfig:
29 """Return the detached configuration for one flat-workbook profile."""
30 del planning_context
31 return FlatWorkbookConfig(
32 sheet_name=profile.adapter_settings.sheet_name,
33 column_map={field: tuple(columns) for field, columns in profile.grouped_column_map().items()},
34 transform_rules=tuple(
35 TransformRule(
36 source_column=rule.source_column,
37 pattern=rule.pattern,
38 group_1_target=rule.group_1_target or "",
39 group_2_target=rule.group_2_target or "",
40 )
41 for rule in profile.column_transform_rules.all()
42 ),
43 capture_extra_data=profile.adapter_settings.capture_extra_data,
44 )
47def _trace_workbook_config(profile, planning_context) -> TraceWorkbookConfig:
48 """Return the trace-workbook configuration: the profile PortClass map, then the preview's own over it."""
49 port_class_map = {row.source_label: row.port_class for row in profile.port_class_mappings.all()}
50 port_class_map.update(preview_port_class_map(planning_context))
51 return TraceWorkbookConfig(port_class_map=port_class_map)
54_CONFIG_PROJECTORS = {
55 FlatWorkbookAdapter.key: _flat_workbook_config,
56 TraceWorkbookAdapter.key: _trace_workbook_config,
57}
60def interpreter_config_for(profile, planning_context: Mapping):
61 """Return the detached interpreter configuration for an Import Profile and one planning context."""
62 projector = _CONFIG_PROJECTORS.get(profile.source_adapter)
63 if projector is None:
64 raise UnknownSourceAdapter(
65 f"No interpreter configuration projector is registered for '{profile.source_adapter}'."
66 )
67 return projector(profile, planning_context)
70__all__ = ("PREVIEW_PORT_CLASS_MAP", "interpreter_config_for", "preview_port_class_map")