Coverage for netbox_data_import/target_runtime.py: 100%

1# SPDX-License-Identifier: Apache-2.0 

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

3"""The seam every Target Module implements and the coordinator calls. 

4 

5It lives apart from the implementations so one Target Module can depend on it without depending 

6on another. 

7""" 

8 

9from __future__ import annotations 

10 

11from collections.abc import Mapping 

12from dataclasses import dataclass, field 

13from typing import Any, Protocol 

14 

15from .plan import PlannedChange, SynchronizationUnit 

16 

17 

18class PreconditionFailed(Exception): 

19 """Target state moved between planning and the write, so the change no longer applies.""" 

20 

21 

22@dataclass(frozen=True) 

23class ExecutionContext: 

24 """What a Target Module needs while the coordinator's transaction is open.""" 

25 

26 actor: Any 

27 reader: Any 

28 profile: Any 

29 

30 

31@dataclass(frozen=True) 

32class DeletedObject: 

33 """One object a Planned Change removed, for the execution's deleted-object snapshot.""" 

34 

35 object_type: str 

36 object_id: int 

37 display: str 

38 detail: Mapping[str, Any] = field(default_factory=dict) 

39 

40 def to_dict(self) -> dict: 

41 """Return the serialized form the audit row stores.""" 

42 return { 

43 "object_type": self.object_type, 

44 "object_id": self.object_id, 

45 "display": self.display, 

46 "detail": dict(self.detail), 

47 } 

48 

49 

50class TargetModuleRuntime(Protocol): 

51 """What the coordinator may call on a Target Module.""" 

52 

53 key: str 

54 consumes: frozenset[str] 

55 

56 def plan( 

57 self, 

58 source_batch, 

59 profile, 

60 catalog, 

61 netbox_reader, 

62 *, 

63 lock_plan_references: bool = False, 

64 ) -> list[SynchronizationUnit]: 

65 """Return this module's units and optionally lock target rows the plan only reads. 

66 

67 The caller must open a transaction before it requests locks. 

68 """ 

69 ... 

70 

71 def apply(self, planned_change: PlannedChange, execution_context) -> Any: 

72 """Apply one Planned Change inside the coordinator's transaction.""" 

73 ... 

74 

75 

76__all__ = ( 

77 "DeletedObject", 

78 "ExecutionContext", 

79 "PreconditionFailed", 

80 "TargetModuleRuntime", 

81)