Coverage for netbox_data_import/target_runtime.py: 100.00%
23 statements
« 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"""The seam every Target Module implements and the coordinator calls.
5It lives apart from the implementations so one Target Module can depend on it without depending
6on another.
7"""
9from __future__ import annotations
11from dataclasses import dataclass
12from typing import Any, Protocol
14from .plan import PlannedChange, SynchronizationUnit
15from .public_refusal import PublicRefusal
18class PreconditionFailed(PublicRefusal):
19 """Target state moved between planning and the write, so the change no longer applies."""
22@dataclass(frozen=True)
23class ExecutionContext:
24 """What a Target Module needs while the coordinator's transaction is open."""
26 actor: Any
27 reader: Any
28 profile: Any
31@dataclass(frozen=True)
32class DeletedObject:
33 """The stable audit identity of one object a Planned Change removed."""
35 object_type: str
36 object_id: int
38 def to_dict(self) -> dict:
39 """Return the serialized form the audit row stores."""
40 return {
41 "object_type": self.object_type,
42 "object_id": self.object_id,
43 }
46class TargetModuleRuntime(Protocol):
47 """What the coordinator may call on a Target Module."""
49 key: str
50 consumes: frozenset[str]
52 def plan(
53 self,
54 source_batch,
55 profile,
56 catalog,
57 netbox_reader,
58 *,
59 lock_plan_references: bool = False,
60 ) -> list[SynchronizationUnit]:
61 """Return this module's units and optionally lock target rows the plan only reads.
63 The caller must open a transaction before it requests locks.
64 """
65 ...
67 def apply(self, planned_change: PlannedChange, execution_context) -> Any:
68 """Apply one Planned Change inside the coordinator's transaction."""
69 ...
72__all__ = (
73 "DeletedObject",
74 "ExecutionContext",
75 "PreconditionFailed",
76 "TargetModuleRuntime",
77)