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