Coverage for src/lexigram/web/routing/router.py: 21%
150 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Router with Controller DI Strategy.
3The Router provides declarative routing for controllers with automatic
4dependency injection. Routes are defined using decorators on controller
5methods, and the router handles request parsing, DI resolution, and
6response formatting.
8Example:
9 Defining routes with a controller::
11 from lexigram.web import Controller, get, post
13 class UserController(Controller):
14 @get("/users")
15 async def list_users(self, limit: int = 20) -> list[User]:
16 return await self.user_repo.list(limit=limit)
18 @post("/users")
19 async def create_user(self, user_data: UserCreate) -> User:
20 return await self.user_repo.create(user_data)
22 The router automatically:
23 - Resolves the controller from DI
24 - Injects dependencies into the controller
25 - Parses query parameters, path parameters, and request body
26 - Validates input using Pydantic models
27 - Wraps responses in JSONResponse
29See Also:
30 - :class:`lexigram.web.Controller`: Base controller class.
31 - :mod:`lexigram.web.routing.decorators`: HTTP method decorators.
32 - :class:`lexigram.web.middleware.Middleware`: Middleware base class.
33"""
35from collections.abc import Callable
36from functools import lru_cache
37import inspect
38from typing import (
39 Any,
40 get_origin,
41 get_type_hints,
42)
43import weakref
45from starlette.requests import Request
47from lexigram.logging import get_logger
49logger = get_logger(__name__)
52@lru_cache(maxsize=1024)
53def _cached_get_type_hints(func: Callable[..., Any]) -> Any:
54 """Cached wrapper around typing.get_type_hints(func, globalns=func.__globals__).
56 - Caching avoids repeatedly resolving forward refs at request time.
57 - Keyed by function object so it's invalidated when the function object is GC'd.
58 - Follows __wrapped__ chain to get the original function's __globals__,
59 which is necessary when decorators use @functools.wraps and the handler
60 has `from __future__ import annotations`.
61 """
62 target = func
63 while hasattr(target, "__wrapped__"):
64 target = target.__wrapped__
66 globalns = getattr(target, "__globals__", None) or getattr(func, "__globals__", {})
67 try:
68 return get_type_hints(func, globalns=globalns) or {}
69 except (NameError, AttributeError, TypeError):
70 try:
71 return get_type_hints(target) or {}
72 except (NameError, AttributeError, TypeError):
73 return {}
76from dataclasses import dataclass, field
79@dataclass
80class Route:
81 """Metadata representing a registered route."""
83 method: str
84 path: str
85 handler: Callable
86 name: str | None = None
87 controller_cls: type | None = None
88 metadata: dict[str, Any] = field(default_factory=dict)
91from lexigram.web.filters.builtin import (
92 DefaultExceptionFilter,
93 DependencyResolutionFilter,
94 ValidationErrorFilter,
95)
96from lexigram.web.filters.pipeline import FilterPipeline
99class Router:
100 """Router with Controller DI Strategy support.
102 The Router manages route registration, controller resolution, and
103 request handling. It integrates with the DI container to provide
104 automatic dependency injection for controllers and their methods.
106 Attributes:
107 routes: List of registered routes with their handlers.
108 _controller_cache: Cache of resolved controller instances.
109 _container: DI container for resolving dependencies.
110 _signature_cache: Cache of handler signatures for performance.
111 _routes_by_name: Dictionary of routes indexed by their unique name.
112 exception_filters: Registry for handling controller exceptions.
113 """
115 def __init__(self, filter_pipeline: FilterPipeline | None = None):
116 self.routes: list[Route] = []
117 self._routes_by_name: dict[str, Route] = {}
118 self._controller_cache: weakref.WeakValueDictionary[type, Any] = (
119 weakref.WeakValueDictionary()
120 )
121 self._container: Any = None
122 # Cache for handler signatures (computed once at registration time)
123 self._signature_cache: dict[tuple[type, str], dict[str, Any]] = {}
124 # Set of (METHOD, path) tuples used to detect duplicate registrations.
125 self._route_registry: set[tuple[str, str]] = set()
126 self.exception_filters = filter_pipeline or FilterPipeline()
127 # Register core filters (last registered runs first, so Default is last fallback)
128 self.exception_filters.add_filter(DefaultExceptionFilter())
129 self.exception_filters.add_filter(DependencyResolutionFilter())
130 self.exception_filters.add_filter(ValidationErrorFilter())
132 def get_route_by_name(self, name: str) -> Route | None:
133 """Get a route by its unique name in O(1) time.
135 Args:
136 name: The unique name of the route.
138 Returns:
139 The Route object if found, otherwise None.
140 """
141 return self._routes_by_name.get(name)
143 def clear_cache(self) -> None:
144 """Clear all cached handler signatures and type-hint lookups.
146 Call this when modules are hot-reloaded so that stale signatures
147 are not used. The caches will be repopulated lazily on the next
148 request that reaches each handler.
150 Example::
152 hot_reload_manager.on_reload(router.clear_cache)
153 """
154 self._signature_cache.clear()
155 _cached_get_type_hints.cache_clear()
157 def _get_handler_signature(
158 self,
159 controller_cls: type,
160 method_name: str,
161 ) -> dict[str, Any]:
162 """Get cached handler signature or compute and cache it.
164 Args:
165 controller_cls: The controller class.
166 method_name: The method name.
168 Returns:
169 Dict with 'sig', 'hints', and computed params.
170 """
171 cache_key = (controller_cls, method_name)
172 if cache_key in self._signature_cache:
173 return self._signature_cache[cache_key]
175 handler = getattr(controller_cls, method_name)
176 sig = inspect.signature(handler)
178 # Resolve string annotations (cached)
179 try:
180 hints = _cached_get_type_hints(handler)
181 except (NameError, AttributeError, TypeError) as e:
182 hints = {}
183 logger.debug(
184 "Failed to resolve type hints for %s.%s: %s",
185 controller_cls.__name__,
186 method_name,
187 e,
188 )
190 # Compute DI params info at registration time
191 di_params = {}
192 body_param = None
194 for param_name, param in sig.parameters.items():
195 if param_name in ("self", "cls"):
196 continue
197 if param.annotation is inspect.Parameter.empty:
198 continue
200 # Determine if this is a DI-injectable param
201 origin = get_origin(param.annotation)
202 if origin is not None:
203 di_params[param_name] = param.annotation
205 result = {
206 "sig": sig,
207 "hints": hints,
208 "di_params": di_params,
209 "body_param": body_param,
210 }
212 self._signature_cache[cache_key] = result
213 return result
215 def _set_container(self, container: Any) -> None:
216 """Set the DI container and preload singleton controllers."""
217 self._container = container
218 if container is None or not hasattr(container, "resolve"):
219 return
221 # Pre-resolve and cache singleton controllers
222 for route in self.routes:
223 controller_cls = route.controller_cls
224 if controller_cls is not None:
225 self._preload_controller(controller_cls)
227 def _preload_controller(self, controller_cls: type) -> None:
228 """Preload a singleton controller if not already cached."""
229 if controller_cls in self._controller_cache:
230 return
232 if self._container is None or not hasattr(self._container, "resolve"):
233 return
235 try:
236 # Check if controller is a singleton in the container
237 if hasattr(
238 self._container,
239 "is_singleton",
240 ) and self._container.is_singleton(controller_cls):
241 import asyncio
243 # Use sync resolution for singletons if available
244 if hasattr(self._container, "resolve"):
245 try:
246 self._controller_cache[controller_cls] = (
247 self._container.resolve(controller_cls)
248 )
249 return
250 except (LookupError, RuntimeError, AttributeError) as exc:
251 logger.debug("controller_sync_resolve_failed", error=str(exc))
252 # Fall back to async
253 try:
254 try:
255 loop = asyncio.get_running_loop()
256 # Can't synchronously resolve in running loop, skip caching
257 return
258 except RuntimeError:
259 # No running loop, use get_event_loop() for sync operations
260 loop = asyncio.new_event_loop()
261 self._controller_cache[controller_cls] = (
262 loop.run_until_complete(
263 self._container.resolve(controller_cls),
264 )
265 )
266 loop.close()
267 except RuntimeError:
268 # No event loop available, skip caching
269 pass
270 except (LookupError, RuntimeError, AttributeError) as exc:
271 logger.debug(
272 "controller_cache_failed", error=str(exc)
273 ) # Skip caching if it fails
275 def _create_endpoint(
276 self,
277 controller_cls: type,
278 method_name: str,
279 container: Any,
280 ) -> Callable:
281 """
282 Creates a wrapper function that:
283 1. Resolves the Controller from DI (or uses cached singleton)
284 2. Parses request body for Pydantic models
285 3. Calls the specific method with proper parameters
286 4. Wraps the result in a JSONResponse if not already a Response
287 """
289 # OPT-WEB-2: Pre-resolve singleton controllers once at route registration time
290 cached_controller_instance = None
291 # Skip pre-resolving controllers at startup - they may have async dependencies
292 # Controllers will be resolved per-request instead
293 if container is not None:
294 try:
295 hasattr(
296 container,
297 "is_singleton",
298 ) and container.is_singleton(controller_cls)
299 # Don't try to sync resolve controllers - they may have async dependencies
300 # Fall through to per-request resolution
301 except (LookupError, RuntimeError, AttributeError) as exc:
302 logger.debug(
303 "singleton_check_failed", error=str(exc)
304 ) # Fall back to per-request resolution
306 async def endpoint(request: Request) -> Any:
307 try:
308 # 1. Resolve scoped container (from DIScopeMiddleware) or use root
309 scoped_container = getattr(request.state, "container", None)
310 effective_container = scoped_container or container
312 # 2. Resolve controller instance
313 try:
314 controller_instance = await effective_container.resolve(
315 controller_cls
316 )
317 except Exception as _resolve_err: # noqa: BLE001 — controller resolution may raise any DI error; log before re-raising
318 logger.exception("Failed to resolve controller %s", controller_cls)
319 raise
321 # 3. Get handler and pre-computed metadata
322 handler = getattr(controller_instance, method_name)
323 sig_info = self._get_handler_signature(controller_cls, method_name)
324 route = self._get_route_for_handler(handler, method_name)
326 # 4. Collect Guards, Interceptors, and Filters
327 # Attached via decorators like @use_guards, @intercept, @use_pipes, @use_filters
328 guards = getattr(handler, "__guards__", [])
329 interceptors = getattr(handler, "__interceptors__", [])
330 filters = getattr(handler, "__filters__", [])
332 # 5. Build Execution Context
333 from lexigram.web.routing.execution_context import WebExecutionContext
335 context = WebExecutionContext(
336 request=request,
337 handler=handler,
338 controller_class=controller_cls,
339 method_name=method_name,
340 route_metadata={
341 "sig_info": sig_info,
342 "route": route,
343 **(route.metadata if route else {}),
344 },
345 container=effective_container,
346 )
348 # 6. Execute through Pipeline
349 from lexigram.web.routing.pipeline import RequestPipeline
350 from lexigram.web.serialization.serializers import ResponseSerializer
352 response_serializer = await effective_container.resolve(
353 ResponseSerializer
354 )
356 pipeline = RequestPipeline(
357 guards=guards,
358 interceptors=interceptors,
359 filters=filters,
360 fallback_pipeline=self.exception_filters,
361 response_serializer=response_serializer,
362 )
363 return await pipeline.execute(context)
364 except Exception as e: # noqa: BLE001 — safety net; pipeline setup failures are routed to global exception filters
365 # This catch is now a safety net, as the pipeline should handle most things.
366 # But if resolution or pipeline setup fails, we still try global filters.
367 return await self.exception_filters.handle(e, request)
369 return endpoint
371 def _get_route_for_handler(
372 self,
373 handler: Callable,
374 method_name: str,
375 ) -> Route | None:
376 """Find the Route object for a given handler and method name."""
377 return next(
378 (
379 r
380 for r in self.routes
381 if r.handler == handler
382 or getattr(r.handler, "__name__", None) == method_name
383 ),
384 None,
385 )
387 def add_route(
388 self,
389 method: str,
390 path: str,
391 handler: Callable,
392 name: str | None = None,
393 controller_cls: type | None = None,
394 **kwargs: Any,
395 ) -> None:
396 """Add a route (used internally)"""
397 route_key = (method.upper(), path)
398 if route_key in self._route_registry:
399 logger.warning(
400 "router.duplicate_route",
401 method=method.upper(),
402 path=path,
403 message="A handler for this method+path is already registered; the earlier registration will be shadowed.",
404 )
405 self._route_registry.add(route_key)
406 route_obj = Route(
407 method=method.upper(),
408 path=path,
409 handler=handler,
410 name=name,
411 controller_cls=controller_cls,
412 metadata=kwargs,
413 )
414 self.routes.append(route_obj)
415 if name:
416 self._routes_by_name[name] = route_obj