Coverage for src/lexigram/admin/core/routing.py: 82%

1from __future__ import annotations 

2 

3from pathlib import Path 

4from typing import Any 

5 

6from starlette.applications import Starlette 

7from starlette.middleware.sessions import SessionMiddleware 

8from starlette.routing import Mount, Route 

9from starlette.staticfiles import StaticFiles 

10 

11from lexigram.admin.auth.services._cookie_config import ( 

12 build_session_cookie_kwargs, 

13) 

14from lexigram.admin.config import AdminConfig 

15from lexigram.admin.controllers.command_palette import CommandPaletteController 

16from lexigram.admin.controllers.search import SearchController 

17from lexigram.admin.openapi.controller import OpenAPIController 

18from lexigram.admin.relations.routes import register_relation_routes 

19from lexigram.admin.resources.handler import ResourceHandler 

20from lexigram.admin.services.search_service import SearchService 

21from lexigram.contracts.auth import AuthorizerProtocol 

22from lexigram.di.decorators import inject 

23from lexigram.logging import get_logger 

24 

25logger = get_logger(__name__) 

26 

27 

28class _ResourceManager: 

29 """Adapter that wraps a ``{name: resource_instance}`` dict for SearchService. 

30 

31 SearchService expects a ``resource_manager`` with a ``get_all_resources()`` 

32 method that returns resource instances. This adapter provides that 

33 interface from the AdminRouter's internal resources dict. 

34 """ 

35 

36 def __init__(self, resources: dict[str, Any]) -> None: 

37 self._resources = resources 

38 

39 def get_all_resources(self) -> list[Any]: 

40 return list(self._resources.values()) 

41 

42 

43@inject 

44class AdminRouter: 

45 """Router and mounting logic for the admin panel.""" 

46 

47 def __init__( 

48 self, 

49 config: AdminConfig, 

50 resources: dict[str, Any] | None = None, 

51 controllers: list[Any] | None = None, 

52 middleware_stack: list[tuple[type, dict]] | None = None, 

53 authorizer: AuthorizerProtocol | None = None, 

54 ): 

55 self._config = config 

56 self._resources = resources or {} 

57 self._controllers = controllers or [] 

58 self._middleware_stack = middleware_stack or [] 

59 self._authorizer = authorizer 

60 self._extra_routes: list[Route] = [] 

61 self._is_mounted = False 

62 

63 def add_route( 

64 self, 

65 path: str, 

66 method: str, 

67 handler: Any, 

68 name: str, 

69 ) -> None: 

70 """Register a single route for later mounting. 

71 

72 Routes added here are included the next time ``mount()`` is called. 

73 """ 

74 self._extra_routes.append( 

75 Route( 

76 path, 

77 endpoint=handler, 

78 methods=[method], 

79 name=name, 

80 ), 

81 ) 

82 

83 def alias_route(self, source_path: str, alias_path: str, name: str) -> bool: 

84 """Register ``alias_path`` with the same endpoint as ``source_path``. 

85 

86 Used to expose a route under an additional path (e.g. cluster 

87 areas under the center namespace). Returns ``True`` when the 

88 source route was found and aliased. 

89 

90 Args: 

91 source_path: Path of the already-registered route. 

92 alias_path: Additional path to register. 

93 name: Route name for the alias. 

94 

95 Returns: 

96 Whether the source route existed and the alias was added. 

97 """ 

98 source = next( 

99 (r for r in self._extra_routes if r.path == source_path), 

100 None, 

101 ) 

102 if source is None: 

103 return False 

104 self._extra_routes.append( 

105 Route( 

106 alias_path, 

107 endpoint=source.endpoint, 

108 methods=source.methods, 

109 name=name, 

110 ), 

111 ) 

112 return True 

113 

114 def mount(self, app: Starlette) -> Starlette | None: 

115 """Mount admin panel to a Starlette application. 

116 

117 Returns: 

118 The created admin sub-app (so callers can set state on it), or None 

119 if mounting failed. 

120 """ 

121 if self._is_mounted: 

122 logger.warning("AdminRouter already mounted, skipping") 

123 return None 

124 

125 routes = self._build_routes() 

126 admin_app = Starlette(routes=routes) 

127 

128 # Sign the admin session cookie from the validated auth config. The 

129 # helper also derives https_only / same_site / max_age from env. 

130 cookie_kwargs = build_session_cookie_kwargs(self._config.auth) 

131 # Add our middleware in reverse order so that the stack list order 

132 # (e.g. [Setup, Csrf, AuthGuard]) becomes the execution order. 

