Coverage for src/lexigram/admin/core/decorators.py: 77%

1"""DI decorators and lifecycle management for lexigram-admin. 

2 

3This module provides decorator integration with lexigram's DI system, 

4including @singleton, @scoped, and @transient decorators. 

5 

6FWK-02: Apply lifecycle decorators to services. 

7""" 

8 

9from __future__ import annotations 

10 

11from collections.abc import Callable 

12from enum import Enum 

13from typing import Any, Self, TypeVar 

14 

15T = TypeVar("T") 

16F = TypeVar("F", bound=Callable[..., Any]) 

17 

18 

19# ============================================================================ 

20# Try importing from lexigram 

21# ============================================================================ 

22 

23try: 

24 from lexigram.di.decorators import scoped as lexigram_scoped 

25 from lexigram.di.decorators import singleton as lexigram_singleton 

26 from lexigram.di.decorators import transient as lexigram_transient 

27 

28 HAS_LEX_DI = True 

29except ImportError: 

30 HAS_LEX_DI = False 

31 lexigram_singleton = None # type: ignore[assignment] 

32 lexigram_scoped = None # type: ignore[assignment] 

33 lexigram_transient = None # type: ignore[assignment] 

34 

35 

36# ============================================================================ 

37# Lifecycle Scope 

38# ============================================================================ 

39 

40 

41class Scope(str, Enum): 

42 """Service lifecycle scopes.""" 

43 

44 SINGLETON = "singleton" # One instance per container 

45 SCOPED = "scoped" # One instance per scope (request) 

46 TRANSIENT = "transient" # New instance each time 

47 

48 

49# ============================================================================ 

50# Fallback Decorators 

51# ============================================================================ 

52 

53_singletons: dict[type, Any] = {} 

54 

55 

56def _fallback_singleton(cls: type[T]) -> type[T]: 

57 """Fallback singleton decorator using class-level caching.""" 

58 

59 def new_method(cls_inner: type[T], *args: Any, **kwargs: Any) -> T: 

60 if cls_inner not in _singletons: 

61 instance = object.__new__(cls_inner) 

62 _singletons[cls_inner] = instance 

63 return _singletons[cls_inner] 

64 

65 cls.__new__ = new_method # type: ignore[method-assign,assignment] 

66 return cls 

67 

68 

69def _fallback_scoped(cls: type[T]) -> type[T]: 

70 """Fallback scoped decorator (acts like transient without request context).""" 

71 # Without lexigram's context system, scoped behaves like transient 

72 return cls 

73 

74 

75def _fallback_transient(cls: type[T]) -> type[T]: 

76 """Fallback transient decorator (no-op).""" 

77 return cls 

78 

79 

80# ============================================================================ 

81# Public Decorators 

82# ============================================================================ 

83 

84 

85def singleton(cls: type[T]) -> type[T]: 

86 """Mark a class as singleton (one instance per container). 

87 

88 Example: 

89 >>> @singleton 

90 ... class ConfigService: 

91 ... def __init__(self): 

92 ... self.settings = load_settings() 

93 """ 

94 if HAS_LEX_DI and lexigram_singleton: # type: ignore[truthy-function] 

95 return lexigram_singleton(cls) 

96 return _fallback_singleton(cls) 

97 

98 

99def scoped(cls: type[T]) -> type[T]: 

100 """Mark a class as scoped (one instance per request). 

101 

102 Example: 

103 >>> @scoped 

104 ... class RequestContext: 

105 ... def __init__(self, request): 

106 ... self.user = request.user 

107 """ 

108 if HAS_LEX_DI and lexigram_scoped: # type: ignore[truthy-function] 

109 return lexigram_scoped(cls) 

110 return _fallback_scoped(cls) 

111 

112 

113def transient(cls: type[T]) -> type[T]: 

114 """Mark a class as transient (new instance each time). 

115 

116 Example: 

117 >>> @transient 

118 ... class QueryBuilder: 

119 ... def __init__(self): 

120 ... self.conditions = [] 

121 """ 

