Coverage for src/lexigram/admin/dashboard/route_integrator.py: 80%
337 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1from __future__ import annotations
3from collections.abc import Sequence
4import inspect
5import re
6import types
7from typing import TYPE_CHECKING, Any, cast, get_args, get_origin, get_type_hints
9from markupsafe import Markup
10from starlette.requests import Request as StarletteRequest
11from starlette.responses import HTMLResponse
13from lexigram.admin.navigation.clusters import (
14 CLUSTER_GROUP,
15 CLUSTER_LABEL,
16 CLUSTER_URL,
17 cluster_child_href,
18 cluster_items,
19 is_cluster_path,
20)
21from lexigram.admin.state.context import wants_fragment
22from lexigram.contracts.admin.page_content import PageContent
23from lexigram.contracts.admin.types import (
24 ManagementPageDefinition,
25 SettingsPanelDefinition,
26)
27from lexigram.contracts.admin.widget_content import EmptyContent
28from lexigram.contracts.exceptions import UnresolvableDependencyError
29from lexigram.logging import get_logger
31if TYPE_CHECKING:
32 from lexigram.admin.core.routing import AdminRouter
33 from lexigram.admin.dashboard.naming_policy import NamingPolicy
34 from lexigram.contracts.admin.contributor import BaseAdminContributor
35 from lexigram.contracts.core.di import ContainerResolverProtocol
37logger = get_logger(__name__)
40def _strip_optional(tp: Any) -> Any:
41 """If *tp* is ``Optional[X]`` (``Union[X, None]`` or ``X | None``),
42 return ``X``. Otherwise return *tp* unchanged."""
43 origin = get_origin(tp)
44 if origin is types.UnionType:
45 args = get_args(tp)
46 non_none = [a for a in args if a is not type(None)]
47 if len(non_none) == 1:
48 return non_none[0]
49 return tp
52_DEFAULT_PRIMARY_COLOR = "#6b7280"
54_CLUSTER_HEADER_DESCRIPTION = (
55 "Monitor and manage the services powering your application: web, data, "
56 "and runtime areas."
57)
60def _cluster_header_html() -> str:
61 """Render the cluster center top-level title + description block.
63 Mirrors the settings center header: a ``mb-2`` wrapper with the
64 section title and a muted one-line description, rendered above the
65 whole center layout (secondary sidebar and page content).
66 """
67 from lexigram.ui import el, render_to_string
69 return render_to_string(
70 el(
71 "div",
72 el("h1", CLUSTER_LABEL, class_="text-2xl font-bold text-foreground"),
73 el(
74 "p",
75 _CLUSTER_HEADER_DESCRIPTION,
76 class_="text-muted-foreground mt-1",
77 ),
78 class_="mb-2",
79 )
80 )
83#: First header block rendered by management pages (h1 + description +
84#: divider). Cluster pages have their own top-level header injected by the
85#: shell wrapper, so this inline header is dropped.
86_PAGE_HEADER_RE = re.compile(
87 r"<h1[^>]*>.*?</h1>\s*<p[^>]*>.*?</p>\s*(?:<hr[^>]*/?>)?",
88 re.S,
89)
92async def _resolve_primary_color(container: Any) -> str:
93 """Resolve the saved branding primary color, best-effort.
95 Falls back to the framework default when no registry/db store is
96 available.
97 """
98 try:
99 from lexigram.admin.settings.panel.registry import ConfigRegistry
101 registry = await container.resolve(
102 ConfigRegistry,
103 bypass_visibility=True,
104 )
105 values = await registry.get_values("admin.branding", "db")
106 color = values.get("primary_color")
107 if color:
108 return str(color)
109 except Exception: # noqa: BLE001 — non-fatal
110 logger.exception("admin.theme_overrides_failed")
111 return _DEFAULT_PRIMARY_COLOR
114class AdminPageHandler:
115 """ASGI adapter that resolves a management page handler from the DI
116 container at request time and delegates to its ``handle()`` method.
118 Starlette treats class endpoints as ASGI apps — it calls
119 ``cls(scope, receive, send)`` which becomes ``__init__(scope, receive,
120 send)``. Management page handlers use keyword-only constructor DI
121 (``def __init__(self, *, repo: ..., ...)``), so direct registration
122 always raises TypeError. This wrapper sidesteps that by storing the
123 page **class** at route-build time and resolving an instance from the
124 container at request time.
125 """
127 def __init__(
128 self,
129 page_cls: type,
130 container: ContainerResolverProtocol,
131 ) -> None:
132 self._page_cls = page_cls
133 self._container = container
135 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
136 request = StarletteRequest(scope, receive, send)
137 try:
138 instance = await self._resolve_page()
139 response = await instance.handle(request)
140 if not isinstance(response, HTMLResponse):
141 from lexigram.admin.dashboard.page_renderer import (
142 render_page_content,
143 )
144 from lexigram.contracts.admin.page_content import PageContent
146 if isinstance(response, PageContent):
147 response = render_page_content(response)
148 else:
149 logger.error(
150 "admin_page_contract_violation",
151 page=self._page_cls.__name__,
152 result_type=type(response).__name__,
153 )
154 response = await _placeholder_page(request, self._container)
155 except Exception:
156 logger.exception(
157 "admin_page_handler_error",
158 page=self._page_cls.__name__,
159 )
160 response = await _placeholder_page(request, self._container)
162 try:
163 is_htmx = wants_fragment(request)
164 except KeyError:
165 is_htmx = False
166 if isinstance(response, HTMLResponse):
167 response = await self._apply_cluster_header(request, response)
168 if not is_htmx and isinstance(response, HTMLResponse):
169 response = await self._wrap_in_shell(request, response)
171 await response(scope, receive, send)
173 async def _apply_cluster_header(
174 self,
175 request: StarletteRequest,
176 response: HTMLResponse,
177 ) -> HTMLResponse:
178 """Drop the page's own inline title/description block.
180 Cluster pages receive a single top-level header rendered above
181 the whole center layout (see ``_cluster_header_html``), so the
182 page's inline header is removed. Only applies to pages living
183 inside a cluster center (e.g. ``/admin/infrastructure/...``).
184 """
185 state = getattr(request, "app", None)
186 groups = (
187 getattr(state.state, "assembler_groups", None)
188 if state and hasattr(state, "state")
189 else None
190 )
191 if not is_cluster_path(request.url.path, cluster_items(groups)):
192 return response
194 content = (
195 response.body.decode()
196 if isinstance(response.body, bytes)
197 else str(response.body)
198 )
199 return HTMLResponse(_PAGE_HEADER_RE.sub("", content, count=1))
201 async def _wrap_in_shell(
202 self,
203 request: StarletteRequest,
204 response: HTMLResponse,
205 ) -> HTMLResponse:
206 from pathlib import Path
208 from starlette.templating import Jinja2Templates
210 from lexigram.admin.engine.renderer import resolve_admin_nav
211 from lexigram.admin.ui.templates.shell import AdminShell
212 from lexigram.ui import raw, render_to_string
214 content = (
215 response.body.decode()
216 if isinstance(response.body, bytes)
217 else str(response.body)
218 )
220 title = self._page_cls.__name__.removesuffix("Page")
222 user = (
223 getattr(request.state, "user", None) if hasattr(request, "state") else None
224 )
225 nav_items, system_menu_items, secondary_nav = resolve_admin_nav(request)
226 state = getattr(request, "app", None)
227 groups = (
228 getattr(state.state, "assembler_groups", None)
229 if state and hasattr(state, "state")
230 else None
231 )
232 is_cluster = is_cluster_path(request.url.path, cluster_items(groups))
233 if secondary_nav:
234 from lexigram.admin.ui.organisms.secondary_nav import ClusterLayout
236 content = render_to_string(
237 ClusterLayout(items=secondary_nav, content=raw(content))
238 )
239 if is_cluster:
240 content = _cluster_header_html() + content
242 breadcrumbs: list[dict[str, str]] | None = None
243 if secondary_nav and is_cluster:
244 breadcrumbs = [
245 {"label": "Home", "url": "/admin/"},
246 {"label": CLUSTER_LABEL, "url": CLUSTER_URL},
247 ]
248 path = request.url.path
249 for item in secondary_nav:
250 item_href = item.get("href", "")
251 if path == item_href:
252 breadcrumbs.append({"label": item.get("label", ""), "url": ""})
253 title = item.get("label", title)
254 break
255 child = next(
256 (c for c in item.get("children", []) if path == c.get("href", "")),
257 None,
258 )
259 if child is not None:
260 breadcrumbs.append(
261 {"label": item.get("label", ""), "url": item_href}
262 )
263 breadcrumbs.append({"label": child.get("label", ""), "url": ""})
264 title = child.get("label", title)
265 break
267 theme_css = ""
268 try:
269 from lexigram.admin.theme.service import AdminThemeService
271 service = AdminThemeService(
272 primary_color=await _resolve_primary_color(self._container)
273 )
274 theme_css = service.generate_theme_css()
275 except Exception: # noqa: BLE001, S110 — non-fatal
276 pass
278 from lexigram.admin.navigation.manager import NavigationManager
280 user_menu_items: list[dict[str, str | None]] = (
281 NavigationManager(request).user_menu_items() if request is not None else []
282 )
284 branding: dict[str, str] = {}
285 try:
286 from lexigram.admin.multitenancy.adapter import resolve_tenant_id
287 from lexigram.admin.services.settings_service import (
288 resolve_admin_settings_service,
289 )
291 container = (
292 getattr(request.state, "root_container", None)
293 or getattr(request.state, "container", None)
294 or getattr(request.app.state, "container", None)
295 or self._container
296 )
297 settings_service = await resolve_admin_settings_service(container)
298 if settings_service is not None:
299 tenant = await resolve_tenant_id(request, default="default")
300 overrides = await settings_service.get_all(tenant)
301 for field in ("primary_color", "site_name", "logo_url", "dark_mode"):
302 value = overrides.get(field) or overrides.get(
303 f"admin.branding.{field}"
304 )
305 if value:
306 branding[field] = value
307 if branding.get("primary_color"):
308 from lexigram.admin.theme.service import AdminThemeService
310 theme_css = AdminThemeService(
311 primary_color=branding["primary_color"]
312 ).generate_theme_css()
313 except Exception: # noqa: BLE001, S110 — non-fatal
314 pass
316 shell = AdminShell(
317 content=content,
318 title=title,
319 user=user,
320 nav_items=nav_items,
321 system_menu_items=system_menu_items,
322 user_menu_items=user_menu_items,
323 breadcrumbs=breadcrumbs,
324 theme_css=theme_css,
325 **cast(
326 "Any",
327 {
328 k: v
329 for k, v in branding.items()
330 if k in ("dark_mode", "site_name", "logo_url")
331 },
332 ),
333 )
334 shell_html = render_to_string(shell)
336 templates_dir = Path(__file__).resolve().parent.parent / "views" / "templates"
337 templates = Jinja2Templates(directory=str(templates_dir))
338 return templates.TemplateResponse(
339 request,
340 "admin_shell.html",
341 context={
342 "content": Markup(shell_html), # noqa: S704 — framework-composed trusted HTML
343 "title": title,
344 "dark_mode": branding.get("dark_mode", ""),
345 },
346 )
348 async def _resolve_page(self) -> Any:
349 """Resolve page instance from container.
351 Uses ``container.call(cls.__init__)`` to resolve each constructor
352 parameter from the DI container, then constructs the instance
353 manually. This is necessary because ``get_type_hints(cls)``
354 returns an empty dict for classes with ``from __future__ import
355 annotations`` (PEP 563), so ``container.call(cls)`` cannot
356 discover parameter types.
357 """
358 init_method = self._page_cls.__init__ # type: ignore[misc]
359 sig = inspect.signature(init_method)
360 hints = get_type_hints(init_method)
361 kwargs: dict[str, Any] = {}
362 for name, param in sig.parameters.items():
363 if name == "self":
364 continue
365 if param.kind in (
366 inspect.Parameter.VAR_POSITIONAL,
367 inspect.Parameter.VAR_KEYWORD,
368 ):
369 continue
370 param_type = hints.get(name)
371 if param_type is not None:
372 try:
373 resolution_target = _strip_optional(param_type)
374 kwargs[name] = await self._container.resolve(resolution_target)
375 continue
376 except UnresolvableDependencyError:
377 pass
378 if param.default is not inspect.Parameter.empty:
379 kwargs[name] = param.default
380 elif param_type is not None:
381 raise UnresolvableDependencyError(
382 f"Cannot resolve parameter {name!r} for "
383 f"{self._page_cls.__name__}: type {param_type} not registered.",
384 )
385 else:
386 raise UnresolvableDependencyError(
387 f"Cannot resolve parameter {name!r} for "
388 f"{self._page_cls.__name__}: no type hint and no default.",
389 )
390 return self._page_cls(**kwargs)
393async def _placeholder_page(
394 request: Any,
395 container: Any | None = None,
396) -> HTMLResponse:
397 """Placeholder for admin pages without an implemented handler.
399 For HTMX requests returns only the content fragment (no shell) so
400 the sidebar/topbar from the existing page stays intact. For direct
401 navigation returns the full admin layout.
403 Args:
404 request: Starlette request.
405 container: Optional resolver for theme settings.
406 """
407 content = (
408 '<div class="flex items-center justify-center h-64">'
409 '<div class="text-center">'
410 '<h2 class="text-xl font-semibold text-muted-foreground">Under Construction</h2>'
411 '<p class="text-muted-foreground mt-2">This page has not been implemented yet.</p>'
412 "</div></div>"
413 )
415 try:
416 from lexigram.admin.engine.renderer import resolve_admin_nav
418 nav_items, system_menu_items, secondary_nav = resolve_admin_nav(request)
419 except Exception: # noqa: BLE001 — non-fatal
420 nav_items, system_menu_items, secondary_nav = [], [], None
422 if secondary_nav:
423 from lexigram.admin.ui.organisms.secondary_nav import ClusterLayout
424 from lexigram.ui import raw, render_to_string
426 content = render_to_string(
427 ClusterLayout(items=secondary_nav, content=raw(content))
428 )
430 is_htmx = wants_fragment(request)
432 if is_htmx:
433 return HTMLResponse(content)
435 try:
436 from pathlib import Path
438 from starlette.templating import Jinja2Templates
440 from lexigram.admin.ui.templates.shell import AdminShell
441 from lexigram.ui import render_to_string
443 user = (
444 getattr(request.state, "user", None) if hasattr(request, "state") else None
445 )
447 from lexigram.admin.navigation.manager import NavigationManager
449 user_menu_items = (
450 NavigationManager(request).user_menu_items(include_plugins=False)
451 if request is not None
452 else []
453 )
455 theme_css = ""
456 try:
457 from lexigram.admin.theme.service import AdminThemeService
459 service = AdminThemeService(
460 primary_color=(
461 await _resolve_primary_color(container)
462 if container is not None
463 else _DEFAULT_PRIMARY_COLOR
464 )
465 )
466 theme_css = service.generate_theme_css()
467 except Exception: # noqa: BLE001, S110 — non-fatal
468 pass
470 shell = AdminShell(
471 content=content,
472 title="Under Construction",
473 user=user,
474 nav_items=nav_items,
475 system_menu_items=system_menu_items,
476 user_menu_items=user_menu_items,
477 theme_css=theme_css,
478 )
479 shell_html = render_to_string(shell)
481 templates_dir = Path(__file__).resolve().parent.parent / "views" / "templates"
482 templates = Jinja2Templates(directory=str(templates_dir))
483 return templates.TemplateResponse(
484 request,
485 "admin_shell.html",
486 context={
487 "content": shell_html,
488 "title": "Under Construction",
489 "dark_mode": "",
490 },
491 )
492 except Exception:
493 return HTMLResponse(content)
496def _resolve_handler(handler: Any) -> Any:
497 """Resolve a string dotted-path handler to the actual callable."""
498 if not isinstance(handler, str):
499 return handler
500 try:
501 module_path, _, func_name = handler.partition(":")
502 mod = __import__(module_path, fromlist=[func_name])
503 resolved = getattr(mod, func_name, None)
504 if resolved is None:
505 logger.warning("handler_import_failed", handler=handler)
506 return resolved
507 except Exception: # noqa: BLE001
508 logger.warning("handler_import_failed", handler=handler, exc_info=True)
509 return None
512class StructuredPageHandler:
513 """Wrap management page handlers so only ``PageContent`` reaches the browser.
515 Starlette treats class endpoints as ASGI apps (``__call__(scope, receive,
516 send)``), so this wrapper builds a ``StarletteRequest`` from the ASGI scope
517 before delegating to the page handler.
519 Any other return (str, HTMLResponse, template, ...) is a contract
520 violation: it is logged and replaced with an error page.
521 """
523 def __init__(self, handler: Any) -> None:
524 self._handler = handler
526 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
527 from lexigram.admin.dashboard.page_renderer import render_page_content
529 request = StarletteRequest(scope, receive, send)
530 handler = self._handler
531 callable_handler = handler.handle if hasattr(handler, "handle") else handler
532 result = await callable_handler(request)
533 if isinstance(result, PageContent):
534 response = render_page_content(result)
535 else:
536 logger.error(
537 "page_contract_violation",
538 handler=type(self._handler).__name__,
539 result_type=type(result).__name__,
540 )
541 response = render_page_content(
542 PageContent(
543 title="Page Contract Violation",
544 body=EmptyContent(
545 title="Invalid Page Content",
546 message=(
547 "The page handler returned raw HTML. "
548 "Convert it to PageContent."
549 ),
550 icon="alert-triangle",
551 ),
552 )
553 )
554 await response(scope, receive, send)
557def _register_pages(
558 router: AdminRouter,
559 naming_policy: NamingPolicy,
560 prefix: str,
561 pages: list[ManagementPageDefinition],
562 container: Any = None,
563) -> None:
564 """Register management page routes on the admin router.
566 Routes are registered relative to the admin app mount point
567 (the prefix is stripped before registration — the admin router
568 already lives under the admin prefix due to ``AdminRouter.mount()``).
569 ``registered_internal_paths`` is updated externally so that
570 ``_ensure_nav_route`` does not create duplicate placeholders.
571 """
572 # prefix is intentionally unused — routes live inside the mounted
573 # admin app and must be relative to its mount point.
574 for page in pages:
575 handler = _resolve_handler(page.handler)
576 if handler is None:
577 continue
578 if inspect.isclass(handler) and container is not None:
579 handler = AdminPageHandler(handler, container)
580 else:
581 handler = StructuredPageHandler(handler)
582 path = page.route_path
583 if not path.startswith("/"):
584 path = f"/{path}"
585 ns_name = naming_policy.namespaced(page.contributor, page.name)
586 naming_policy.register("page", ns_name)
587 router.add_route(path=path, method="GET", handler=handler, name=ns_name)
590def _register_settings(
591 router: AdminRouter,
592 naming_policy: NamingPolicy,
593 prefix: str,
594 panels: list[SettingsPanelDefinition],
595 container: Any = None,
596) -> None:
597 """Register settings panel routes on the admin router.
599 Routes are registered relative to the admin app mount point
600 (the prefix is stripped before registration — see ``_register_pages``).
601 """
602 for panel in panels:
603 handler = _resolve_handler(panel.handler)
604 if handler is None:
605 continue
606 if inspect.isclass(handler) and container is not None:
607 handler = AdminPageHandler(handler, container)
608 else:
609 handler = StructuredPageHandler(handler)
610 path = panel.route_path
611 if not path.startswith("/"):
612 path = f"/{path}"
613 ns_name = naming_policy.namespaced(panel.contributor, panel.name)
614 naming_policy.register("panel", ns_name)
615 router.add_route(path=path, method="GET", handler=handler, name=ns_name)
618class RouteIntegrator:
619 """Collects ``AdminRouteSpec``, ``ManagementPageDefinition``, and
620 ``SettingsPanelDefinition`` from contributors and registers them
621 on the router."""
623 def __init__(
624 self,
625 *,
626 router: AdminRouter,
627 naming_policy: NamingPolicy,
628 route_prefix: str = "",
629 container: Any = None,
630 ) -> None:
631 self._router = router
632 self._naming = naming_policy
633 self._prefix = route_prefix
634 self._container = container
636 def register(self, contributors: Sequence[BaseAdminContributor]) -> None:
637 """Register each contributor's routes, management pages, and
638 settings panels on the admin router. Nav-item URLs that don't
639 have a corresponding handler automatically get a placeholder
640 route so they never 404."""
641 registered_internal_paths: set[str] = set()
643 for c in contributors:
644 # Routes
645 for spec in c.get_routes():
646 ns_name = self._naming.namespaced(c.package_source, spec.name)
647 self._naming.register("route", ns_name)
648 path = spec.path
649 # Contributor specs carry the full URL (e.g. "/admin/...")
650 # but routes live inside the mounted admin app, so strip
651 # the mount prefix like _ensure_nav_route does.
652 if self._prefix and path.startswith(self._prefix):
653 path = path[len(self._prefix) :]
654 if not path:
655 path = "/"
656 registered_internal_paths.add(path)
657 self._router.add_route(
658 path=path,
659 method=spec.method,
660 handler=spec.handler,
661 name=ns_name,
662 )
664 # Management pages
665 pages = c.get_management_pages()
666 if pages:
667 for page in pages:
668 internal = page.route_path
669 if not internal.startswith("/"):
670 internal = f"/{internal}"
671 registered_internal_paths.add(internal)
672 _register_pages(
673 self._router,
674 self._naming,
675 self._prefix,
676 pages, # type: ignore[arg-type]
677 container=self._container,
678 )
680 # Settings panels
681 panels = c.get_settings_panels()
682 if panels:
683 for panel in panels:
684 internal = panel.route_path
685 if not internal.startswith("/"):
686 internal = f"/{internal}"
687 registered_internal_paths.add(internal)
688 _register_settings(
689 self._router,
690 self._naming,
691 self._prefix,
692 panels, # type: ignore[arg-type]
693 container=self._container,
694 )
696 # Auto-register placeholder routes for nav items without handlers.
697 for c in contributors:
698 for item in c.get_navigation_items():
699 self._ensure_nav_route(item, registered_internal_paths)
700 for child in item.children or ():
701 self._ensure_nav_route(child, registered_internal_paths)
703 # Cluster areas are also reachable under the center namespace
704 # (e.g. /admin/infrastructure/web), mirroring how settings
705 # sub-pages nest below /admin/settings. Aliases share the source
706 # route's handler — real page or placeholder alike.
707 for c in contributors:
708 for item in c.get_navigation_items():
709 self._register_cluster_alias(item)
710 for child in item.children or ():
711 self._register_cluster_alias(child)
713 def _ensure_nav_route(
714 self,
715 item: Any,
716 registered_paths: set[str],
717 ) -> None:
718 """Register a placeholder route for *item* if its URL isn't covered."""
719 url = item.url
720 if not url or url.startswith("http"):
721 return
723 internal = url
724 if self._prefix and internal.startswith(self._prefix):
725 internal = internal[len(self._prefix) :]
726 if not internal:
727 internal = "/"
729 if internal in registered_paths or url in registered_paths:
730 return
732 safe_label = item.label.lower().replace(" ", "_").replace("/", "_")
733 self._router.add_route(
734 path=internal,
735 method="GET",
736 handler=_placeholder_page,
737 name=f"placeholder_{safe_label}",
738 )
739 registered_paths.add(internal)
741 def _register_cluster_alias(self, item: Any) -> None:
742 """Register a namespaced alias for a cluster group nav item.
744 Cluster areas live under the center namespace (``/admin/
745 infrastructure/web``) in addition to their contributor URL. When
746 the source URL has a real handler, the alias reuses it; otherwise
747 the alias falls back to the placeholder page.
748 """
749 if getattr(item, "group", None) != CLUSTER_GROUP:
750 return
751 url = item.url
752 if not url or url.startswith("http"):
753 return
754 namespaced = cluster_child_href(url)
755 if not namespaced or namespaced == url:
756 return
758 internal = url
759 internal_ns = namespaced
760 if self._prefix and internal.startswith(self._prefix):
761 internal = internal[len(self._prefix) :]
762 if self._prefix and internal_ns.startswith(self._prefix):
763 internal_ns = internal_ns[len(self._prefix) :]
764 if internal_ns == internal:
765 return
767 safe_label = item.label.lower().replace(" ", "_").replace("/", "_")
768 if not self._router.alias_route(
769 internal,
770 internal_ns,
771 name=f"cluster_alias_{safe_label}",
772 ):
773 self._router.add_route(
774 path=internal_ns,
775 method="GET",
776 handler=_placeholder_page,
777 name=f"cluster_alias_{safe_label}",
778 )