Coverage for src/lexigram/web/routing/route_handlers.py: 13%

1"""Registry for route handlers in lexigram-web.""" 

2 

3from __future__ import annotations 

4 

5import inspect 

6from typing import TYPE_CHECKING, Any, Protocol, cast 

7 

8from lexigram.logging import get_logger 

9 

10logger = get_logger(__name__) 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable 

14 

15 from lexigram.app.base import Application 

16 from lexigram.web.routing.manager import WebRouterManager 

17 

18 

19class RouteHandlerProtocol(Protocol): 

20 """Protocol for route handlers.""" 

21 

22 async def register(self, manager: WebRouterManager, app: Application) -> None: 

23 """Register routes into the application.""" 

24 ... 

25 

26 

27def _is_body_model(annotation: type) -> bool: 

28 """Return True if *annotation* is a Pydantic BaseModel or DomainModel subclass. 

29 

30 Both types can be deserialised from a JSON request body. Imported lazily 

31 to avoid circular-dependency issues at module load time. 

32 

33 Args: 

34 annotation: The type annotation to test. 

35 

36 Returns: 

37 True when the annotation is a recognised model type. 

38 """ 

39 try: 

40 from pydantic import BaseModel 

41 

42 if issubclass(annotation, BaseModel): 

43 return True 

44 except ImportError: 

45 pass 

46 try: 

47 from lexigram.domain import DomainModel 

48 

49 if issubclass(annotation, DomainModel): 

50 return True 

51 except ImportError: 

52 pass 

53 return False 

54 

55 

56def _wrap_script_handler(fn: Callable[..., Any]) -> Callable[..., Any]: 

57 """Wrap a script-mode standalone handler to be Starlette-compatible. 

58 

59 Handles parameter binding: 

60 - Path params: matched by name from ``request.path_params`` 

61 - Body param: ``data: dict`` / Pydantic ``BaseModel`` / ``DomainModel`` subclass 

62 (or any unhinted param) ← parsed from JSON body 

63 - ``request``: passes the raw Starlette ``Request`` object 

64 Return values: dict/list → JSONResponse, None → 204, (body, status) tuple 

65 """ 

66 from starlette.requests import Request 

67 from starlette.responses import JSONResponse, Response 

68 

69 sig = inspect.signature(fn) 

70 params = list(sig.parameters.values()) 

71 

72 async def endpoint(request: Request) -> Response: 

73 kwargs: dict[str, Any] = {} 

74 

75 # Parse JSON body once (only for methods that carry a body) 

76 body: dict | list | None = None 

77 if request.method in ("POST", "PUT", "PATCH"): 

78 try: 

79 body = await request.json() 

80 except (ValueError, UnicodeDecodeError): 

81 body = None 

82 

83 for param in params: 

84 name = param.name 

85 annotation = param.annotation 

86 

87 if annotation is inspect.Parameter.empty: 

88 # Unnamed hint: treat as body if body available, else skip 

89 if body is not None: 

90 kwargs[name] = body 

91 elif annotation is Request or ( 

92 isinstance(annotation, type) and issubclass(annotation, Request) 

93 ): 

94 kwargs[name] = request 

95 elif name in request.path_params: 

96 kwargs[name] = request.path_params[name] 

97 elif annotation is dict or ( 

98 isinstance(annotation, type) and issubclass(annotation, dict) 

99 ): 

100 kwargs[name] = body or {} 

101 elif isinstance(annotation, type) and _is_body_model(annotation): 

102 body_dict: dict[str, Any] = body if isinstance(body, dict) else {} 

103 try: 

104 from pydantic import BaseModel, ValidationError 

105 

106 if issubclass(annotation, BaseModel): 

107 try: 

108 kwargs[name] = annotation(**body_dict) 

109 except ValidationError as exc: 

110 return JSONResponse( 

111 {"errors": exc.errors()}, status_code=422 

112 ) 

113 else: 

114 kwargs[name] = annotation(**body_dict) 

115 except ImportError: 

116 kwargs[name] = body_dict 

117 elif name in request.query_params: 

118 raw = request.query_params[name] 

119 # Best-effort type coercion for query params 

120 if annotation in (int,): 

121 try: 

122 kwargs[name] = int(raw) 

123 except ValueError: 

124 kwargs[name] = raw 

125 elif annotation in (float,): 

126 try: 

127 kwargs[name] = float(raw) 

128 except ValueError: 

129 kwargs[name] = raw 

130 elif annotation in (bool,): 

131 kwargs[name] = raw.lower() in ("1", "true", "yes") 

132 else: 

133 kwargs[name] = raw 

134 elif isinstance(annotation, type) and not annotation.__module__.startswith( 

135 "builtins" 

136 ): 

137 # Try to resolve arbitrary service from DI container (script-mode DI injection) 

138 try: 