122 if HAS_LEX_DI and lexigram_transient: # type: ignore[truthy-function] 

123 return lexigram_transient(cls) 

124 return _fallback_transient(cls) 

125 

126 

127# ============================================================================ 

128# Injectable Decorator 

129# ============================================================================ 

130 

131 

132def injectable( 

133 scope: Scope = Scope.TRANSIENT, 

134 token: str | None = None, 

135) -> Callable[[type[T]], type[T]]: 

136 """Mark a class as injectable with specified scope. 

137 

138 Args: 

139 scope: Lifecycle scope (singleton, scoped, transient) 

140 token: Optional token for registration 

141 

142 Example: 

143 >>> @injectable(scope=Scope.SINGLETON) 

144 ... class DatabaseConnection: 

145 ... pass 

146 """ 

147 

148 def decorator(cls: type[T]) -> type[T]: 

149 # Apply appropriate scope decorator 

150 if scope == Scope.SINGLETON: 

151 decorated = singleton(cls) 

152 elif scope == Scope.SCOPED: 

153 decorated = scoped(cls) 

154 else: 

155 decorated = transient(cls) 

156 

157 # Store metadata 

158 decorated.__injectable_scope__ = scope # type: ignore[attr-defined] 

159 decorated.__injectable_token__ = token # type: ignore[attr-defined] 

160 

161 return decorated 

162 

163 return decorator 

164 

165 

166# ============================================================================ 

167# Context Manager for Scopes 

168# ============================================================================ 

169 

170 

171class ScopeContext: 

172 """Context manager for managing scoped instances. 

173 

174 Example: 

175 >>> async with ScopeContext() as ctx: 

176 ... service = ctx.resolve(RequestScopedService) 

177 ... # Service is scoped to this context 

178 """ 

179 

180 def __init__(self) -> None: 

181 self._instances: dict[type, Any] = {} 

182 

183 def __enter__(self) -> Self: 

184 return self 

185 

186 def __exit__(self, *args: object) -> None: 

187 self._instances.clear() 

188 

189 async def __aenter__(self) -> Self: 

190 return self 

191 

192 async def __aexit__(self, *args: object) -> None: 

193 # Cleanup async resources 

194 for instance in self._instances.values(): 

195 if hasattr(instance, "close") and callable(instance.close): 

196 close = instance.close 

197 if callable(close): 

198 result = close() 

199 if hasattr(result, "__await__"): 

200 await result 

201 self._instances.clear() 

202 

203 def resolve(self, cls: type[T], *args: Any, **kwargs: Any) -> T: 

204 """Resolve an instance within this scope.""" 

205 scope = getattr(cls, "__injectable_scope__", Scope.TRANSIENT) 

206 

207 if scope == Scope.SINGLETON: 

208 # Use global singleton 

209 if cls not in _singletons: 

210 _singletons[cls] = cls(*args, **kwargs) 

211 return _singletons[cls] 

212 

213 if scope == Scope.SCOPED: 

214 # Use scope-local instance 

215 if cls not in self._instances: 

216 self._instances[cls] = cls(*args, **kwargs) 

217 return self._instances[cls] 

218 

219 # Transient - new instance 

220 return cls(*args, **kwargs) 

221 

222 

223# ============================================================================ 

224# Clear Singletons (for testing) 

225# ============================================================================ 

226 

227 

228def clear_singletons() -> None: 

229 """Clear all singleton instances (useful for testing).""" 

230 _singletons.clear() 

231 

232 

233__all__ = [ 

234 # Flags 

235 "HAS_LEX_DI", 

236 # Scope enum 

237 "Scope", 

238 # Context 

239 "ScopeContext", 

240 # Utils 

241 "clear_singletons", 

242 "injectable", 

243 "scoped", 

244 # Decorators 

245 "singleton", 

246 "transient", 

247]