Coverage for src / lexigram / admin / core / routing.py: 29%
101 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1from __future__ import annotations
3from pathlib import Path
4from typing import Any
6from starlette.applications import Starlette
7from starlette.middleware.sessions import SessionMiddleware
8from starlette.routing import Mount, Route
9from starlette.staticfiles import StaticFiles
11from lexigram.admin.config import AdminConfig
12from lexigram.admin.controllers.command_palette import CommandPaletteController
13from lexigram.admin.controllers.search import SearchController
14from lexigram.admin.openapi.controller import OpenAPIController
15from lexigram.admin.relations.routes import register_relation_routes
16from lexigram.admin.resources.handler import ResourceHandler
17from lexigram.admin.services.search_service import SearchService
18from lexigram.di.decorators import inject
19from lexigram.logging import get_logger
21logger = get_logger(__name__)
24class _ResourceManager:
25 """Adapter that wraps a ``{name: resource_instance}`` dict for SearchService.
27 SearchService expects a ``resource_manager`` with a ``get_all_resources()``
28 method that returns resource instances. This adapter provides that
29 interface from the AdminRouter's internal resources dict.
30 """
32 def __init__(self, resources: dict[str, Any]) -> None:
33 self._resources = resources
35 def get_all_resources(self) -> list[Any]:
36 return list(self._resources.values())
39@inject
40class AdminRouter:
41 """Router and mounting logic for the admin panel."""
43 def __init__(
44 self,
45 config: AdminConfig,
46 resources: dict[str, Any] | None = None,
47 controllers: list[Any] | None = None,
48 middleware_stack: list[tuple[type, dict]] | None = None,
49 ):
50 self._config = config
51 self._resources = resources or {}
52 self._controllers = controllers or []
53 self._middleware_stack = middleware_stack or []
54 self._extra_routes: list[Route] = []
55 self._is_mounted = False
57 def add_route(
58 self,
59 path: str,
60 method: str,
61 handler: Any,
62 name: str,
63 ) -> None:
64 """Register a single route for later mounting.
66 Routes added here are included the next time ``mount()`` is called.
67 """
68 self._extra_routes.append(
69 Route(
70 path,
71 endpoint=handler,
72 methods=[method],
73 name=name,
74 ),
75 )
77 def alias_route(self, source_path: str, alias_path: str, name: str) -> bool:
78 """Register ``alias_path`` with the same endpoint as ``source_path``.
80 Used to expose a route under an additional path (e.g. cluster
81 areas under the center namespace). Returns ``True`` when the
82 source route was found and aliased.
84 Args:
85 source_path: Path of the already-registered route.
86 alias_path: Additional path to register.
87 name: Route name for the alias.
89 Returns:
90 Whether the source route existed and the alias was added.
91 """
92 source = next(
93 (r for r in self._extra_routes if r.path == source_path),
94 None,
95 )
96 if source is None:
97 return False
98 self._extra_routes.append(
99 Route(
100 alias_path,
101 endpoint=source.endpoint,
102 methods=source.methods,
103 name=name,
104 ),
105 )
106 return True
108 def mount(self, app: Starlette) -> Starlette | None:
109 """Mount admin panel to a Starlette application.
111 Returns:
112 The created admin sub-app (so callers can set state on it), or None
113 if mounting failed.
114 """
115 if self._is_mounted:
116 logger.warning("AdminRouter already mounted, skipping")
117 return None
119 routes = self._build_routes()
120 admin_app = Starlette(routes=routes)
122 secret_key = (
123 getattr(self._config, "secret_key", None)
124 or "dev-secret-key-change-in-production"
125 )
126 # Add our middleware in reverse order so that the stack list order
127 # (e.g. [Setup, Csrf, AuthGuard]) becomes the execution order.
128 # Starlette's add_middleware inserts at position 0 (outermost), so
129 # to get Session → Setup → Csrf → AuthGuard → Routes we must:
130 # 1. add AuthGuard first (innermost)
131 # 2. add Csrf
132 # 3. add Setup
133 # 4. add Session last (outermost — runs first, populates scope["session"])
134 for middleware_class, options in reversed(self._middleware_stack):
135 admin_app.add_middleware(middleware_class, **options) # type: ignore[arg-type]
137 admin_app.add_middleware(SessionMiddleware, secret_key=secret_key)
139 admin_mount = Mount(
140 self._config.prefix,
141 app=admin_app,
142 name="admin",
143 )
145 if hasattr(app, "routes"):
146 app.routes.append(admin_mount)
147 elif hasattr(app, "include_router"):
148 app.include_router(admin_mount)
149 elif hasattr(app, "_invoker"):
150 invoker = app._invoker
151 if hasattr(invoker, "routes"):
152 invoker.routes.append(admin_mount)
153 else:
154 logger.warning("Could not mount admin - Lexigram invoker has no routes")
155 return None
156 else:
157 logger.warning("Could not mount admin - unknown app type")
158 return None
160 self._is_mounted = True
161 logger.info("admin.mounted", prefix=self._config.prefix)
162 return admin_app
164 def _build_routes(self) -> list[Route | Mount]:
165 """Build all admin routes."""
166 routes: list[Route | Mount] = []
168 static_dir = self._resolve_static_dir()
169 if static_dir and static_dir.exists():
170 routes.append(
171 Mount(
172 "/static",
173 app=StaticFiles(directory=str(static_dir)),
174 name="admin_static",
175 ),
176 )
178 for controller in self._controllers:
179 if not isinstance(controller, type) and hasattr(controller, "get_routes"):
180 routes.extend(controller.get_routes())
182 for name, resource in self._resources.items():
183 routes.extend(self._build_resource_routes(name, resource))
185 # Global search endpoint
186 search_service = SearchService(
187 resource_manager=_ResourceManager(self._resources),
188 )
189 search_controller = SearchController(search_service=search_service)
190 routes.append(
191 Route(
192 "/search",
193 endpoint=search_controller.search,
194 methods=["GET"],
195 name="admin_search",
196 ),
197 )
199 # Command palette endpoint
200 palette_controller = CommandPaletteController(search_service=search_service)
201 routes.append(
202 Route(
203 "/command-palette",
204 endpoint=palette_controller.search,
205 methods=["GET"],
206 name="admin_command_palette",
207 ),
208 )
210 # OpenAPI spec endpoint
211 openapi_controller = OpenAPIController(resources=self._resources)
212 routes.append(
213 Route(
214 "/openapi.json",
215 endpoint=openapi_controller.get_spec,
216 methods=["GET"],
217 name="admin_openapi",
218 ),
219 )
221 # Extra routes registered via add_route (e.g. from RouteIntegrator)
222 routes.extend(self._extra_routes)
224 return routes
226 def _build_resource_routes(
227 self,
228 name: str,
229 resource: Any | None = None,
230 ) -> list[Route]:
231 """Build routes for a resource."""
232 prefix = f"/{name}"
233 # Pass the resource in a single-entry dict so ResourceHandler can look it up
234 resources_dict = {name: resource} if resource is not None else {}
235 routes: list[Route] = [
236 Route(
237 prefix,
238 ResourceHandler(self._config, name, "list", resources=resources_dict),
239 name=f"admin_{name}_list",
240 ),
241 Route(
242 f"{prefix}/create",
243 ResourceHandler(self._config, name, "create", resources=resources_dict),
244 name=f"admin_{name}_create",
245 methods=["GET", "POST"],
246 ),
247 Route(
248 f"{prefix}/create/form",
249 ResourceHandler(self._config, name, "create", resources=resources_dict),
250 name=f"admin_{name}_create_form",
251 ),
252 # Fixed-path routes must come before {prefix}/{id} to avoid
253 # the catch-all parameterised route stealing them.
254 Route(
255 f"{prefix}/bulk",
256 ResourceHandler(self._config, name, "bulk", resources=resources_dict),
257 name=f"admin_{name}_bulk",
258 methods=["POST"],
259 ),
260 Route(
261 f"{prefix}/bulk-delete-confirm",
262 ResourceHandler(
263 self._config, name, "bulk-delete-confirm", resources=resources_dict
264 ),
265 name=f"admin_{name}_bulk_delete_confirm",
266 methods=["GET"],
267 ),
268 Route(
269 f"{prefix}/{{id}}",
270 ResourceHandler(self._config, name, "detail", resources=resources_dict),
271 name=f"admin_{name}_detail",
272 ),
273 Route(
274 f"{prefix}/{{id}}/edit",
275 ResourceHandler(self._config, name, "edit", resources=resources_dict),
276 name=f"admin_{name}_edit",
277 methods=["GET", "POST"],
278 ),
279 Route(
280 f"{prefix}/{{id}}/clone",
281 ResourceHandler(self._config, name, "clone", resources=resources_dict),
282 name=f"admin_{name}_clone",
283 methods=["GET"],
284 ),
285 Route(
286 f"{prefix}/{{id}}/restore",
287 ResourceHandler(
288 self._config, name, "restore", resources=resources_dict
289 ),
290 name=f"admin_{name}_restore",
291 methods=["GET"],
292 ),
293 Route(
294 f"{prefix}/{{id}}/purge",
295 ResourceHandler(self._config, name, "purge", resources=resources_dict),
296 name=f"admin_{name}_purge",
297 methods=["GET"],
298 ),
299 Route(
300 f"{prefix}/{{id}}/delete-confirm",
301 ResourceHandler(
302 self._config, name, "delete-confirm", resources=resources_dict
303 ),
304 name=f"admin_{name}_delete_confirm",
305 methods=["GET"],
306 ),
307 Route(
308 f"{prefix}/{{id}}/delete",
309 ResourceHandler(self._config, name, "delete", resources=resources_dict),
310 name=f"admin_{name}_delete",
311 methods=["DELETE", "POST"],
312 ),
313 ]
315 # Register relation routes for each RelationManager on this resource
316 if (
317 resource is not None
318 and hasattr(resource, "relations")
319 and resource.relations
320 ):
321 for rel_cls in resource.relations:
322 rel_routes = register_relation_routes(name, rel_cls)
323 routes.extend(rel_routes)
325 return routes
327 def _resolve_static_dir(self) -> Path | None:
328 """Resolve static files directory."""
329 if self._config.static_dir:
330 return Path(self._config.static_dir)
332 package_dir = Path(__file__).parent.parent
333 default_static = package_dir / "static"
334 if default_static.exists():
335 return default_static
337 return None