Coverage for netbox_data_import/source_resolution.py: 100%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""Apply saved Source Resolutions to pristine flat source rows.""" 

4 

5from __future__ import annotations 

6 

7from collections.abc import Mapping 

8from typing import Any 

9 

10from django.core.exceptions import ValidationError 

11 

12from .models import ImportProfile, validate_source_resolution_fields 

13from .values import source_text as _str_val 

14 

15 

16def _build_source_to_targets_map(profile: ImportProfile) -> dict[str, list[str]]: 

17 """Return a source-column keyed map of all target fields that column feeds.""" 

18 source_to_targets: dict[str, list[str]] = {} 

19 for mapping in profile.column_mappings.all(): 

20 source_to_targets.setdefault(mapping.source_column, []).append(mapping.target_field) 

21 return source_to_targets 

22 

23 

24def _clear_resolved_conflicts(row_dict: dict[str, Any], resolved_fields: Mapping) -> None: 

25 """Remove conflicts for fields overridden by saved resolutions.""" 

26 for resolved_field in resolved_fields: 

27 row_dict.get("_conflicts", {}).pop(resolved_field, None) 

28 if not row_dict.get("_conflicts"): 

29 row_dict.pop("_conflicts", None) 

30 

31 

32def _apply_one_resolution( 

33 row_dict: dict, 

34 resolution, 

35 source_to_targets: dict[str, list[str]], 

36 profile: ImportProfile, 

37) -> None: 

38 """Apply one saved Source Resolution and clear its ignored mapped values.""" 

39 resolved_fields = resolution.resolved_fields 

40 try: 

41 validate_source_resolution_fields(profile, resolution.source_column, resolved_fields) 

42 except ValidationError: 

43 return 

44 row_dict.update(resolved_fields) 

45 _clear_resolved_conflicts(row_dict, resolved_fields) 

46 

47 candidate_targets = list(source_to_targets.get(resolution.source_column, [])) 

48 if resolution.source_column not in candidate_targets: 

49 candidate_targets.append(resolution.source_column) 

50 for target_field in candidate_targets: 

51 if target_field in resolved_fields: 

52 continue 

53 current = row_dict.get(target_field) 

54 if current is None: 

55 continue 

56 if str(current) == str(resolution.original_value): 

57 row_dict[target_field] = None 

58 

59 

60def derive_effective_rows(rows: list[dict], profile) -> list[dict]: 

61 """Return new effective rows derived from pristine source rows and saved resolutions.""" 

62 resolutions_by_source_id: dict[str, list] = {} 

63 # Source columns give resolutions with one source ID a stable override order. 

64 for resolution in profile.source_resolutions.order_by("source_id", "source_column"): 

65 resolutions_by_source_id.setdefault(str(resolution.source_id), []).append(resolution) 

66 

67 if not resolutions_by_source_id: 

68 return rows 

69 

70 source_to_targets = _build_source_to_targets_map(profile) 

71 

72 result = [] 

73 for row in rows: 

74 source_id = _str_val(row.get("source_id")) 

75 if source_id and source_id in resolutions_by_source_id: 

76 effective_row = dict(row) 

77 if "_conflicts" in effective_row: 

78 effective_row["_conflicts"] = dict(effective_row["_conflicts"]) 

79 for resolution in resolutions_by_source_id[source_id]: 

80 _apply_one_resolution(effective_row, resolution, source_to_targets, profile) 

81 else: 

82 effective_row = row 

83 result.append(effective_row) 

84 return result 

85 

86 

87__all__ = ("derive_effective_rows",)