139 from lexigram.contracts.exceptions.container import ( 

140 UnresolvableDependencyError, 

141 ) 

142 

143 container = getattr( 

144 getattr(request.app, "state", None), "container", None 

145 ) 

146 if container is not None: 

147 kwargs[name] = await container.resolve(annotation) 

148 except (UnresolvableDependencyError, AttributeError, LookupError): 

149 pass 

150 elif param.default is not inspect.Parameter.empty: 

151 # Optional param with default — skip (let Python fill default) 

152 pass 

153 

154 result = await fn(**kwargs) 

155 

156 # Tuple (body, status_code) 

157 if isinstance(result, tuple) and len(result) == 2: 

158 body_out, status = result 

159 if isinstance(body_out, Response): 

160 return body_out 

161 return JSONResponse(body_out, status_code=int(status)) 

162 

163 if result is None: 

164 return Response(status_code=204) 

165 if isinstance(result, Response): 

166 return result 

167 return JSONResponse(result) 

168 

169 return endpoint 

170 

171 

172def _wrap_websocket_handler(handler_cls: type) -> Callable: 

173 """Wrap a AbstractWebSocketHandler class to be Starlette/ASGI compatible. 

174 

175 Resolves the handler instance from the DI container (preferring singletons 

176 for broadcast support) and dispatches the connection to his ``handle()`` 

177 method. 

178 """ 

179 

180 from lexigram.web.transport.websockets import WebSocket 

181 

182 async def endpoint(starlette_ws: Any) -> None: 

183 # Resolve handler from container if available 

184 container = None 

185 if hasattr(starlette_ws, "app"): 

186 container = getattr( 

187 getattr(starlette_ws.app, "state", None), "container", None 

188 ) 

189 

190 handler: Any = None 

191 if container is not None: 

192 try: 

193 # Try to resolve from container (preferred for singleton/lifecycle) 

194 handler = await container.resolve(handler_cls) 

195 except (LookupError, RuntimeError): 

196 # Fallback to direct instantiation if not injectable 

197 handler = handler_cls() 

198 else: 

199 handler = handler_cls() 

200 

201 # Wrap Starlette WebSocket in Lexigram WebSocket 

202 ws = WebSocket(starlette_ws) 

203 

204 # Dispatch to Lexigram handler 

205 await handler.handle(ws) 

206 

207 return endpoint 

208 

209 

210class CoreRouteHandler: 

211 """Handles core routes from application decorators.""" 

212 

213 async def register(self, manager: WebRouterManager, app: Application) -> None: 

214 starlette = manager.provider.starlette 

215 if starlette is None: 

216 raise RuntimeError("Starlette application not initialized") 

217 

218 # `app` here is the Starlette instance. Resolve the Lexigram Application 

219 # from the container to access script-mode _pending_routes. 

220 lex_app: Any = app 

221 try: 

222 from lexigram.app.base import Application as LexigramApp 

223 from lexigram.contracts.exceptions.container import ( 

224 UnresolvableDependencyError, 

225 ) 

226 

227 starlette_app = manager.provider.starlette 

228 container = getattr(starlette_app, "state", None) 

229 container = getattr(container, "container", None) if container else None 

230 if container is not None: 

231 resolved = await container.resolve( 

232 LexigramApp, 

233 bypass_visibility=True, 

234 ) 

235 if resolved is not None: 

236 lex_app = resolved 

237 except ( 

238 LookupError, 

239 RuntimeError, 

240 AttributeError, 

241 TypeError, 

242 UnresolvableDependencyError, 

243 ) as _exc: 

244 logger.debug("quickstart_app_resolution_skipped", reason=str(_exc)) 

245 

246 pending = getattr(lex_app, "_pending_routes", []) 

247 if not isinstance(pending, (list, tuple)): 

248 pending = [] 

249 

250 if not pending: 

251 return 

252 

253 logger.info("Registering %d core routes from application", len(pending)) 

254 

255 for route_def in pending: 

256 # WEBSOCKET routes should be wrapped using _wrap_websocket_handler 

257 # to bridge the ASGI interface (REVISION COR-1) 

258 if route_def.method == "WEBSOCKET": 

259 use_handler = _wrap_websocket_handler(route_def.handler) 

260 else: 

261 use_handler = _wrap_script_handler(route_def.handler) 

262 

263 await manager.add_route( 

264 path=str(route_def.path), 

265 handler=use_handler, 

266 method=str(route_def.method), 

267 origin_type="core", 

268 handler_metadata=route_def.handler, 

269 ) 

270 

271 # Consume pending routes to prevent double registration (REVISION MAJ-6) 

272 if hasattr(lex_app, "_pending_routes"): 

273 lex_app._pending_routes = [] 

274 

275 

276class ControllerRouteHandler: 

