Coverage for src / lexigram / contracts / core / di.py: 0%

27 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-08-15 18:57 +0800

1"""Dependency injection protocols for Lexigram Framework.""" 

2 

3from __future__ import annotations 

4 

5from typing import ( 

6 TYPE_CHECKING, 

7 Any, 

8 Protocol, 

9 TypeVar, 

10 overload, 

11 runtime_checkable, 

12) 

13 

14if TYPE_CHECKING: 

15 from collections.abc import Awaitable, Callable 

16 

17 from lexigram.contracts.exceptions.container import OrphanedRegistration 

18 

19 

20T = TypeVar("T") 

21 

22 

23@runtime_checkable 

24class ContainerRegistrarProtocol(Protocol): 

25 """Protocol for registering dependencies in the container. 

26 

27 Used during the registration phase where resolution is not yet permitted. 

28 """ 

29 

30 def transient( 

31 self, 

32 service_type: type[T], 

33 factory: Any, 

34 validate: bool = True, 

35 ) -> None: 

36 """Register a transient service.""" 

37 ... 

38 

39 @overload 

40 def singleton( 

41 self, 

42 service_type: type[T], 

43 instance: T | None = None, 

44 *, 

45 name: str | None = None, 

46 factory: Any | None = None, 

47 validate: bool = True, 

48 ) -> None: ... 

49 

50 @overload 

51 def singleton( 

52 self, 

53 service_type: Any, 

54 instance: Any = None, 

55 *, 

56 name: str | None = None, 

57 factory: Any | None = None, 

58 validate: bool = True, 

59 ) -> None: ... 

60 

61 def singleton( 

62 self, 

63 service_type: Any, 

64 instance: Any = None, 

65 *, 

66 name: str | None = None, 

67 factory: Any | None = None, 

68 validate: bool = True, 

69 ) -> None: 

70 """Register a singleton service (shared instance). 

71 

72 Args: 

73 service_type: The abstract type (protocol/class) being registered. 

74 Accepts both concrete classes (``type[T]`` with full type 

75 inference) and Protocol types (via ``Any`` fallback). 

76 instance: Pre-built singleton instance. 

77 name: Optional string key. When provided, the binding is stored 

78 under this name instead of ``service_type``. Resolve via 

79 ``Annotated[T, Named(name)]``. 

80 factory: Factory callable for lazy creation. 

81 validate: Whether to validate protocol conformance. 

82 """ 

83 ... 

84 

85 def scoped( 

86 self, 

87 service_type: type[T], 

88 factory: Any, 

89 validate: bool = True, 

90 *, 

91 name: str | None = None, 

92 ) -> None: 

93 """Register a scoped service (one instance per scope). 

94 

95 Args: 

96 service_type: The abstract type being registered. 

97 factory: Factory callable, called once per scope. 

98 validate: Whether to validate protocol conformance. 

99 name: Optional string key for named scoped registration. 

100 Resolve via ``Annotated[T, Named(name)]``. 

101 """ 

102 ... 

103 

104 def has(self, service_type: Any) -> bool: 

105 """Check if a service is registered.""" 

106 ... 

107 

108 def bind( 

109 self, 

110 service_type: type[T], 

111 instance: T, 

112 ) -> None: 

113 """Bind a pre-built singleton instance, overwriting any existing binding. 

114 

115 Unlike ``singleton()``, this works on frozen containers. Designed for 

116 updating singleton instances during the boot phase (e.g. wrapping a 

117 store with a tenancy decorator). 

118 

119 The service *must* already be registered as a singleton. 

120 

121 Args: 

122 service_type: The registered service type to rebind. 

123 instance: The replacement singleton instance. 

124 """ 

125 ... 

126 

127 

128@runtime_checkable 

129class ContainerResolverProtocol(Protocol): 

130 """Protocol for resolving dependencies from the container. 

131 

132 Async-first code should depend on this protocol. It provides only an 

133 asynchronous ``resolve()`` method and makes no guarantees about sync 

134 behaviour. All synchronous helpers have been removed to keep the API 

135 simple and forward‑looking. 

136 

137 Used during the boot phase and runtime after registration is complete. 

138 """ 

139 

140 @overload 

141 async def resolve( 

142 self, 

143 service_type: type[T], 

144 *, 

145 bypass_visibility: bool = False, 

146 ) -> T: ... 

147 

148 @overload 

149 async def resolve( 

150 self, 

151 service_type: Any, 

152 *, 

153 bypass_visibility: bool = False, 

154 ) -> Any: ... 

155 

156 async def resolve( 

157 self, 

158 service_type: Any, 

159 *, 

160 bypass_visibility: bool = False, 

161 ) -> Any: 

