Coverage for src / lexigram / admin / dashboard / assembler.py: 37%
63 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Dashboard assembler — implements AdminDashboardProtocol."""
3from __future__ import annotations
5from collections.abc import Sequence
6from typing import TYPE_CHECKING
8from lexigram.admin.contributors.exceptions import (
9 ContributorNotFoundError,
10 ContributorPermissionError,
11)
12from lexigram.contracts.admin.protocols import AdminContributorProtocol
13from lexigram.contracts.admin.types import (
14 AdminActionDefinition,
15 DashboardWidgetDefinition,
16 ManagementPageDefinition,
17 NavigationContribution,
18 SettingsPanelDefinition,
19)
20from lexigram.contracts.core.health import HealthCheckResult, HealthStatus
21from lexigram.logging import get_logger
23if TYPE_CHECKING:
24 from lexigram.admin.dashboard.page_assembler import PageAssembler
25 from lexigram.admin.dashboard.settings_assembler import SettingsPanelAssembler
26 from lexigram.admin.types import AdminUser
28logger = get_logger(__name__)
31class DashboardAssembler:
32 """Assembles the admin dashboard from contributor definitions.
34 Collects widgets, navigation, health definitions, pages, settings
35 panels, and actions from all registered contributors and presents a
36 unified view. Enforces RBAC on ``execute_action`` via each
37 contributor's ``required_permissions`` attribute.
39 Args:
40 contributors: Ordered sequence of contributors to assemble.
41 page_assembler: Optional PageAssembler for management pages.
42 settings_assembler: Optional SettingsPanelAssembler for settings.
43 """
45 def __init__(
46 self,
47 contributors: Sequence[AdminContributorProtocol],
48 *,
49 page_assembler: PageAssembler | None = None,
50 settings_assembler: SettingsPanelAssembler | None = None,
51 ) -> None:
52 self._contributors: Sequence[AdminContributorProtocol] = contributors
53 self._contributors_by_id: dict[str, AdminContributorProtocol] = {
54 c.contributor_id: c for c in contributors
55 }
56 self._page_assembler = page_assembler
57 self._settings_assembler = settings_assembler
59 async def get_all_widgets(self) -> Sequence[DashboardWidgetDefinition]:
60 """Collect and sort widgets from all contributors by category then order.
62 When two contributors register a widget with the same ``name``, a
63 ``admin_widget_conflict`` warning is emitted and the later contributor's
64 widget replaces the earlier one (last writer wins).
65 """
66 seen: dict[str, str] = {} # name → contributor_id of last writer
67 by_name: dict[str, DashboardWidgetDefinition] = {}
68 for contributor in self._contributors:
69 for widget in contributor.get_dashboard_widgets():
70 if widget.name in seen:
71 logger.warning(
72 "admin_widget_conflict",
73 widget_name=widget.name,
74 first_contributor=seen[widget.name],
75 second_contributor=contributor.contributor_id,
76 )
77 seen[widget.name] = contributor.contributor_id
78 by_name[widget.name] = widget
79 return sorted(by_name.values(), key=lambda w: (w.category.value, w.order))
81 async def get_all_navigation(self) -> Sequence[NavigationContribution]:
82 """Collect and sort navigation from all contributors by group then order.
84 When two contributors register a navigation item with the same ``label``,
85 an ``admin_navigation_conflict`` warning is emitted and the later
86 contributor's item replaces the earlier one (last writer wins).
87 """
88 seen: dict[str, str] = {} # label → contributor_id of last writer
89 by_label: dict[str, NavigationContribution] = {}
90 for contributor in self._contributors:
91 for item in contributor.get_navigation_items():
92 if item.label in seen:
93 logger.warning(
94 "admin_navigation_conflict",
95 item_label=item.label,
96 first_contributor=seen[item.label],
97 second_contributor=contributor.contributor_id,
98 )
99 seen[item.label] = contributor.contributor_id
100 by_label[item.label] = item
101 return sorted(by_label.values(), key=lambda n: (n.group, n.order))
103 async def get_framework_health(self) -> dict[str, HealthCheckResult]:
104 """Collect health definitions from all contributors.
106 Returns a dict mapping health definition name to a placeholder
107 HealthCheckResult. Actual checks are performed lazily via
108 the ``check_endpoint`` HTMX calls.
109 """
110 results: dict[str, HealthCheckResult] = {}
111 for contributor in self._contributors:
112 for health_def in contributor.get_health_definitions():
113 results[health_def.name] = HealthCheckResult(
114 component=health_def.component,
115 status=HealthStatus.UNKNOWN,
116 message="Pending health check",
117 )
118 return results
120 async def get_all_actions(self) -> Sequence[AdminActionDefinition]:
121 """Collect actions from all contributors."""
122 actions: list[AdminActionDefinition] = []
123 for contributor in self._contributors:
124 actions.extend(contributor.get_actions())
125 return actions
127 async def get_management_pages(
128 self,
129 user: AdminUser | None = None,
130 ) -> list[ManagementPageDefinition]:
131 """Collect management pages via PageAssembler or return empty."""
132 if self._page_assembler is not None:
133 return self._page_assembler.assemble(self._contributors, user=user) # type: ignore[arg-type]
134 return []
136 async def get_settings_panels(
137 self,
138 user: AdminUser | None = None,
139 ) -> list[SettingsPanelDefinition]:
140 """Collect settings panels via SettingsPanelAssembler or return empty."""
141 if self._settings_assembler is not None:
142 return self._settings_assembler.assemble(self._contributors, user=user) # type: ignore[arg-type]
143 return []
145 async def execute_action(
146 self,
147 contributor_id: str,
148 action_name: str,
149 params: dict[str, object],
150 user_permissions: frozenset[str],
151 ) -> object:
152 """Execute a framework-level action after RBAC enforcement.
154 Looks up the contributor by *contributor_id*, verifies that
155 *user_permissions* is a superset of the contributor's
156 ``required_permissions``, then delegates to
157 ``contributor.execute_action(action_name, params)``.
159 Args:
160 contributor_id: Identifier of the target contributor.
161 action_name: Name of the action to execute.
162 params: Parameters forwarded to the action handler.
163 user_permissions: Permissions held by the requesting user.
165 Returns:
166 Whatever the contributor's action handler returns.
168 Raises:
169 ContributorNotFoundError: If no contributor matches *contributor_id*.
170 ContributorPermissionError: If *user_permissions* is missing any
171 permissions declared by the contributor.
172 """
173 contributor = self._contributors_by_id.get(contributor_id)
174 if contributor is None:
175 raise ContributorNotFoundError(contributor_id)
177 missing = contributor.required_permissions - user_permissions
178 if missing:
179 raise ContributorPermissionError(contributor_id, action_name, missing)
181 return await contributor.execute_action(action_name, params) # type: ignore[attr-defined]
184__all__ = ["DashboardAssembler"]