133 # Starlette's add_middleware inserts at position 0 (outermost), so 

134 # to get Session → Setup → Csrf → AuthGuard → Routes we must: 

135 # 1. add AuthGuard first (innermost) 

136 # 2. add Csrf 

137 # 3. add Setup 

138 # 4. add Session last (outermost — runs first, populates scope["session"]) 

139 for middleware_class, options in reversed(self._middleware_stack): 

140 admin_app.add_middleware(middleware_class, **options) # type: ignore[arg-type] 

141 

142 admin_app.add_middleware(SessionMiddleware, **cookie_kwargs) 

143 

144 admin_mount = Mount( 

145 self._config.prefix, 

146 app=admin_app, 

147 name="admin", 

148 ) 

149 

150 if hasattr(app, "routes"): 

151 app.routes.append(admin_mount) 

152 elif hasattr(app, "include_router"): 

153 app.include_router(admin_mount) 

154 elif hasattr(app, "_invoker"): 

155 invoker = app._invoker 

156 if hasattr(invoker, "routes"): 

157 invoker.routes.append(admin_mount) 

158 else: 

159 logger.warning("Could not mount admin - Lexigram invoker has no routes") 

160 return None 

161 else: 

162 logger.warning("Could not mount admin - unknown app type") 

163 return None 

164 

165 self._is_mounted = True 

166 logger.info("admin.mounted", prefix=self._config.prefix) 

167 return admin_app 

168 

169 def _build_routes(self) -> list[Route | Mount]: 

170 """Build all admin routes.""" 

171 routes: list[Route | Mount] = [] 

172 

173 static_dir = self._resolve_static_dir() 

174 if static_dir and static_dir.exists(): 

175 routes.append( 

176 Mount( 

177 "/static", 

178 app=StaticFiles(directory=str(static_dir)), 

179 name="admin_static", 

180 ), 

181 ) 

182 

183 for controller in self._controllers: 

184 if not isinstance(controller, type) and hasattr(controller, "get_routes"): 

185 routes.extend(controller.get_routes()) 

186 

187 for name, resource in self._resources.items(): 

188 routes.extend(self._build_resource_routes(name, resource)) 

189 

190 # Global search endpoint 

191 search_service = SearchService( 

192 resource_manager=_ResourceManager(self._resources), 

193 authorizer=self._authorizer, 

194 ) 

195 search_controller = SearchController(search_service=search_service) 

196 routes.append( 

197 Route( 

198 "/search", 

199 endpoint=search_controller.search, 

200 methods=["GET"], 

201 name="admin_search", 

202 ), 

203 ) 

204 

205 # Command palette endpoint 

206 palette_controller = CommandPaletteController(search_service=search_service) 

207 routes.append( 

208 Route( 

209 "/command-palette", 

210 endpoint=palette_controller.search, 

211 methods=["GET"], 

212 name="admin_command_palette", 

213 ), 

214 ) 

215 

216 # OpenAPI spec endpoint 

217 openapi_controller = OpenAPIController(resources=self._resources) 

218 routes.append( 

219 Route( 

220 "/openapi.json", 

221 endpoint=openapi_controller.get_spec, 

222 methods=["GET"], 

223 name="admin_openapi", 

224 ), 

225 ) 

226 

227 # Extra routes registered via add_route (e.g. from RouteIntegrator) 

228 routes.extend(self._extra_routes) 

229 

230 return routes 

231 

232 def _build_resource_routes( 

233 self, 

234 name: str, 

235 resource: Any | None = None, 

236 ) -> list[Route]: 

237 """Build routes for a resource.""" 

238 prefix = f"/{name}" 

239 # Pass the resource in a single-entry dict so ResourceHandler can look it up 

240 resources_dict = {name: resource} if resource is not None else {} 

