Coverage for src/lexigram/admin/state/dependency_tracker.py: 37%

1"""Dependency Tracking Service for Reactive Updates.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7 

8class DependencyStack: 

9 """Manages the stack of currently computing properties.""" 

10 

11 def __init__(self) -> None: 

12 self._stack: list[str] = [] 

13 

14 def push(self, key: str) -> None: 

15 """Push a computed property key onto the running stack.""" 

16 self._stack.append(key) 

17 

18 def pop(self) -> str | None: 

19 """Pop the current computed property key.""" 

20 if self._stack: 

21 return self._stack.pop() 

22 return None 

23 

24 @property 

25 def current(self) -> str | None: 

26 """Get the currently executing computed property key.""" 

27 if self._stack: 

28 return self._stack[-1] 

29 return None 

30 

31 

32class DependencyTracker: 

33 """Tracks dependencies between computed properties for reactive updates.""" 

34 

35 def __init__(self) -> None: 

36 if getattr(self, "_initialized", False): 

37 return 

38 

39 # Maps computed property key -> set of source keys it depends on (Computed -> {Sources}) 

40 # This is primarily for debugging/introspection if needed, but invalidation relies on Source -> {Computeds} 

41 # Actually, for efficient invalidation we need Source -> {Computeds}. 

42 

43 # Source Key -> Set of Subscriber Keys (Who depends on me?) 

44 self._subscribers: dict[str, set[str]] = {} 

45 

46 # Computed Key -> Cached Value 

47 self._cache: dict[str, Any] = {} 

48 

49 # The stack of currently running computations 

50 self.stack = DependencyStack() 

51 

52 self._initialized = True 

53 

54 def track(self, source_key: str) -> None: 

55 """Record that the currently running computation depends on this source. 

56 

57 Called by Signals when they are read. 

58 """ 

59 current_computation = self.stack.current 

60 if current_computation: 

61 # The current computation depends on source_key 

62 if source_key not in self._subscribers: 

63 self._subscribers[source_key] = set() 

64 self._subscribers[source_key].add(current_computation) 

65 

66 def trigger(self, source_key: str) -> None: 

67 """Trigger invalidation for all dependants of this source. 

68 

69 Called by Signals when they change. 

70 """ 

71 if source_key in self._subscribers: 

72 dependants = self._subscribers[source_key] 

73 # Copy to avoid modification while iterating (though we just invalidate cache) 

74 to_invalidate = list(dependants) 

75 

76 for computed_key in to_invalidate: 

77 self.invalidate(computed_key) 

78 

79 def invalidate(self, key: str) -> None: 

80 """Invalidate a specific key from cache.""" 

81 if key in self._cache: 

82 del self._cache[key] 

83 

84 # If this computed property is also a source for others, trigger them too! 

85 # (Propagation of invalidation) 

86 self.trigger(key) 

87 

88 def get_cached(self, key: str) -> Any: 

89 return self._cache.get(key) 

90 

91 def set_cache(self, key: str, value: Any) -> None: 

92 self._cache[key] = value 

93 

94 def has_cache(self, key: str) -> bool: 

95 return key in self._cache 

96 

97 def clear(self) -> None: 

98 """Clear all state (testing mostly).""" 

99 self._subscribers.clear() 

100 self._cache.clear() 

101 

102 

103async def get_dependency_tracker() -> DependencyTracker: 

104 """Get the global dependency tracker instance.""" 

105 from lexigram.admin.lib.di import get_admin_resolver 

106 

107 resolver = get_admin_resolver() 

108 return await resolver.resolve(DependencyTracker) 

109 

110 

111def __getattr__(name: str) -> Any: 

112 if name == "dependency_tracker": 

113 raise AttributeError( 

114 "dependency_tracker is now async. Use: await get_dependency_tracker()", 

115 ) 

116 raise AttributeError(f"module {__name__} has no attribute {name}")