Coverage for src/objectstate/construction_binding.py: 81%

1"""Historical construction declarations, without application lifecycle resources.""" 

2 

3from __future__ import annotations 

4 

5from abc import ABC, abstractmethod 

6from collections.abc import Collection 

7from dataclasses import dataclass 

8from typing import Any, TYPE_CHECKING 

9 

10if TYPE_CHECKING: 

11 from objectstate.object_state import ObjectState 

12 

13 

14class StateScopeOwner(ABC): 

15 """A declaration that authoritatively owns independently scoped states.""" 

16 

17 __slots__ = () 

18 

19 @property 

20 @abstractmethod 

21 def owned_scope_ids(self) -> tuple[str, ...]: 

22 """Project membership from the declaration's existing canonical fields.""" 

23 

24 def validate_scope_membership( 

25 self, scope: str | None, available: Collection[str] 

26 ) -> None: 

27 owned = self.owned_scope_ids 

28 if len(set(owned)) != len(owned): 

29 raise ValueError( 

30 f"ObjectState owner {scope!r} declares duplicate scope membership." 

31 ) 

32 for child in owned: 

33 if child not in available: 

34 raise ValueError( 

35 f"ObjectState owner {scope!r} declares missing ObjectState scope {child!r}." 

36 ) 

37 

38 

39@dataclass(frozen=True) 

40class StateConstructionBinding(ABC): 

41 """The actual extraction owner and independently scoped parent of a state.""" 

42 

43 state_type: type[ObjectState] 

44 target: Any 

45 parent_scope: str | None 

46 exclusions: tuple[str, ...] 

47 parent_field: str | None 

48 

49 @abstractmethod 

50 def validate_lifecycle(self, state: ObjectState | None, scope: str) -> None: 

51 """Validate application ownership before any registry mutation.""" 

52 

53 @abstractmethod 

54 def attach(self, state: ObjectState) -> None: 

55 """Install the extraction owner, retaining application lifecycle resources.""" 

56 

57 def prepare( 

58 self, scope: str, state: ObjectState | None, parent: ObjectState | None 

59 ) -> ObjectState: 

60 """Extract a complete replacement schema outside the active registry.""" 

61 

62 self.validate_lifecycle(state, scope) 

63 prepared = self.state_type( 

64 object_instance=self.target, 

65 scope_id=scope, 

66 parent_state=parent, 

67 exclude_params=list(self.exclusions), 

68 ) 

69 prepared._parent_field_name = self.parent_field 

70 return prepared 

71 

72 def validate_scope_membership( 

73 self, prepared: ObjectState, available: Collection[str] 

74 ) -> None: 

75 """Validate the prepared raw declaration, not its saved backing target.""" 

76 

77 if isinstance(self.target, StateScopeOwner): 

78 owner = prepared.to_object(sync_delegate=False) 

79 if not isinstance(owner, StateScopeOwner): 

80 raise ValueError( 

81 f"Historical scope owner changed role at {prepared.scope_id!r}." 

82 ) 

83 owner.validate_scope_membership(prepared.scope_id, available) 

84 

85 

86@dataclass(frozen=True) 

87class DirectStateBinding(StateConstructionBinding): 

88 """A declaration whose ObjectState can be constructed independently.""" 

89 

90 def validate_lifecycle(self, state: ObjectState | None, scope: str) -> None: 

91 if state is not None and ( 

92 type(state) is not self.state_type or state.has_delegate 

93 ): 

94 raise ValueError(f"Incompatible direct ObjectState lifecycle at {scope!r}.") 

95 

96 def attach(self, state: ObjectState) -> None: 

97 state.object_instance = self.target 

98 state._extraction_target = self.target 

99 state._delegate_attr = None 

100 

101 

102@dataclass(frozen=True) 

103class DelegatedStateBinding(StateConstructionBinding): 

104 """An editable declaration owned by an already-live application wrapper.""" 

105 

106 lifecycle_type: type 

107 delegate_attribute: str 

108 

109 def validate_lifecycle(self, state: ObjectState | None, scope: str) -> None: 

110 if ( 

111 state is None 

112 or type(state) is not self.state_type 

113 or type(state.object_instance) is not self.lifecycle_type 

114 or state._delegate_attr != self.delegate_attribute 

115 ): 

116 raise ValueError( 

117 f"Missing or incompatible delegated lifecycle at {scope!r}." 

118 ) 

119 

120 def attach(self, state: ObjectState) -> None: 

121 setattr(state.object_instance, self.delegate_attribute, self.target) 

122 state._extraction_target = self.target 

123 state._delegate_attr = self.delegate_attribute