241 routes: list[Route] = [ 

242 Route( 

243 prefix, 

244 ResourceHandler(self._config, name, "list", resources=resources_dict), 

245 name=f"admin_{name}_list", 

246 ), 

247 Route( 

248 f"{prefix}/create", 

249 ResourceHandler(self._config, name, "create", resources=resources_dict), 

250 name=f"admin_{name}_create", 

251 methods=["GET", "POST"], 

252 ), 

253 Route( 

254 f"{prefix}/create/form", 

255 ResourceHandler(self._config, name, "create", resources=resources_dict), 

256 name=f"admin_{name}_create_form", 

257 ), 

258 # Fixed-path routes must come before {prefix}/{id} to avoid 

259 # the catch-all parameterised route stealing them. 

260 Route( 

261 f"{prefix}/bulk", 

262 ResourceHandler(self._config, name, "bulk", resources=resources_dict), 

263 name=f"admin_{name}_bulk", 

264 methods=["POST"], 

265 ), 

266 Route( 

267 f"{prefix}/bulk-delete-confirm", 

268 ResourceHandler( 

269 self._config, name, "bulk-delete-confirm", resources=resources_dict 

270 ), 

271 name=f"admin_{name}_bulk_delete_confirm", 

272 methods=["GET"], 

273 ), 

274 Route( 

275 f"{prefix}/bulk-purge-confirm", 

276 ResourceHandler( 

277 self._config, name, "bulk-purge-confirm", resources=resources_dict 

278 ), 

279 name=f"admin_{name}_bulk_purge_confirm", 

280 methods=["GET"], 

281 ), 

282 Route( 

283 f"{prefix}/bulk-restore-confirm", 

284 ResourceHandler( 

285 self._config, name, "bulk-restore-confirm", resources=resources_dict 

286 ), 

287 name=f"admin_{name}_bulk_restore_confirm", 

288 methods=["GET"], 

289 ), 

290 Route( 

291 f"{prefix}/import-example", 

292 ResourceHandler( 

293 self._config, name, "import-example", resources=resources_dict 

294 ), 

295 name=f"admin_{name}_import_example", 

296 methods=["GET"], 

297 ), 

298 Route( 

299 f"{prefix}/import-report", 

300 ResourceHandler( 

301 self._config, name, "import-report", resources=resources_dict 

302 ), 

303 name=f"admin_{name}_import_report", 

304 methods=["GET"], 

305 ), 

306 Route( 

307 f"{prefix}/{{id}}", 

308 ResourceHandler(self._config, name, "detail", resources=resources_dict), 

309 name=f"admin_{name}_detail", 

310 ), 

311 Route( 

312 f"{prefix}/{{id}}/edit", 

313 ResourceHandler(self._config, name, "edit", resources=resources_dict), 

314 name=f"admin_{name}_edit", 

315 methods=["GET", "POST"], 

316 ), 

317 Route( 

318 f"{prefix}/{{id}}/clone", 

319 ResourceHandler(self._config, name, "clone", resources=resources_dict), 

320 name=f"admin_{name}_clone", 

321 methods=["GET"], 

322 ), 

323 Route( 

324 f"{prefix}/{{id}}/restore", 

325 ResourceHandler( 

326 self._config, name, "restore", resources=resources_dict 

327 ), 

328 name=f"admin_{name}_restore", 

329 methods=["GET"], 

330 ), 

331 Route( 

332 f"{prefix}/{{id}}/purge", 

333 ResourceHandler(self._config, name, "purge", resources=resources_dict), 

334 name=f"admin_{name}_purge", 

335 methods=["GET"], 

336 ), 

337 Route( 

338 f"{prefix}/{{id}}/delete-confirm", 

339 ResourceHandler( 

340 self._config, name, "delete-confirm", resources=resources_dict 

341 ), 

342 name=f"admin_{name}_delete_confirm", 

343 methods=["GET"], 

344 ), 

345 Route( 

346 f"{prefix}/{{id}}/delete", 

347 ResourceHandler(self._config, name, "delete", resources=resources_dict), 

348 name=f"admin_{name}_delete", 

349 methods=["DELETE", "POST"], 

350 ), 

351 Route( 

352 f"{prefix}/{{id}}/permissions", 

353 ResourceHandler( 

354 self._config, name, "permissions", resources=resources_dict 

355 ), 

356 name=f"admin_{name}_permissions", 

357 methods=["GET", "POST"], 

358 ), 

359 ] 

360 

361 # Register relation routes for each RelationManager on this resource 

362 if ( 

363 resource is not None 

364 and hasattr(resource, "relations") 

365 and resource.relations 

366 ): 

367 for rel_cls in resource.relations: 

368 rel_routes = register_relation_routes( 

369 name, 

370 rel_cls, 

371 parent_data_source=getattr(resource, "_data_source", None), 

372 ) 

373 routes.extend(rel_routes) 

374 

375 return routes 

376 

377 def _resolve_static_dir(self) -> Path | None: 

378 """Resolve static files directory.""" 

379 if self._config.static_dir: 

380 return Path(self._config.static_dir) 

381 

382 package_dir = Path(__file__).parent.parent 

383 default_static = package_dir / "static" 

384 if default_static.exists(): 

385 return default_static 

386 

387 return None