Coverage for src/lexigram/web/routing/registry.py: 32%

1"""Route registration and discovery. 

2 

3This module provides the RouteRegistry. 

4 

5Note: ControllerRegistry (in controller_registry.py) is the canonical source 

6for controller classes. RouteRegistry extends this to track route-level metadata. 

7Both registries work together: ControllerRegistry stores the controller classes 

8while RouteRegistry stores the resolved route paths and methods. 

9""" 

10 

11from __future__ import annotations 

12 

13import inspect 

14from typing import TYPE_CHECKING, Any 

15 

16from lexigram.logging import get_logger 

17from lexigram.primitives.registry import Registry 

18from lexigram.web.routing.controllers import Controller 

19 

20if TYPE_CHECKING: 

21 from lexigram.web.routing.controller_registry import ControllerRegistry 

22 

23logger = get_logger(__name__) 

24 

25 

26class RouteRegistry(Registry[str, dict[str, Any]]): 

27 """Registry for managing route registrations and discovery. 

28 

29 Canonical source for routes: RouteRegistry. 

30 Use RouteRegistry for route-level operations (path, method, handler). 

31 Use ControllerRegistry for controller-level operations (class registration). 

32 """ 

33 

34 def __init__(self, debug: bool = False): 

35 super().__init__(name="routes") 

36 self._controllers: list[type[Controller]] = [] 

37 self._debug = debug 

38 

39 def register_controller( 

40 self, 

41 controller_cls: type[Controller], 

42 controller_registry: ControllerRegistry | None = None, 

43 ) -> None: 

44 """Register a controller class and synchronize to ControllerRegistry. 

45 

46 Args: 

47 controller_cls: The controller class to register. 

48 controller_registry: If provided, also registers the class by name 

49 in the ControllerRegistry for atomic synchronization. This ensures 

50 a controller exists in both registries, preventing lookup inconsistencies. 

51 """ 

52 if controller_cls not in self._controllers: 

53 self._controllers.append(controller_cls) 

54 

55 # Register routes from controller 

56 routes = ( 

57 controller_cls.collect_routes() 

58 if hasattr(controller_cls, "collect_routes") 

59 else [] 

60 ) 

61 for route_meta in routes: 

62 self._register_route(controller_cls, route_meta) 

63 

64 # Synchronize to ControllerRegistry if provided 

65 # Only sync if not already registered to avoid RegistryAlreadyExistsError 

66 if controller_registry is not None: 

67 name = controller_cls.__name__ 

68 if controller_registry.get(name) is None: 

69 controller_registry.register(name, controller_cls) 

70 

71 def _register_route( 

72 self, 

73 controller_cls: type[Controller], 

74 route_meta: dict[str, Any], 

75 ) -> None: 

76 """Register a single route""" 

77 path = str(route_meta["path"]) 

78 method = str(route_meta["method"]).upper() 

79 prefix = str(getattr(controller_cls, "prefix", "")).rstrip("/") 

80 

81 # Prepend prefix 

82 if prefix: 

83 if not path.startswith("/"): 

84 path = f"/{path}" 

85 path = prefix if path == "/" else f"{prefix}{path}" 

86 

87 # Ensure non-empty path 

88 if not path: 

89 path = "/" 

90 

91 if path not in self._items: 

92 self._items[path] = {} 

93 

94 if method in self._items[path]: 

95 # Duplicate registrations can occur during test collection or when 

96 # modules are imported multiple times. Treat as no-op to improve DX. 

97 logger.debug( 

98 "Route %s %s already registered, ignoring duplicate", 

99 method, 

100 path, 

101 ) 

102 return 

103 

104 # Capture best-effort origin metadata for diagnostics (only when debug mode is enabled) 

105 registered_file = None 

106 registered_line = None 

107 registered_stack = None 

108 if self._debug: 

109 try: 

110 stack = inspect.stack() 

111 if len(stack) > 1: 

112 registered_file = stack[1].filename 

113 registered_line = stack[1].lineno 

114 registered_stack = [ 

115 f"{fi.function!s}@{fi.filename!s}:{fi.lineno!s}" 

116 for fi in stack[1:6] 

117 ] 

118 except (RuntimeError, OSError, AttributeError) as e: 

119 logger.debug( 

120 "Failed to capture registration stack metadata for route %s %s: %s", 

121 method, 

122 path, 

123 e, 

124 ) 

125 

126 self._items[path][method] = { 

127 "controller": controller_cls, 

128 "handler_name": route_meta["handler_name"], 

129 "response_model": route_meta.get("response_model"), 

130 "status_code": route_meta.get("status_code", 200), 

131 "summary": route_meta.get("summary"), 

132 "description": route_meta.get("description"), 

133 "tags": route_meta.get("tags", []), 

134 "registered_file": registered_file, 

135 "registered_line": registered_line, 

136 "registered_stack": registered_stack, 

137 } 

138 

139 def get_all_routes(self) -> dict[str, dict[str, Any]]: 

140 """Get all registered routes""" 

141 return self._items.copy() 

142 

143 def get_controllers(self) -> list[type[Controller]]: 

144 """Get all registered controllers""" 

145 return self._controllers.copy() 

146 

147 def find_route(self, path: str, method: str) -> dict[str, Any] | None: 

148 """Find a route by path and method""" 

149 path_routes = self._items.get(path) 

150 if path_routes: 

151 return path_routes.get(str(method).upper()) 

152 return None 

153 

154 def get_routes_by_controller( 

155 self, 

156 controller_cls: type[Controller], 

157 ) -> list[dict[str, Any]]: 

158 """Get all routes for a specific controller""" 

159 routes = [] 

160 for path, methods in self._items.items(): 

161 for method, route_info in methods.items(): 

162 if route_info["controller"] == controller_cls: 

163 routes.append({"path": path, "method": method, **route_info}) 

164 return routes 

165 

166 def clear(self) -> None: 

167 """Clear all registrations""" 

168 super().clear() 

169 self._controllers.clear() 

170 

171 

172# Global registry instance — written to by @route decorators at import time. 

173# WebProvider registers this instance with the container so DI resolution 

174# and direct module access always return the same object. 

175route_registry = RouteRegistry() 

176 

177 

178def register_controller(controller_cls: type[Controller]) -> None: 

179 """Convenience function to register a controller. 

180 

181 Registers the controller in both RouteRegistry and ControllerRegistry, 

182 ensuring synchronization and preventing lookup inconsistencies. 

183 """ 

184 from lexigram.web.routing.controller_registry import controller_registry 

185 

186 route_registry.register_controller( 

187 controller_cls, 

188 controller_registry=controller_registry, 

189 ) 

190 

191 

192def get_route_registry() -> RouteRegistry: 

193 """Get the global route registry.""" 

194 return route_registry