Coverage for fastblocks / actions / gather / routes.py: 50%

1"""Route gathering functionality to replace scattered route discovery.""" 

2 

3import typing as t 

4from contextlib import suppress 

5from importlib import import_module 

6from pathlib import Path 

7 

8from acb.adapters import get_adapters, root_path 

9from acb.debug import debug 

10from anyio import Path as AsyncPath 

11from starlette.routing import Host, Mount, Route, Router, WebSocketRoute 

12 

13from .strategies import GatherStrategy, gather_with_strategy 

14 

15RouteType = Route | Router | Mount | Host | WebSocketRoute 

16 

17 

18class RouteGatherResult: 

19 def __init__( 

20 self, 

21 *, 

22 routes: list[RouteType] | None = None, 

23 adapter_routes: dict[str, list[RouteType]] | None = None, 

24 base_routes: list[RouteType] | None = None, 

25 errors: list[Exception] | None = None, 

26 ) -> None: 

27 self.routes = routes if routes is not None else [] 

28 self.adapter_routes = adapter_routes if adapter_routes is not None else {} 

29 self.base_routes = base_routes if base_routes is not None else [] 

30 self.errors = errors if errors is not None else [] 

31 

32 @property 

33 def total_routes(self) -> int: 

34 return len(self.routes) 

35 

36 @property 

37 def has_errors(self) -> bool: 

38 return len(self.errors) > 0 

39 

40 def extend_routes(self, additional_routes: list[RouteType]) -> None: 

41 self.routes.extend(additional_routes) 

42 

43 

44async def gather_routes( 

45 *, 

46 sources: list[str] | None = None, 

47 patterns: list[str] | None = None, 

48 include_base: bool = True, 

49 include_adapters: bool = True, 

50 strategy: GatherStrategy | None = None, 

51) -> RouteGatherResult: 

52 if sources is None: 

53 sources = ["adapters", "base_routes"] 

54 

55 if patterns is None: 

56 patterns = ["_routes.py", "routes.py"] 

57 

58 if strategy is None: 

59 strategy = GatherStrategy() 

60 

61 result = RouteGatherResult() 

62 

63 tasks: list[t.Coroutine[t.Any, t.Any, t.Any]] = [] 

64 

65 if "adapters" in sources and include_adapters: 

66 tasks.append(_gather_adapter_routes(patterns, strategy)) 

67 

68 if "base_routes" in sources and include_base: 

69 tasks.append(_gather_base_routes(patterns)) 

70 

71 if "custom" in sources: 

72 tasks.append(_gather_custom_routes(patterns, strategy)) 

73 

74 gather_result = await gather_with_strategy( 

75 tasks, 

76 strategy, 

77 cache_key=f"routes:{':'.join(sources)}:{':'.join(patterns)}", 

78 ) 

79 

80 for success in gather_result.success: 

81 if isinstance(success, dict): 

82 result.adapter_routes.update(success) 

83 for routes in success.values(): 

84 result.routes.extend(routes) 

85 elif isinstance(success, list): 

86 result.base_routes.extend(success) 

87 result.routes.extend(success) 

88 

89 result.errors.extend(gather_result.errors) 

90 

91 debug(f"Gathered {result.total_routes} routes from {len(sources)} sources") 

92 

93 return result 

94 

95 

96async def _gather_adapter_routes( 

97 patterns: list[str], 

98 strategy: GatherStrategy, 

99) -> dict[str, list[RouteType]]: 

100 adapter_routes: dict[str, list[RouteType]] = {} 

101 

102 for adapter in get_adapters(): 

103 await _process_adapter_routes(adapter, patterns, strategy, adapter_routes) 

104 

105 return adapter_routes 

106 

107 

108async def _process_adapter_routes( 

109 adapter: t.Any, 

110 patterns: list[str], 

111 strategy: GatherStrategy, 

112 adapter_routes: dict[str, list[RouteType]], 

113) -> None: 

114 adapter_name = adapter.name 

115 adapter_path = adapter.path.parent 

116 routes = [] 

117 

118 for pattern in patterns: 

119 routes_path = adapter_path / pattern 

120 

121 if await AsyncPath(routes_path).exists(): 

122 try: 

123 found_routes = await _extract_routes_from_file(routes_path) 

124 if found_routes: 

125 routes.extend(found_routes) 

126 debug( 

127 f"Found {len(found_routes)} routes in {adapter_name}/{pattern}", 

128 ) 

129 except Exception as e: 

130 debug(f"Error gathering routes from {adapter_name}/{pattern}: {e}") 

131 raise 

132 

133 if routes: 

134 adapter_routes[adapter_name] = routes 

135 

136 

137async def _gather_base_routes(patterns: list[str]) -> list[RouteType]: 

138 base_routes = [] 

139 for pattern in patterns: 

140 routes_path = root_path / pattern 

141 if await AsyncPath(routes_path).exists(): 

142 try: 

143 routes = await _extract_routes_from_file(Path(routes_path)) 

144 if routes: 

145 base_routes.extend(routes) 

146 debug(f"Found {len(routes)} base routes in {pattern}") 

147 except Exception as e: 

148 debug(f"Error gathering base routes from {pattern}: {e}") 

149 

150 return base_routes 

151 

152 

153async def _gather_custom_routes( 

154 patterns: list[str], 

155 strategy: GatherStrategy, 

156) -> list[RouteType]: 

157 custom_routes = [] 

158 

159 custom_paths = [ 

160 root_path / "app" / "routes.py", 

161 root_path / "custom" / "routes.py", 

162 root_path / "src" / "routes.py", 

163 ] 

164 

165 for custom_path in custom_paths: 

166 if await AsyncPath(custom_path).exists(): 

167 try: 

168 routes = await _extract_routes_from_file(Path(custom_path)) 