162 """Asynchronously resolve a service by its type. 

163 

164 The framework is async-first and does not support string or other 

165 loose keys. Consumers should only register and resolve using types 

166 so that the DI container remains fully type-safe. 

167 

168 Accepts both concrete classes (``type[T]`` with full return-type 

169 inference) and Protocol types (via ``Any`` fallback). 

170 

171 Args: 

172 service_type: The service type to resolve. 

173 bypass_visibility: If True, skip module visibility enforcement. 

174 Use only in framework-internal resolution paths. 

175 """ 

176 ... 

177 

178 async def call( 

179 self, 

180 func: Callable[..., Awaitable[T] | T], 

181 *args: Any, 

182 **kwargs: Any, 

183 ) -> T: 

184 """Call a function with dependency injection.""" 

185 ... 

186 

187 def create_scope(self) -> Any: 

188 """Create a request-scoped resolution context.""" 

189 ... 

190 

191 def has(self, service_type: Any) -> bool: 

192 """Check if a service is registered.""" 

193 ... 

194 

195 @overload 

196 async def resolve_optional(self, service_type: type[T]) -> T | None: ... 

197 

198 @overload 

199 async def resolve_optional(self, service_type: Any) -> Any | None: ... 

200 

201 async def resolve_optional(self, service_type: Any) -> Any | None: 

202 """Resolve a service, returning None if not registered. 

203 

204 This provides a graceful way to handle optional dependencies 

205 without catching exceptions. 

206 

207 Args: 

208 service_type: The type to resolve. 

209 

210 Returns: 

211 The resolved instance or None if the service is not registered. 

212 """ 

213 ... 

214 

215 @overload 

216 async def resolve_all(self, service_type: type[T]) -> list[T]: ... 

217 

218 @overload 

219 async def resolve_all(self, service_type: Any) -> list[Any]: ... 

220 

221 async def resolve_all(self, service_type: Any) -> list[Any]: 

222 """Resolve all registered implementations that are subtypes of a service type. 

223 

224 Useful for collecting multiple implementations of a protocol or 

225 abstract base class (e.g. all registered ``EventHandlerProtocol`` instances). 

226 

227 Args: 

228 service_type: The base type whose implementations to resolve. 

229 

230 Returns: 

231 A list of resolved instances. 

232 """ 

233 ... 

234 

235 

236@runtime_checkable 

237class ContainerValidationProtocol(Protocol): 

238 """Protocol for container validation methods. 

239 

240 Development-time validators for detecting configuration issues. 

241 """ 

242 

243 def validate(self) -> list[str]: 

244 """Validate the container configuration. 

245 

246 Checks: 

247 - All registered services have resolvable dependencies 

248 - No circular dependencies 

249 - No scope violations 

250 

251 Returns: 

252 List of validation issues (empty if valid). 

253 """ 

254 ... 

255 

256 def validate_no_orphans(self) -> list[OrphanedRegistration]: 

257 """Find registrations that no other service depends on. 

258 

259 Identifies dead code services registered but never used. 

260 

261 Returns: 

262 List of potentially orphaned registrations. 

263 """ 

264 ... 

265 

266 

267@runtime_checkable 

268class BootContainerProtocol( 

269 ContainerRegistrarProtocol, 

270 ContainerResolverProtocol, 

271 Protocol, 

272): 

273 """Container interface for the provider boot phase. 

274 

275 During boot, providers need to both resolve existing services (e.g. 

276 retrieving a raw LLM client) and register new ones (e.g. wrapping 

277 that client with observability decorators). This protocol exposes 

278 exactly those two capabilities without granting access to validation 

279 or lifecycle methods. 

280 

281 Any object implementing both :class:`ContainerRegistrarProtocol` and 

282 :class:`ContainerResolverProtocol` satisfies this protocol via 

283 structural subtyping. The concrete :class:`Container` class is the 

284 primary implementation. 

285 """ 

286 

287 

288@runtime_checkable 

289class ContainerProtocol( 

290 ContainerRegistrarProtocol, 

291 ContainerResolverProtocol, 

292 ContainerValidationProtocol, 

293 Protocol, 

294): 

295 """Combined DI container protocol covering registration, resolution, and validation. 

296 

297 Use this instead of ``Any`` when a component needs to both register 

298 services (via :meth:`singleton`/transient) and resolve them (via 

299 :meth:`resolve`) within the same method. This is a structural Protocol 

300 — any object implementing both :class:`ContainerRegistrarProtocol` and 

301 :class:`ContainerResolverProtocol` satisfies this type. 

302 """ 

303 

304 

305__all__ = [ 

306 "BootContainerProtocol", 

307 "ContainerProtocol", 

308 "ContainerRegistrarProtocol", 

309 "ContainerResolverProtocol", 

310 "ContainerValidationProtocol", 

311]