Coverage for src / lexigram / admin / state / store.py: 0%

1"""Reactive state management based on Signals for Lexigram Admin.""" 

2 

3from __future__ import annotations 

4 

5import contextvars 

6from typing import TYPE_CHECKING, Generic, TypeVar 

7 

8if TYPE_CHECKING: 

9 from collections.abc import Callable 

10 

11T = TypeVar("T") 

12 

13# Context for tracking active computations 

14_active_effect: contextvars.ContextVar[Effect | None] = contextvars.ContextVar( 

15 "_active_effect", 

16 default=None, 

17) 

18 

19 

20class Signal(Generic[T]): 

21 """A reactive value that can be observed/subscribed to.""" 

22 

23 def __init__(self, value: T) -> None: 

24 self._value = value 

25 self._subscribers: set[Effect | Computed] = set() 

26 

27 def get(self) -> T: 

28 """Get the value and track the dependency if an effect is active.""" 

29 effect = _active_effect.get() 

30 if effect: 

31 self._subscribers.add(effect) 

32 effect.add_dependency(self) 

33 return self._value 

34 

35 def set(self, value: T) -> None: 

36 """Update the value and notify all subscribers.""" 

37 if self._value == value: 

38 return 

39 self._value = value 

40 # Copy subscribers to avoid modification during iteration 

41 for subscriber in list(self._subscribers): 

42 subscriber.update() 

43 

44 def add_subscriber(self, subscriber: Effect | Computed) -> None: 

45 """Add a subscriber to this signal.""" 

46 self._subscribers.add(subscriber) 

47 

48 def remove_subscriber(self, subscriber: Effect | Computed) -> None: 

49 """Remove a subscriber from this signal.""" 

50 self._subscribers.discard(subscriber) 

51 

52 def __repr__(self) -> str: 

53 return f"Signal({self._value})" 

54 

55 

56class Computed(Generic[T]): 

57 """A value derived from other signals or computed values.""" 

58 

59 def __init__(self, fn: Callable[[], T]) -> None: 

60 self._fn = fn 

61 self._value: T | None = None 

62 self._dirty = True 

63 self._subscribers: set[Effect | Computed] = set() 

64 self._dependencies: set[Signal | Computed] = set() 

65 

66 def get(self) -> T: 

67 """Get the computed value, re-evaluating if dirty.""" 

68 effect = _active_effect.get() 

69 if effect: 

70 self._subscribers.add(effect) 

71 effect.add_dependency(self) 

72 

73 if self._dirty: 

74 self._evaluate() 

75 return self._value # type: ignore[return-value] 

76 

77 def _evaluate(self) -> None: 

78 """Run the computation and track dependencies.""" 

79 # Unsubscribe from old dependencies 

80 for dep in self._dependencies: 

81 dep.remove_subscriber(self) 

82 self._dependencies.clear() 

83 

84 # Run with tracking 

85 token = _active_effect.set(self) # type: ignore[arg-type] 

86 try: 

87 self._value = self._fn() 

88 finally: 

89 _active_effect.reset(token) 

90 

91 self._dirty = False 

92 

93 def update(self) -> None: 

94 """Mark as dirty and notify subscribers.""" 

95 if not self._dirty: 

96 self._dirty = True 

97 for subscriber in list(self._subscribers): 

98 subscriber.update() 

99 

100 def add_dependency(self, dependency: Signal | Computed) -> None: 

101 """Add a dependency to this computed value.""" 

102 self._dependencies.add(dependency) 

103 

104 def remove_subscriber(self, subscriber: Effect | Computed) -> None: 

105 """Remove a subscriber from this computed value.""" 

106 self._subscribers.discard(subscriber) 

107 

108 def __repr__(self) -> str: 

109 return f"Computed({self._value})" 

110 

111 

112class Effect: 

113 """A side-effect that runs whenever its dependencies change.""" 

114 

115 def __init__(self, fn: Callable[[], None]) -> None: 

116 self._fn = fn 

117 self._dependencies: set[Signal | Computed] = set() 

118 self._run() 

119 

120 def _run(self) -> None: 

121 """Run the effect and track dependencies.""" 

122 # Unsubscribe from old dependencies 

123 for dep in self._dependencies: 

124 dep.remove_subscriber(self) 

125 self._dependencies.clear() 

126 

127 # Run with tracking 

128 token = _active_effect.set(self) 

129 try: 

130 self._fn() 

131 finally: 

132 _active_effect.reset(token) 

133 

134 def update(self) -> None: 

135 """Re-run the effect because a dependency changed.""" 

136 self._run() 

137 

138 def add_dependency(self, dependency: Signal | Computed) -> None: 

139 """Add a dependency to this effect.""" 

140 self._dependencies.add(dependency) 

141 

142 

143def signal(value: T) -> Signal[T]: 

144 """Helper to create a signal.""" 

145 return Signal(value) 

146 

147 

148def computed(fn: Callable[[], T]) -> Computed[T]: 

149 """Helper to create a computed value.""" 

150 return Computed(fn) 

151 

152 

153def watch(fn: Callable[[], None]) -> Effect: 

154 """Helper to create an effect (watcher).""" 

155 return Effect(fn) 

156 

157 

158class GlobalStore: 

159 """Base class for global state containers."""