Coverage for src/lexigram/web/routing/manager.py: 13%
119 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Router manager for lexigram-web."""
3from __future__ import annotations
5import inspect
6from typing import TYPE_CHECKING, Any, cast
8from lexigram.logging import get_logger
9from lexigram.web.protocols import WebProviderProtocol
11if TYPE_CHECKING:
12 from lexigram.contracts.core.di import ContainerResolverProtocol
13 from lexigram.web.routing.controllers import Controller
15logger = get_logger(__name__)
18class WebRouterManager:
19 """Manages route registration, controller discovery, and OpenAPI generation."""
21 def __init__(self, provider: WebProviderProtocol):
22 self.provider = provider
23 self._registered_routes: dict[tuple[str, str], list[dict]] = {}
24 from lexigram.web.routing.route_handlers import (
25 RouteHandlerRegistry,
26 )
28 self._route_registry = RouteHandlerRegistry()
30 def add_handler(self, handler: Any) -> None:
31 """Add a custom route handler to the registry."""
32 self._route_registry.add_handler(handler)
34 def _format_route_origin(self, origin: dict) -> str:
35 parts = []
36 t = origin.get("type")
37 if t == "core":
38 parts.append(f"handler={origin.get('handler_name')}")
39 elif t == "controller":
40 parts.append(f"controller={origin.get('controller')}")
41 parts.append(f"handler={origin.get('handler_name')}")
42 module = origin.get("module")
43 if module:
44 parts.append(f"module={module}")
45 file = origin.get("file")
46 if file:
47 parts.append(f"file={file}")
48 line = origin.get("line")
49 if line:
50 parts.append(f"line={line}")
51 return ", ".join(parts)
53 async def register_routes(
54 self, app: Any, container: ContainerResolverProtocol
55 ) -> Any:
56 """Discovers and registers all routes using the registry and modules."""
57 logger.info("Initializing web route discovery...")
59 # 1. Run custom route handlers (e.g. CoreRouteHandler, ControllerRouteHandler)
61 # 3. Run custom route handlers (e.g. legacy @route decorator handlers)
62 for handler in self._route_registry.handlers:
63 handler_name = type(handler).__name__
64 logger.debug("Running route discovery for %s", handler_name)
65 await handler.register(self, app)
67 route_count = len(self._registered_routes)
68 logger.info("Route registration complete. Total routes: %s", route_count)
70 def should_enable_debug_routes(self) -> bool:
71 cfg = self.provider.web_config
72 p_cfg = self.provider.provider_config
73 return (
74 getattr(cfg, "debug_routes", False)
75 or getattr(p_cfg, "debug_routes", False)
76 or self.provider.debug_routes_auth is not None
77 or getattr(cfg, "debug_routes_token", None) is not None
78 or getattr(p_cfg, "debug_routes_token", None) is not None
79 or getattr(cfg, "debug_routes_rate_limit", 0) > 0
80 or getattr(p_cfg, "debug_routes_rate_limit", 0) > 0
81 or getattr(self.provider, "_debug_routes_redis_client_arg", None)
82 is not None
83 )
85 async def add_route(
86 self,
87 path: str,
88 handler: Any,
89 method: str,
90 origin_type: str,
91 handler_metadata: Any = None,
92 controller_cls: type | None = None,
93 route_meta: dict[str, Any] | None = None,
94 ) -> None:
95 """Centralized route registration with duplicate detection."""
96 path = str(path)
97 method = str(method).upper()
99 starlette = self.provider.starlette
100 if starlette is None:
101 raise RuntimeError("Starlette application not initialized")
103 key = (method, path)
104 origin: dict[str, Any] = {"type": origin_type}
106 if origin_type == "core" and handler_metadata:
107 origin["handler_name"] = str(
108 getattr(
109 handler_metadata,
110 "__name__",
111 repr(handler_metadata),
112 ),
113 )
114 origin["module"] = getattr(handler_metadata, "__module__", None)
115 try:
116 code = getattr(handler_metadata, "__code__", None)
117 if code is not None:
118 origin["file"] = getattr(code, "co_filename", None)
119 origin["line"] = getattr(code, "co_firstlineno", None)
120 except (AttributeError, TypeError):
121 pass
122 elif origin_type == "controller" and controller_cls:
123 origin["controller"] = controller_cls.__name__
124 origin["handler_name"] = (
125 handler_metadata if isinstance(handler_metadata, str) else "unknown"
126 )
127 origin["module"] = controller_cls.__module__
128 try:
129 func = getattr(controller_cls, origin["handler_name"], None)
130 if func:
131 source_file = inspect.getsourcefile(func)
132 if source_file:
133 origin["file"] = source_file
134 code = getattr(func, "__code__", None)
135 if code:
136 origin["line"] = getattr(code, "co_firstlineno", None)
137 except (OSError, TypeError, AttributeError):
138 pass
140 if key in self._registered_routes:
141 existing = self._registered_routes[key][0]
142 msg = (
143 f"Duplicate route registration detected: {method} {path}\n"
144 f"Existing: {self._format_route_origin(existing)}\n"
145 f"New: {self._format_route_origin(origin)}"
146 )
147 if self.provider.fail_on_route_conflict:
148 raise RuntimeError(msg)
149 logger.warning(msg)
150 return
152 self._registered_routes[key] = [origin]
153 if method == "WEBSOCKET":
154 from starlette.routing import WebSocketRoute
156 starlette.router.routes.append(WebSocketRoute(path, handler))
157 else:
158 starlette.add_route(path, handler, methods=[method])
160 # Store Lexigram Route metadata if supported
161 if hasattr(self.provider.router, "add_route"):
162 original_handler = handler
163 if (
164 origin_type == "controller"
165 and controller_cls
166 and isinstance(handler_metadata, str)
167 ):
168 original_handler = getattr(controller_cls, handler_metadata, handler)
170 meta_kwargs = route_meta or {}
171 name = meta_kwargs.pop("name", None)
172 meta_kwargs.pop("method", None)
173 meta_kwargs.pop("path", None)
175 self.provider.router.add_route(
176 method=method,
177 path=path,
178 handler=original_handler,
179 name=name,
180 controller_cls=controller_cls,
181 **meta_kwargs,
182 )
184 logger.debug("Registered route: %s %s (origin: %s)", method, path, origin_type)
186 async def register_controller_routes(
187 self,
188 controller_cls: type[Controller],
189 container: ContainerResolverProtocol,
190 ) -> None:
191 """Registers routes for a specific controller.
193 Uses collect_routes() if available (recommended), otherwise falls back
194 to _routes attribute (legacy metaclass-based approach).
195 """
196 # Mirror the controller into the global route registry so admin
197 # pages and tooling can discover the routes actually being mounted.
198 from lexigram.web.routing import register_controller
200 register_controller(controller_cls)
202 # Collect routes from controller
203 routes = controller_cls.collect_routes()
205 prefix = getattr(controller_cls, "prefix", "")
207 for route_meta in routes:
208 create_endpoint = getattr(self.provider.router, "_create_endpoint", None)
209 if create_endpoint is None:
210 raise RuntimeError("Router missing internal endpoint creator")
212 handler = create_endpoint(
213 controller_cls,
214 route_meta["handler_name"],
215 container,
216 )
218 raw_path = str(
219 route_meta.get("path")
220 if isinstance(route_meta, dict)
221 else getattr(route_meta, "path", None),
222 )
223 method = str(
224 route_meta.get("method", "GET").upper()
225 if isinstance(route_meta, dict)
226 else getattr(route_meta, "method", "GET").upper(),
227 )
228 full_path = (
229 raw_path
230 if raw_path.startswith(prefix)
231 else prefix.rstrip("/") + raw_path
232 )
233 if not full_path.startswith("/"):
234 full_path = "/" + full_path
236 await self.add_route(
237 path=full_path,
238 handler=handler,
239 method=method,
240 origin_type="controller",
241 handler_metadata=str(route_meta.get("handler_name", "unknown")),
242 controller_cls=controller_cls,
243 route_meta=route_meta,
244 )
246 def generate_openapi_spec(self) -> dict[str, Any]:
247 """Generates OpenAPI specification."""
248 if not self.provider.openapi_generator:
249 return {}
250 return cast(
251 "dict[str, Any]",
252 self.provider.openapi_generator.generate_spec(self.provider.controllers),
253 )