Coverage for src / lexigram / admin / dashboard / route_integrator.py: 14%
268 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1from __future__ import annotations
3from collections.abc import Sequence
4import inspect
5import types
6from typing import TYPE_CHECKING, Any, get_args, get_origin, get_type_hints
8from starlette.requests import Request as StarletteRequest
9from starlette.responses import HTMLResponse
11from lexigram.admin.navigation.clusters import (
12 CLUSTER_GROUP,
13 CLUSTER_ICON,
14 CLUSTER_LABEL,
15 CLUSTER_URL,
16 cluster_child_href,
17)
18from lexigram.admin.state.context import wants_fragment
19from lexigram.contracts.admin.types import (
20 ManagementPageDefinition,
21 SettingsPanelDefinition,
22)
23from lexigram.contracts.exceptions import UnresolvableDependencyError
24from lexigram.logging import get_logger
26if TYPE_CHECKING:
27 from lexigram.admin.core.routing import AdminRouter
28 from lexigram.admin.dashboard.naming_policy import NamingPolicy
29 from lexigram.contracts.admin.contributor import BaseAdminContributor
30 from lexigram.contracts.core.di import ContainerResolverProtocol
32logger = get_logger(__name__)
35def _strip_optional(tp: Any) -> Any:
36 """If *tp* is ``Optional[X]`` (``Union[X, None]`` or ``X | None``),
37 return ``X``. Otherwise return *tp* unchanged."""
38 origin = get_origin(tp)
39 if origin is types.UnionType:
40 args = get_args(tp)
41 non_none = [a for a in args if a is not type(None)]
42 if len(non_none) == 1:
43 return non_none[0]
44 return tp
47_DEFAULT_PRIMARY_COLOR = "#6b7280"
50async def _resolve_primary_color(container: Any) -> str:
51 """Resolve the saved branding primary color, best-effort.
53 Falls back to the framework default when no registry/db store is
54 available.
55 """
56 try:
57 from lexigram.admin.settings.panel.registry import ConfigRegistry
59 registry = await container.resolve(
60 ConfigRegistry,
61 bypass_visibility=True,
62 )
63 values = await registry.get_values("admin.branding", "db")
64 color = values.get("primary_color")
65 if color:
66 return str(color)
67 except Exception: # noqa: BLE001 — non-fatal
68 logger.exception("admin.theme_overrides_failed")
69 return _DEFAULT_PRIMARY_COLOR
72class AdminPageHandler:
73 """ASGI adapter that resolves a management page handler from the DI
74 container at request time and delegates to its ``handle()`` method.
76 Starlette treats class endpoints as ASGI apps — it calls
77 ``cls(scope, receive, send)`` which becomes ``__init__(scope, receive,
78 send)``. Management page handlers use keyword-only constructor DI
79 (``def __init__(self, *, repo: ..., ...)``), so direct registration
80 always raises TypeError. This wrapper sidesteps that by storing the
81 page **class** at route-build time and resolving an instance from the
82 container at request time.
83 """
85 def __init__(
86 self,
87 page_cls: type,
88 container: ContainerResolverProtocol,
89 ) -> None:
90 self._page_cls = page_cls
91 self._container = container
93 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
94 request = StarletteRequest(scope, receive, send)
95 try:
96 instance = await self._resolve_page()
97 response = await instance.handle(request)
98 except Exception:
99 logger.exception(
100 "admin_page_handler_error",
101 page=self._page_cls.__name__,
102 )
103 response = await _placeholder_page(request, self._container)
105 try:
106 is_htmx = wants_fragment(request)
107 except KeyError:
108 is_htmx = False
109 if not is_htmx and isinstance(response, HTMLResponse):
110 response = await self._wrap_in_shell(request, response)
112 await response(scope, receive, send)
114 async def _wrap_in_shell(
115 self,
116 request: StarletteRequest,
117 response: HTMLResponse,
118 ) -> HTMLResponse:
119 from pathlib import Path
121 from starlette.templating import Jinja2Templates
123 from lexigram.admin.engine.renderer import resolve_admin_nav
124 from lexigram.admin.ui.templates.shell import AdminShell
125 from lexigram.ui import raw, render_to_string
127 content = (
128 response.body.decode()
129 if isinstance(response.body, bytes)
130 else str(response.body)
131 )
133 title = self._page_cls.__name__.removesuffix("Page")
135 user = (
136 getattr(request.state, "user", None) if hasattr(request, "state") else None
137 )
138 nav_items, system_menu_items, secondary_nav = resolve_admin_nav(request)
139 if secondary_nav:
140 from lexigram.admin.ui.organisms.secondary_nav import ClusterLayout
142 content = render_to_string(
143 ClusterLayout(items=secondary_nav, content=raw(content))
144 )
146 theme_css = ""
147 try:
148 from lexigram.admin.theme.service import AdminThemeService
150 service = AdminThemeService(
151 primary_color=await _resolve_primary_color(self._container)
152 )
153 theme_css = service.generate_theme_css()
154 except Exception: # noqa: BLE001 — non-fatal
155 pass
157 user_menu_items: list[dict[str, str]] = [
158 {
159 "label": CLUSTER_LABEL,
160 "href": CLUSTER_URL,
161 "icon": CLUSTER_ICON,
162 },
163 {
164 "label": "Settings",
165 "href": "/admin/settings",
166 "icon": "settings",
167 },
168 ]
170 branding: dict[str, str] = {}
171 try:
172 from lexigram.admin.multitenancy.adapter import resolve_tenant_id
173 from lexigram.admin.services.settings_service import (
174 resolve_admin_settings_service,
175 )
177 container = (
178 getattr(request.state, "root_container", None)
179 or getattr(request.state, "container", None)
180 or getattr(request.app.state, "container", None)
181 or self._container
182 )
183 settings_service = await resolve_admin_settings_service(container)
184 if settings_service is not None:
185 tenant = await resolve_tenant_id(request, default="default")
186 overrides = await settings_service.get_all(tenant)
187 for field in ("primary_color", "site_name", "logo_url", "dark_mode"):
188 value = overrides.get(field) or overrides.get(
189 f"admin.branding.{field}"
190 )
191 if value:
192 branding[field] = value
193 if branding.get("primary_color"):
194 from lexigram.admin.theme.service import AdminThemeService
196 theme_css = AdminThemeService(
197 primary_color=branding["primary_color"]
198 ).generate_theme_css()
199 except Exception: # noqa: BLE001 — non-fatal
200 pass
202 shell = AdminShell(
203 content=content,
204 title=title,
205 user=user,
206 nav_items=nav_items,
207 system_menu_items=system_menu_items,
208 user_menu_items=user_menu_items,
209 theme_css=theme_css,
210 **{
211 k: v
212 for k, v in branding.items()
213 if k in ("dark_mode", "site_name", "logo_url")
214 },
215 )
216 shell_html = render_to_string(shell)
218 templates_dir = Path(__file__).resolve().parent.parent / "views" / "templates"
219 templates = Jinja2Templates(directory=str(templates_dir))
220 return templates.TemplateResponse(
221 request,
222 "admin_shell.html",
223 context={
224 "content": shell_html,
225 "title": title,
226 "dark_mode": branding.get("dark_mode", ""),
227 },
228 )
230 async def _resolve_page(self) -> Any:
231 """Resolve page instance from container.
233 Uses ``container.call(cls.__init__)`` to resolve each constructor
234 parameter from the DI container, then constructs the instance
235 manually. This is necessary because ``get_type_hints(cls)``
236 returns an empty dict for classes with ``from __future__ import
237 annotations`` (PEP 563), so ``container.call(cls)`` cannot
238 discover parameter types.
239 """
240 init_method = self._page_cls.__init__ # type: ignore[misc]
241 sig = inspect.signature(init_method)
242 hints = get_type_hints(init_method)
243 kwargs: dict[str, Any] = {}
244 for name, param in sig.parameters.items():
245 if name == "self":
246 continue
247 if param.kind in (
248 inspect.Parameter.VAR_POSITIONAL,
249 inspect.Parameter.VAR_KEYWORD,
250 ):
251 continue
252 param_type = hints.get(name)
253 if param_type is not None:
254 try:
255 resolution_target = _strip_optional(param_type)
256 kwargs[name] = await self._container.resolve(resolution_target)
257 continue
258 except UnresolvableDependencyError:
259 pass
260 if param.default is not inspect.Parameter.empty:
261 kwargs[name] = param.default
262 elif param_type is not None:
263 raise UnresolvableDependencyError(
264 f"Cannot resolve parameter {name!r} for "
265 f"{self._page_cls.__name__}: type {param_type} not registered.",
266 )
267 else:
268 raise UnresolvableDependencyError(
269 f"Cannot resolve parameter {name!r} for "
270 f"{self._page_cls.__name__}: no type hint and no default.",
271 )
272 return self._page_cls(**kwargs)
275async def _placeholder_page(
276 request: Any,
277 container: Any | None = None,
278) -> HTMLResponse:
279 """Placeholder for admin pages without an implemented handler.
281 For HTMX requests returns only the content fragment (no shell) so
282 the sidebar/topbar from the existing page stays intact. For direct
283 navigation returns the full admin layout.
285 Args:
286 request: Starlette request.
287 container: Optional resolver for theme settings.
288 """
289 content = (
290 '<div class="flex items-center justify-center h-64">'
291 '<div class="text-center">'
292 '<h2 class="text-xl font-semibold text-muted-foreground">Under Construction</h2>'
293 '<p class="text-muted-foreground mt-2">This page has not been implemented yet.</p>'
294 "</div></div>"
295 )
297 try:
298 from lexigram.admin.engine.renderer import resolve_admin_nav
300 nav_items, system_menu_items, secondary_nav = resolve_admin_nav(request)
301 except Exception: # noqa: BLE001 — non-fatal
302 nav_items, system_menu_items, secondary_nav = [], [], None
304 if secondary_nav:
305 from lexigram.admin.ui.organisms.secondary_nav import ClusterLayout
306 from lexigram.ui import raw, render_to_string
308 content = render_to_string(
309 ClusterLayout(items=secondary_nav, content=raw(content))
310 )
312 is_htmx = wants_fragment(request)
314 if is_htmx:
315 return HTMLResponse(content)
317 try:
318 from pathlib import Path
320 from starlette.templating import Jinja2Templates
322 from lexigram.admin.ui.templates.shell import AdminShell
323 from lexigram.ui import render_to_string
325 user = (
326 getattr(request.state, "user", None) if hasattr(request, "state") else None
327 )
329 user_menu_items = [
330 {
331 "label": CLUSTER_LABEL,
332 "href": CLUSTER_URL,
333 "icon": CLUSTER_ICON,
334 },
335 {
336 "label": "Settings",
337 "href": "/admin/settings",
338 "icon": "settings",
339 },
340 ]
342 theme_css = ""
343 try:
344 from lexigram.admin.theme.service import AdminThemeService
346 service = AdminThemeService(
347 primary_color=(
348 await _resolve_primary_color(container)
349 if container is not None
350 else _DEFAULT_PRIMARY_COLOR
351 )
352 )
353 theme_css = service.generate_theme_css()
354 except Exception: # noqa: BLE001 — non-fatal
355 pass
357 shell = AdminShell(
358 content=content,
359 title="Under Construction",
360 user=user,
361 nav_items=nav_items,
362 system_menu_items=system_menu_items,
363 user_menu_items=user_menu_items,
364 theme_css=theme_css,
365 )
366 shell_html = render_to_string(shell)
368 templates_dir = Path(__file__).resolve().parent.parent / "views" / "templates"
369 templates = Jinja2Templates(directory=str(templates_dir))
370 return templates.TemplateResponse(
371 request,
372 "admin_shell.html",
373 context={
374 "content": shell_html,
375 "title": "Under Construction",
376 "dark_mode": "",
377 },
378 )
379 except Exception:
380 return HTMLResponse(content)
383def _resolve_handler(handler: Any) -> Any:
384 """Resolve a string dotted-path handler to the actual callable."""
385 if not isinstance(handler, str):
386 return handler
387 try:
388 module_path, _, func_name = handler.partition(":")
389 mod = __import__(module_path, fromlist=[func_name])
390 resolved = getattr(mod, func_name, None)
391 if resolved is None:
392 logger.warning("handler_import_failed", handler=handler)
393 return resolved
394 except Exception: # noqa: BLE001
395 logger.warning("handler_import_failed", handler=handler, exc_info=True)
396 return None
399def _register_pages(
400 router: AdminRouter,
401 naming_policy: NamingPolicy,
402 prefix: str,
403 pages: list[ManagementPageDefinition],
404 container: Any = None,
405) -> None:
406 """Register management page routes on the admin router.
408 Routes are registered relative to the admin app mount point
409 (the prefix is stripped before registration — the admin router
410 already lives under the admin prefix due to ``AdminRouter.mount()``).
411 ``registered_internal_paths`` is updated externally so that
412 ``_ensure_nav_route`` does not create duplicate placeholders.
413 """
414 # prefix is intentionally unused — routes live inside the mounted
415 # admin app and must be relative to its mount point.
416 for page in pages:
417 handler = _resolve_handler(page.handler)
418 if handler is None:
419 continue
420 if inspect.isclass(handler) and container is not None:
421 handler = AdminPageHandler(handler, container)
422 path = page.route_path
423 if not path.startswith("/"):
424 path = f"/{path}"
425 ns_name = naming_policy.namespaced(page.contributor, page.name)
426 naming_policy.register("page", ns_name)
427 router.add_route(path=path, method="GET", handler=handler, name=ns_name)
430def _register_settings(
431 router: AdminRouter,
432 naming_policy: NamingPolicy,
433 prefix: str,
434 panels: list[SettingsPanelDefinition],
435 container: Any = None,
436) -> None:
437 """Register settings panel routes on the admin router.
439 Routes are registered relative to the admin app mount point
440 (the prefix is stripped before registration — see ``_register_pages``).
441 """
442 for panel in panels:
443 handler = _resolve_handler(panel.handler)
444 if handler is None:
445 continue
446 if inspect.isclass(handler) and container is not None:
447 handler = AdminPageHandler(handler, container)
448 path = panel.route_path
449 if not path.startswith("/"):
450 path = f"/{path}"
451 ns_name = naming_policy.namespaced(panel.contributor, panel.name)
452 naming_policy.register("panel", ns_name)
453 router.add_route(path=path, method="GET", handler=handler, name=ns_name)
456class RouteIntegrator:
457 """Collects ``AdminRouteSpec``, ``ManagementPageDefinition``, and
458 ``SettingsPanelDefinition`` from contributors and registers them
459 on the router."""
461 def __init__(
462 self,
463 *,
464 router: AdminRouter,
465 naming_policy: NamingPolicy,
466 route_prefix: str = "",
467 container: Any = None,
468 ) -> None:
469 self._router = router
470 self._naming = naming_policy
471 self._prefix = route_prefix
472 self._container = container
474 def register(self, contributors: Sequence[BaseAdminContributor]) -> None:
475 """Register each contributor's routes, management pages, and
476 settings panels on the admin router. Nav-item URLs that don't
477 have a corresponding handler automatically get a placeholder
478 route so they never 404."""
479 registered_internal_paths: set[str] = set()
481 for c in contributors:
482 # Routes
483 for spec in c.get_routes():
484 ns_name = self._naming.namespaced(c.package_source, spec.name)
485 self._naming.register("route", ns_name)
486 registered_internal_paths.add(spec.path)
487 self._router.add_route(
488 path=spec.path,
489 method=spec.method,
490 handler=spec.handler,
491 name=ns_name,
492 )
494 # Management pages
495 pages = c.get_management_pages()
496 if pages:
497 for page in pages:
498 internal = page.route_path
499 if not internal.startswith("/"):
500 internal = f"/{internal}"
501 registered_internal_paths.add(internal)
502 _register_pages(
503 self._router,
504 self._naming,
505 self._prefix,
506 pages, # type: ignore[arg-type]
507 container=self._container,
508 )
510 # Settings panels
511 panels = c.get_settings_panels()
512 if panels:
513 for panel in panels:
514 internal = panel.route_path
515 if not internal.startswith("/"):
516 internal = f"/{internal}"
517 registered_internal_paths.add(internal)
518 _register_settings(
519 self._router,
520 self._naming,
521 self._prefix,
522 panels, # type: ignore[arg-type]
523 container=self._container,
524 )
526 # Auto-register placeholder routes for nav items without handlers.
527 for c in contributors:
528 for item in c.get_navigation_items():
529 self._ensure_nav_route(item, registered_internal_paths)
530 for child in item.children or ():
531 self._ensure_nav_route(child, registered_internal_paths)
533 # Cluster areas are also reachable under the center namespace
534 # (e.g. /admin/infrastructure/web), mirroring how settings
535 # sub-pages nest below /admin/settings. Aliases share the source
536 # route's handler — real page or placeholder alike.
537 for c in contributors:
538 for item in c.get_navigation_items():
539 self._register_cluster_alias(item)
540 for child in item.children or ():
541 self._register_cluster_alias(child)
543 def _ensure_nav_route(
544 self,
545 item: Any,
546 registered_paths: set[str],
547 ) -> None:
548 """Register a placeholder route for *item* if its URL isn't covered."""
549 url = item.url
550 if not url or url.startswith("http"):
551 return
553 internal = url
554 if self._prefix and internal.startswith(self._prefix):
555 internal = internal[len(self._prefix) :]
556 if not internal:
557 internal = "/"
559 if internal in registered_paths or url in registered_paths:
560 return
562 safe_label = item.label.lower().replace(" ", "_").replace("/", "_")
563 self._router.add_route(
564 path=internal,
565 method="GET",
566 handler=_placeholder_page,
567 name=f"placeholder_{safe_label}",
568 )
569 registered_paths.add(internal)
571 def _register_cluster_alias(self, item: Any) -> None:
572 """Register a namespaced alias for a cluster group nav item.
574 Cluster areas live under the center namespace (``/admin/
575 infrastructure/web``) in addition to their contributor URL. When
576 the source URL has a real handler, the alias reuses it; otherwise
577 the alias falls back to the placeholder page.
578 """
579 if getattr(item, "group", None) != CLUSTER_GROUP:
580 return
581 url = item.url
582 if not url or url.startswith("http"):
583 return
584 namespaced = cluster_child_href(url)
585 if not namespaced or namespaced == url:
586 return
588 internal = url
589 internal_ns = namespaced
590 if self._prefix and internal.startswith(self._prefix):
591 internal = internal[len(self._prefix) :]
592 if self._prefix and internal_ns.startswith(self._prefix):
593 internal_ns = internal_ns[len(self._prefix) :]
594 if internal_ns == internal:
595 return
597 safe_label = item.label.lower().replace(" ", "_").replace("/", "_")
598 if not self._router.alias_route(
599 internal,
600 internal_ns,
601 name=f"cluster_alias_{safe_label}",
602 ):
603 self._router.add_route(
604 path=internal_ns,
605 method="GET",
606 handler=_placeholder_page,
607 name=f"cluster_alias_{safe_label}",
608 )