169 if routes: 

170 custom_routes.extend(routes) 

171 debug(f"Found {len(routes)} custom routes in {custom_path}") 

172 

173 except Exception as e: 

174 debug(f"Error gathering custom routes from {custom_path}: {e}") 

175 if strategy.error_strategy.value == "fail_fast": 

176 raise 

177 

178 return custom_routes 

179 

180 

181async def _extract_routes_from_file(file_path: Path) -> list[RouteType]: 

182 module_path = _get_module_path_from_file_path(file_path) 

183 debug(f"Extracting routes from {file_path} -> {module_path}") 

184 try: 

185 with suppress(ModuleNotFoundError, ImportError): 

186 module = import_module(module_path) 

187 return _extract_routes_from_module(module, module_path) 

188 except Exception as e: 

189 debug(f"Error extracting routes from {file_path}: {e}") 

190 raise 

191 

192 return [] 

193 

194 

195def _get_module_path_from_file_path(file_path: Path) -> str: 

196 depth = -2 

197 if "adapters" in file_path.parts: 

198 depth = -4 

199 return ".".join(file_path.parts[depth:]).removesuffix(".py") 

200 

201 

202def _extract_routes_from_module(module: t.Any, module_path: str) -> list[RouteType]: 

203 if not hasattr(module, "routes"): 

204 debug(f"No routes attribute found in {module_path}") 

205 return [] 

206 module_routes = module.routes 

207 if not isinstance(module_routes, list): 

208 debug(f"Routes attribute in {module_path} is not a list: {type(module_routes)}") 

209 return [] 

210 

211 return _validate_route_objects(module_routes) 

212 

213 

214def _validate_route_objects(module_routes: list[t.Any]) -> list[RouteType]: 

215 valid_routes = [] 

216 for route in module_routes: 

217 if isinstance(route, Route | Router | Mount | Host | WebSocketRoute): 

218 valid_routes.append(route) 

219 else: 

220 debug(f"Skipping invalid route object: {type(route)}") 

221 

222 return valid_routes 

223 

224 

225async def gather_route_patterns( 

226 route_objects: list[RouteType], 

227) -> dict[str, t.Any]: 

228 patterns: dict[str, t.Any] = { 

229 "total_routes": len(route_objects), 

230 "route_types": {}, 

231 "path_patterns": [], 

232 "methods": set(), 

233 "endpoints": set(), 

234 } 

235 

236 for route in route_objects: 

237 route_type = type(route).__name__ 

238 patterns["route_types"][route_type] = ( 

239 patterns["route_types"].get(route_type, 0) + 1 

240 ) 

241 

242 path = getattr(route, "path", None) 

243 if path is not None: 

244 patterns["path_patterns"].append(path) 

245 

246 methods = getattr(route, "methods", None) 

247 if methods is not None: 

248 patterns["methods"].update(methods) 

249 

250 endpoint = getattr(route, "endpoint", None) 

251 if endpoint is not None: 

252 endpoint_name = getattr(endpoint, "__name__", str(endpoint)) 

253 patterns["endpoints"].add(endpoint_name) 

254 

255 patterns["methods"] = list(patterns["methods"]) 

256 patterns["endpoints"] = list(patterns["endpoints"]) 

257 

258 return patterns 

259 

260 

261def create_default_routes() -> list[RouteType]: 

262 try: 

263 routes_module = __import__( 

264 "fastblocks.adapters.routes.default", 

265 fromlist=["Routes"], 

266 ) 

267 Routes = routes_module.Routes 

268 routes_instance = Routes() 

269 return [ 

270 Route("/favicon.ico", endpoint=routes_instance.favicon, methods=["GET"]), 

271 Route("/robots.txt", endpoint=routes_instance.robots, methods=["GET"]), 

272 ] 

273 except (ImportError, AttributeError) as e: 

274 debug(f"Error loading default routes: {e}") 

275 return [] 

276 

277 

278async def validate_routes(routes: list[RouteType]) -> dict[str, t.Any]: 

279 validation: dict[str, t.Any] = { 

280 "valid_routes": [], 

281 "invalid_routes": [], 

282 "warnings": [], 

283 "total_checked": len(routes), 

284 } 

285 path_patterns: set[str] = set() 

286 for route in routes: 

287 _validate_single_route(route, validation, path_patterns) 

288 

289 return validation 

290 

291 

292def _validate_single_route( 

293 route: RouteType, 

294 validation: dict[str, t.Any], 

295 path_patterns: set[str], 

296) -> None: 

297 try: 

298 path = getattr(route, "path", None) 

299 if path is None: 

300 validation["invalid_routes"].append( 

301 {"route": str(route), "error": "Missing path attribute"}, 

302 ) 

303 return 

304 

305 _check_route_path_duplicates(route, validation, path_patterns) 

306 _check_route_endpoint(route, validation) 

307 validation["valid_routes"].append(route) 

308 

309 except Exception as e: 

310 validation["invalid_routes"].append({"route": str(route), "error": str(e)}) 

311 

312 

313def _check_route_path_duplicates( 

314 route: RouteType, 

315 validation: dict[str, t.Any], 

316 path_patterns: set[str], 

317) -> None: 

318 path = getattr(route, "path", None) 

319 if path is not None: 

320 if path in path_patterns: 

321 validation["warnings"].append(f"Duplicate path: {path}") 

322 path_patterns.add(path) 

323 

324 

325def _check_route_endpoint(route: RouteType, validation: dict[str, t.Any]) -> None: 

326 endpoint = getattr(route, "endpoint", None) 

327 path = getattr(route, "path", "unknown") 

328 if hasattr(route, "endpoint") and endpoint is None: 

329 validation["warnings"].append(f"Route {path} has no endpoint")