277 """Handles controller-based routes. 

278 

279 Discovers controllers from two sources and merges them: 

280 1. Controllers registered directly on the ``WebProvider`` instance. 

281 2. Controllers declared in the ``controllers=`` field of any ``@module`` 

282 decorated class that is registered with the Application container. 

283 """ 

284 

285 async def register(self, manager: WebRouterManager, app: Application) -> None: 

286 # Start with directly-registered controllers 

287 seen: set[type] = set() 

288 controllers_to_register: list[type] = [] 

289 

290 for ctrl in manager.provider.controllers: 

291 if ctrl not in seen: 

292 seen.add(ctrl) 

293 controllers_to_register.append(ctrl) 

294 

295 # Auto-discover controllers from @module-decorated classes 

296 try: 

297 from lexigram.di.module import Module as LexigramModule 

298 

299 if hasattr(app, "container") and app.container is not None: 

300 # Iterate all registered singletons — find module metadata owners 

301 # Use the registry to discover all registered classes 

302 container = cast("Any", app.container) 

303 for descriptor in list(container._registry.all()): 

304 registered_cls = descriptor.service_type 

305 meta = getattr(registered_cls, "__lexigram_module__", None) 

306 if meta is not None and isinstance(meta, LexigramModule): 

307 for ctrl_cls in meta.controllers: 

308 if ctrl_cls not in seen: 

309 seen.add(ctrl_cls) 

310 controllers_to_register.append(ctrl_cls) 

311 logger.debug( 

312 "auto_discovered_controller_from_module", 

313 controller=ctrl_cls.__name__, 

314 module=meta.name, 

315 ) 

316 except ( 

317 ImportError, 

318 AttributeError, 

319 LookupError, 

320 RuntimeError, 

321 TypeError, 

322 ) as exc: 

323 logger.debug("module_controller_discovery_skipped", reason=str(exc)) 

324 

325 for controller_cls in controllers_to_register: 

326 await manager.register_controller_routes(controller_cls, app.container) 

327 

328 

329class OpenAPIRouteHandler: 

330 """Handles OpenAPI documentation routes.""" 

331 

332 async def register(self, manager: WebRouterManager, _app: Application) -> None: 

333 web_config = manager.provider.web_config 

334 openapi_title = web_config.openapi_title 

335 

336 if openapi_title and manager.provider.openapi_generator: 

337 from lexigram.web.routing.openapi import ( 

338 register_openapi_routes, 

339 ) 

340 

341 register_openapi_routes(manager.provider) # type: ignore[arg-type] 

342 

343 

344class HealthRouteHandler: 

345 """Handles health check routes.""" 

346 

347 async def register(self, manager: WebRouterManager, app: Application) -> None: 

348 from lexigram.web.routing.health import register_health_route 

349 

350 register_health_route(manager.provider, app) # type: ignore[arg-type] 

351 

352 

353class DebugRouteHandler: 

354 """Handles debug routes when enabled.""" 

355 

356 async def register(self, manager: WebRouterManager, _app: Application) -> None: 

357 if manager.should_enable_debug_routes(): 

358 from lexigram.web.routing.debug import ( 

359 register_debug_routes, 

360 ) 

361 

362 register_debug_routes(manager.provider) # type: ignore[arg-type] 

363 

364 

365class RouteHandlerRegistry: 

366 """Registry for route source handlers. 

367 

368 Each handler (controllers, OpenAPI, health, debug) knows how to 

369 discover and register its routes with the router manager. 

370 """ 

371 

372 def __init__(self) -> None: 

373 self._handlers: list[RouteHandlerProtocol] = [ 

374 CoreRouteHandler(), 

375 ControllerRouteHandler(), 

376 OpenAPIRouteHandler(), 

377 HealthRouteHandler(), 

378 DebugRouteHandler(), 

379 ] 

380 

381 def add_handler(self, handler: RouteHandlerProtocol) -> None: 

382 """Add a custom route source handler. 

383 

384 **Public Extension Point** — use this to inject custom route discovery 

385 logic without subclassing ``WebProvider``. Custom handlers run after 

386 all built-in handlers, so they can reference routes registered by the 

387 framework. 

388 

389 Register via the provider:: 

390 

391 class MyRouteHandler: 

392 async def register_routes(self, router_manager, container): 

393 router_manager.add_route("/my/path", my_view, methods=["GET"]) 

394 

395 class AppProvider(WebProvider): 

396 async def boot(self, container): 

397 await super().boot(container) 

398 route_handler_registry = await container.resolve(RouteHandlerRegistry) 

399 route_handler_registry.add_handler(MyRouteHandler()) 

400 

401 Args: 

402 handler: A :class:`RouteHandlerProtocol` implementation. 

403 """ 

404 self._handlers.append(handler) 

405 

406 @property 

407 def handlers(self) -> list[RouteHandlerProtocol]: 

408 """Get all registered handlers.""" 

409 return self._handlers