Coverage for greyhorse / river / component.py: 100%

1"""Component — lifecycle container for operators and resource slots. 

2 

3Implements ``OperatorResolver`` protocol and adds lifecycle phases 

4(IDLE → SETUP → STARTED → STOPPED → IDLE) with a slot registry. 

5 

6Component does **not own** fragments — it receives them already setup'd. 

7Fragment lifecycle is managed by Module (future). 

8""" 

9 

10from __future__ import annotations 

11 

12import enum 

13from collections.abc import Iterable, Mapping 

14from enum import IntEnum 

15from types import MappingProxyType 

16from typing import Any 

17 

18from greyhorse.app.private.functional.linker import FragmentLinker 

19from greyhorse.app.resources.slot import OwnerSlot 

20 

21 

22class ComponentState(enum.Enum): 

23 """Lifecycle state of a Component.""" 

24 

25 IDLE = enum.auto() 

26 SETUP = enum.auto() 

27 STARTED = enum.auto() 

28 STOPPED = enum.auto() 

29 

30 

31class Component: 

32 """Lifecycle container for operators and resource slots. 

33 

34 Wraps fragments into a ``FragmentLinker``, manages lifecycle phases, 

35 and provides a slot registry for controller-owned resources. 

36 

37 Component does not own fragments — they must be setup'd before 

38 being passed in. Fragment lifecycle is Module's responsibility. 

39 

40 Args: 

41 name: Component identifier. 

42 fragments: Iterable of ``Fragment`` instances (must be setup'd). 

43 scope: Optional phase scope for ``linker.link()``. 

44 context_values: Pre-resolved runtime dependencies. 

45 """ 

46 

47 __slots__ = ('_context_values', '_linker', '_name', '_scope', '_slot_registry', '_state') 

48 

49 def __init__( 

50 self, 

51 name: str, 

52 fragments: Iterable = (), 

53 scope: IntEnum | None = None, 

54 context_values: Mapping[str, Any] | None = None, 

55 ) -> None: 

56 self._name = name 

57 self._linker = FragmentLinker(fragments) 

58 self._scope = scope 

59 self._context_values = MappingProxyType(dict(context_values or {})) 

60 self._slot_registry: dict[str, OwnerSlot] = {} 

61 self._state = ComponentState.IDLE 

62 

63 # ------------------------------------------------------------------ 

64 # OperatorResolver protocol 

65 # ------------------------------------------------------------------ 

66 

67 @property 

68 def linker(self) -> FragmentLinker: 

69 """FragmentLinker for multi-root type resolution across fragments.""" 

70 return self._linker 

71 

72 @property 

73 def scope(self) -> IntEnum | None: 

74 """Optional phase scope for ``linker.link()`` filtering.""" 

75 return self._scope 

76 

77 @property 

78 def context_values(self) -> Mapping[str, Any]: 

79 """Read-only mapping of pre-resolved runtime dependencies.""" 

80 return self._context_values 

81 

82 # ------------------------------------------------------------------ 

83 # Identity and state 

84 # ------------------------------------------------------------------ 

85 

86 @property 

87 def name(self) -> str: 

88 return self._name 

89 

90 @property 

91 def state(self) -> ComponentState: 

92 return self._state 

93 

94 # ------------------------------------------------------------------ 

95 # Lifecycle 

96 # ------------------------------------------------------------------ 

97 

98 def _require_state(self, expected: ComponentState, method: str) -> None: 

99 if self._state != expected: 

100 raise RuntimeError( 

101 f'{self._name}: {method}() requires {expected.name}, got {self._state.name}' 

102 ) 

103 

104 def setup(self) -> None: 

105 """Initialize component. IDLE → SETUP.""" 

106 self._require_state(ComponentState.IDLE, 'setup') 

107 self._state = ComponentState.SETUP 

108 

109 def start(self) -> None: 

110 """Start component. SETUP → STARTED.""" 

111 self._require_state(ComponentState.SETUP, 'start') 

112 self._state = ComponentState.STARTED 

113 

114 def stop(self) -> None: 

115 """Stop component. STARTED → STOPPED.""" 

116 self._require_state(ComponentState.STARTED, 'stop') 

117 self._state = ComponentState.STOPPED 

118 

119 def teardown(self) -> None: 

120 """Teardown component. STOPPED|SETUP → IDLE.""" 

121 if self._state not in (ComponentState.STOPPED, ComponentState.SETUP): 

122 self._require_state(ComponentState.STOPPED, 'teardown') 

123 for slot in reversed(list(self._slot_registry.values())): 

124 slot.remove('teardown') 

125 self._slot_registry.clear() 

126 self._state = ComponentState.IDLE 

127 

128 # ------------------------------------------------------------------ 

129 # Slot registry 

130 # ------------------------------------------------------------------ 

131 

132 def create_slot[T](self, name: str) -> OwnerSlot[T]: 

133 """Create and register a named resource slot.""" 

134 if name in self._slot_registry: 

135 raise RuntimeError(f'Component {self._name}: slot "{name}" already exists') 

136 slot: OwnerSlot[T] = OwnerSlot(name) 

137 self._slot_registry[name] = slot 

138 return slot 

139 

140 def get_slot(self, name: str) -> OwnerSlot | None: 

141 """Get a registered slot by name, or None.""" 

142 return self._slot_registry.get(name) 

143 

144 @property 

145 def slots(self) -> Mapping[str, OwnerSlot]: 

146 """Read-only view of registered slots.""" 

147 return MappingProxyType(self._slot_registry)