Coverage for src/lexigram/admin/dashboard/page_handlers.py: 0%
228 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1from __future__ import annotations
3import inspect
4import re
5import types
6from typing import TYPE_CHECKING, Any, cast, get_args, get_origin, get_type_hints
8from markupsafe import Markup
9from starlette.requests import Request as StarletteRequest
10from starlette.responses import HTMLResponse
12from lexigram.admin.navigation.clusters import (
13 CLUSTER_LABEL,
14 CLUSTER_URL,
15 cluster_items,
16 is_cluster_path,
17)
18from lexigram.admin.state.context import wants_fragment
19from lexigram.contracts.admin.page_content import PageContent
20from lexigram.contracts.admin.widget_content import EmptyContent
21from lexigram.contracts.exceptions import UnresolvableDependencyError
22from lexigram.logging import get_logger
24if TYPE_CHECKING:
25 from lexigram.contracts.core.di import ContainerResolverProtocol
27logger = get_logger(__name__)
30def _strip_optional(tp: Any) -> Any:
31 """If *tp* is ``Optional[X]`` (``Union[X, None]`` or ``X | None``),
32 return ``X``. Otherwise return *tp* unchanged."""
33 origin = get_origin(tp)
34 if origin is types.UnionType:
35 args = get_args(tp)
36 non_none = [a for a in args if a is not type(None)]
37 if len(non_none) == 1:
38 return non_none[0]
39 return tp
42_DEFAULT_PRIMARY_COLOR = "#6b7280"
44_CLUSTER_HEADER_DESCRIPTION = (
45 "Monitor and manage the services powering your application: web, data, "
46 "and runtime areas."
47)
50def _cluster_header_html() -> str:
51 """Render the cluster center top-level title + description block.
53 Mirrors the settings center header: a ``mb-2`` wrapper with the
54 section title and a muted one-line description, rendered above the
55 whole center layout (secondary sidebar and page content).
56 """
57 from lexigram.ui import el, render_to_string
59 return render_to_string(
60 el(
61 "div",
62 el("h1", CLUSTER_LABEL, class_="text-2xl font-bold text-foreground"),
63 el(
64 "p",
65 _CLUSTER_HEADER_DESCRIPTION,
66 class_="text-muted-foreground mt-1",
67 ),
68 class_="mb-2",
69 )
70 )
73#: First header block rendered by management pages (h1 + description +
74#: divider). Cluster pages have their own top-level header injected by the
75#: shell wrapper, so this inline header is dropped.
76_PAGE_HEADER_RE = re.compile(
77 r"<h1[^>]*>.*?</h1>\s*<p[^>]*>.*?</p>\s*(?:<hr[^>]*/?>)?",
78 re.S,
79)
82async def _resolve_primary_color(container: Any) -> str:
83 """Resolve the saved branding primary color, best-effort.
85 Falls back to the framework default when no registry/db store is
86 available.
87 """
88 try:
89 from lexigram.admin.settings.panel.registry import ConfigRegistry
91 registry = await container.resolve(
92 ConfigRegistry,
93 bypass_visibility=True,
94 )
95 values = await registry.get_values("admin.branding", "db")
96 color = values.get("primary_color")
97 if color:
98 return str(color)
99 except Exception: # noqa: BLE001 — non-fatal
100 logger.exception("admin.theme_overrides_failed")
101 return _DEFAULT_PRIMARY_COLOR
104class AdminPageHandler:
105 """ASGI adapter that resolves a management page handler from the DI
106 container at request time and delegates to its ``handle()`` method.
108 Starlette treats class endpoints as ASGI apps — it calls
109 ``cls(scope, receive, send)`` which becomes ``__init__(scope, receive,
110 send)``. Management page handlers use keyword-only constructor DI
111 (``def __init__(self, *, repo: ..., ...)``), so direct registration
112 always raises TypeError. This wrapper sidesteps that by storing the
113 page **class** at route-build time and resolving an instance from the
114 container at request time.
115 """
117 def __init__(
118 self,
119 page_cls: type,
120 container: ContainerResolverProtocol,
121 ) -> None:
122 self._page_cls = page_cls
123 self._container = container
125 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
126 request = StarletteRequest(scope, receive, send)
127 try:
128 instance = await self._resolve_page()
129 response = await instance.handle(request)
130 if not isinstance(response, HTMLResponse):
131 from lexigram.admin.dashboard.page_renderer import (
132 render_page_content,
133 )
134 from lexigram.contracts.admin.page_content import PageContent
136 if isinstance(response, PageContent):
137 response = render_page_content(response)
138 else:
139 logger.error(
140 "admin_page_contract_violation",
141 page=self._page_cls.__name__,
142 result_type=type(response).__name__,
143 )
144 response = await _placeholder_page(request, self._container)
145 except Exception:
146 logger.exception(
147 "admin_page_handler_error",
148 page=self._page_cls.__name__,
149 )
150 response = await _placeholder_page(request, self._container)
152 try:
153 is_htmx = wants_fragment(request)
154 except KeyError:
155 is_htmx = False
156 if isinstance(response, HTMLResponse):
157 response = await self._apply_cluster_header(request, response)
158 if not is_htmx and isinstance(response, HTMLResponse):
159 response = await self._wrap_in_shell(request, response)
161 await response(scope, receive, send)
163 async def _apply_cluster_header(
164 self,
165 request: StarletteRequest,
166 response: HTMLResponse,
167 ) -> HTMLResponse:
168 """Drop the page's own inline title/description block.
170 Cluster pages receive a single top-level header rendered above
171 the whole center layout (see ``_cluster_header_html``), so the
172 page's inline header is removed. Only applies to pages living
173 inside a cluster center (e.g. ``/admin/infrastructure/...``).
174 """
175 state = getattr(request, "app", None)
176 groups = (
177 getattr(state.state, "assembler_groups", None)
178 if state and hasattr(state, "state")
179 else None
180 )
181 if not is_cluster_path(request.url.path, cluster_items(groups)):
182 return response
184 content = (
185 response.body.decode()
186 if isinstance(response.body, bytes)
187 else str(response.body)
188 )
189 return HTMLResponse(_PAGE_HEADER_RE.sub("", content, count=1))
191 async def _wrap_in_shell(
192 self,
193 request: StarletteRequest,
194 response: HTMLResponse,
195 ) -> HTMLResponse:
196 from pathlib import Path
198 from starlette.templating import Jinja2Templates
200 from lexigram.admin.engine.renderer import resolve_admin_nav
201 from lexigram.admin.ui.templates.shell import AdminShell
202 from lexigram.ui import raw, render_to_string
204 content = (
205 response.body.decode()
206 if isinstance(response.body, bytes)
207 else str(response.body)
208 )
210 title = self._page_cls.__name__.removesuffix("Page")
212 user = (
213 getattr(request.state, "user", None) if hasattr(request, "state") else None
214 )
215 nav_items, system_menu_items, secondary_nav = resolve_admin_nav(request)
216 state = getattr(request, "app", None)
217 groups = (
218 getattr(state.state, "assembler_groups", None)
219 if state and hasattr(state, "state")
220 else None
221 )
222 is_cluster = is_cluster_path(request.url.path, cluster_items(groups))
223 if secondary_nav:
224 from lexigram.admin.ui.organisms.secondary_nav import ClusterLayout
226 content = render_to_string(
227 ClusterLayout(items=secondary_nav, content=raw(content))
228 )
229 if is_cluster:
230 content = _cluster_header_html() + content
232 breadcrumbs: list[dict[str, str]] | None = None
233 if secondary_nav and is_cluster:
234 breadcrumbs = [
235 {"label": "Home", "url": "/admin/"},
236 {"label": CLUSTER_LABEL, "url": CLUSTER_URL},
237 ]
238 path = request.url.path
239 for item in secondary_nav:
240 item_href = item.get("href", "")
241 if path == item_href:
242 breadcrumbs.append({"label": item.get("label", ""), "url": ""})
243 title = item.get("label", title)
244 break
245 child = next(
246 (c for c in item.get("children", []) if path == c.get("href", "")),
247 None,
248 )
249 if child is not None:
250 breadcrumbs.append(
251 {"label": item.get("label", ""), "url": item_href}
252 )
253 breadcrumbs.append({"label": child.get("label", ""), "url": ""})
254 title = child.get("label", title)
255 break
257 theme_css = ""
258 try:
259 from lexigram.admin.theme.service import AdminThemeService
261 service = AdminThemeService(
262 primary_color=await _resolve_primary_color(self._container)
263 )
264 theme_css = service.generate_theme_css()
265 except Exception: # noqa: BLE001, S110 — non-fatal
266 pass
268 from lexigram.admin.navigation.manager import NavigationManager
270 user_menu_items: list[dict[str, str | None]] = (
271 NavigationManager(request).user_menu_items() if request is not None else []
272 )
274 branding: dict[str, str] = {}
275 try:
276 from lexigram.admin.multitenancy.adapter import resolve_tenant_id
277 from lexigram.admin.services.settings_service import (
278 resolve_admin_settings_service,
279 )
281 container = (
282 getattr(request.state, "root_container", None)
283 or getattr(request.state, "container", None)
284 or getattr(request.app.state, "container", None)
285 or self._container
286 )
287 settings_service = await resolve_admin_settings_service(container)
288 if settings_service is not None:
289 tenant = await resolve_tenant_id(request, default="default")
290 overrides = await settings_service.get_all(tenant)
291 for field in ("primary_color", "site_name", "logo_url", "dark_mode"):
292 value = overrides.get(field) or overrides.get(
293 f"admin.branding.{field}"
294 )
295 if value:
296 branding[field] = value
297 if branding.get("primary_color"):
298 from lexigram.admin.theme.service import AdminThemeService
300 theme_css = AdminThemeService(
301 primary_color=branding["primary_color"]
302 ).generate_theme_css()
303 except Exception: # noqa: BLE001, S110 — non-fatal
304 pass
306 shell = AdminShell(
307 content=content,
308 title=title,
309 user=user,
310 nav_items=nav_items,
311 system_menu_items=system_menu_items,
312 user_menu_items=user_menu_items,
313 breadcrumbs=breadcrumbs,
314 theme_css=theme_css,
315 **cast(
316 "Any",
317 {
318 k: v
319 for k, v in branding.items()
320 if k in ("dark_mode", "site_name", "logo_url")
321 },
322 ),
323 )
324 shell_html = render_to_string(shell)
326 templates_dir = Path(__file__).resolve().parent.parent / "views" / "templates"
327 templates = Jinja2Templates(directory=str(templates_dir))
328 return templates.TemplateResponse(
329 request,
330 "admin_shell.html",
331 context={
332 "content": Markup(shell_html), # noqa: S704 — framework-composed trusted HTML
333 "title": title,
334 "dark_mode": branding.get("dark_mode", ""),
335 },
336 )
338 async def _resolve_page(self) -> Any:
339 """Resolve page instance from container.
341 Uses ``container.call(cls.__init__)`` to resolve each constructor
342 parameter from the DI container, then constructs the instance
343 manually. This is necessary because ``get_type_hints(cls)``
344 returns an empty dict for classes with ``from __future__ import
345 annotations`` (PEP 563), so ``container.call(cls)`` cannot
346 discover parameter types.
347 """
348 init_method = self._page_cls.__init__ # type: ignore[misc]
349 sig = inspect.signature(init_method)
350 hints = get_type_hints(init_method)
351 kwargs: dict[str, Any] = {}
352 for name, param in sig.parameters.items():
353 if name == "self":
354 continue
355 if param.kind in (
356 inspect.Parameter.VAR_POSITIONAL,
357 inspect.Parameter.VAR_KEYWORD,
358 ):
359 continue
360 param_type = hints.get(name)
361 if param_type is not None:
362 try:
363 resolution_target = _strip_optional(param_type)
364 kwargs[name] = await self._container.resolve(resolution_target)
365 continue
366 except UnresolvableDependencyError:
367 pass
368 if param.default is not inspect.Parameter.empty:
369 kwargs[name] = param.default
370 elif param_type is not None:
371 raise UnresolvableDependencyError(
372 f"Cannot resolve parameter {name!r} for "
373 f"{self._page_cls.__name__}: type {param_type} not registered.",
374 )
375 else:
376 raise UnresolvableDependencyError(
377 f"Cannot resolve parameter {name!r} for "
378 f"{self._page_cls.__name__}: no type hint and no default.",
379 )
380 return self._page_cls(**kwargs)
383async def _placeholder_page(
384 request: Any,
385 container: Any | None = None,
386) -> HTMLResponse:
387 """Placeholder for admin pages without an implemented handler.
389 For HTMX requests returns only the content fragment (no shell) so
390 the sidebar/topbar from the existing page stays intact. For direct
391 navigation returns the full admin layout.
393 Args:
394 request: Starlette request.
395 container: Optional resolver for theme settings.
396 """
397 content = (
398 '<div class="flex items-center justify-center h-64">'
399 '<div class="text-center">'
400 '<h2 class="text-xl font-semibold text-muted-foreground">Under Construction</h2>'
401 '<p class="text-muted-foreground mt-2">This page has not been implemented yet.</p>'
402 "</div></div>"
403 )
405 try:
406 from lexigram.admin.engine.renderer import resolve_admin_nav
408 nav_items, system_menu_items, secondary_nav = resolve_admin_nav(request)
409 except Exception: # noqa: BLE001 — non-fatal
410 nav_items, system_menu_items, secondary_nav = [], [], None
412 if secondary_nav:
413 from lexigram.admin.ui.organisms.secondary_nav import ClusterLayout
414 from lexigram.ui import raw, render_to_string
416 content = render_to_string(
417 ClusterLayout(items=secondary_nav, content=raw(content))
418 )
420 is_htmx = wants_fragment(request)
422 if is_htmx:
423 return HTMLResponse(content)
425 try:
426 from pathlib import Path
428 from starlette.templating import Jinja2Templates
430 from lexigram.admin.ui.templates.shell import AdminShell
431 from lexigram.ui import render_to_string
433 user = (
434 getattr(request.state, "user", None) if hasattr(request, "state") else None
435 )
437 from lexigram.admin.navigation.manager import NavigationManager
439 user_menu_items = (
440 NavigationManager(request).user_menu_items(include_plugins=False)
441 if request is not None
442 else []
443 )
445 theme_css = ""
446 try:
447 from lexigram.admin.theme.service import AdminThemeService
449 service = AdminThemeService(
450 primary_color=(
451 await _resolve_primary_color(container)
452 if container is not None
453 else _DEFAULT_PRIMARY_COLOR
454 )
455 )
456 theme_css = service.generate_theme_css()
457 except Exception: # noqa: BLE001, S110 — non-fatal
458 pass
460 shell = AdminShell(
461 content=content,
462 title="Under Construction",
463 user=user,
464 nav_items=nav_items,
465 system_menu_items=system_menu_items,
466 user_menu_items=user_menu_items,
467 theme_css=theme_css,
468 )
469 shell_html = render_to_string(shell)
471 templates_dir = Path(__file__).resolve().parent.parent / "views" / "templates"
472 templates = Jinja2Templates(directory=str(templates_dir))
473 return templates.TemplateResponse(
474 request,
475 "admin_shell.html",
476 context={
477 "content": shell_html,
478 "title": "Under Construction",
479 "dark_mode": "",
480 },
481 )
482 except Exception:
483 return HTMLResponse(content)
486def _resolve_handler(handler: Any) -> Any:
487 """Resolve a string dotted-path handler to the actual callable."""
488 if not isinstance(handler, str):
489 return handler
490 try:
491 module_path, _, func_name = handler.partition(":")
492 mod = __import__(module_path, fromlist=[func_name])
493 resolved = getattr(mod, func_name, None)
494 if resolved is None:
495 logger.warning("handler_import_failed", handler=handler)
496 return resolved
497 except Exception: # noqa: BLE001
498 logger.warning("handler_import_failed", handler=handler, exc_info=True)
499 return None
502class StructuredPageHandler:
503 """Wrap management page handlers so only ``PageContent`` reaches the browser.
505 Starlette treats class endpoints as ASGI apps (``__call__(scope, receive,
506 send)``), so this wrapper builds a ``StarletteRequest`` from the ASGI scope
507 before delegating to the page handler.
509 Any other return (str, HTMLResponse, template, ...) is a contract
510 violation: it is logged and replaced with an error page.
511 """
513 def __init__(self, handler: Any) -> None:
514 self._handler = handler
516 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
517 from lexigram.admin.dashboard.page_renderer import render_page_content
519 request = StarletteRequest(scope, receive, send)
520 handler = self._handler
521 callable_handler = handler.handle if hasattr(handler, "handle") else handler
522 result = await callable_handler(request)
523 if isinstance(result, PageContent):
524 response = render_page_content(result)
525 else:
526 logger.error(
527 "page_contract_violation",
528 handler=type(self._handler).__name__,
529 result_type=type(result).__name__,
530 )
531 response = render_page_content(
532 PageContent(
533 title="Page Contract Violation",
534 body=EmptyContent(
535 title="Invalid Page Content",
536 message=(
537 "The page handler returned raw HTML. "
538 "Convert it to PageContent."
539 ),
540 icon="alert-triangle",
541 ),
542 )
543 )
544 await response(scope, receive, send)