Coverage for src / lexigram / contracts / admin / protocols.py: 100%
50 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-15 20:20 +0800
1"""Admin contract protocols — service boundaries for the admin contributor system."""
3from __future__ import annotations
5from collections.abc import Sequence
6from typing import TYPE_CHECKING, Protocol, runtime_checkable
8if TYPE_CHECKING:
9 from lexigram.contracts.admin.errors import AdminError
10 from lexigram.contracts.admin.health_payload import HealthCheckPayload
11 from lexigram.contracts.admin.types import (
12 AdminActionDefinition,
13 AdminHealthDefinition,
14 AdminRouteSpec,
15 DashboardWidgetDefinition,
16 ManagementPageDefinition,
17 NavigationContribution,
18 SettingsPanelDefinition,
19 WidgetParams,
20 WidgetViewModel,
21 )
22 from lexigram.contracts.core.di import ContainerResolverProtocol
23 from lexigram.contracts.core.result import Result
26@runtime_checkable
27class AdminContributorProtocol(Protocol):
28 """Contract for packages that contribute admin dashboard surfaces.
30 Any lexigram extension package can implement this protocol and register
31 it via the ``lexigram.admin.contributors`` entry point. The admin
32 dashboard discovers contributors at boot and assembles their widgets,
33 pages, navigation, and actions into the unified admin UI.
34 """
36 @property
37 def name(self) -> str:
38 """Unique contributor identifier (e.g. 'cache', 'events', 'ai')."""
39 ...
41 @property
42 def display_name(self) -> str:
43 """Human-readable contributor name for the admin UI."""
44 ...
46 @property
47 def group(self) -> str:
48 """Navigation group this contributor belongs to."""
49 ...
51 @property
52 def icon(self) -> str:
53 """Lucide icon name for the contributor."""
54 ...
56 @property
57 def depends_on(self) -> tuple[str, ...]:
58 """Contributor names that must boot before this contributor."""
59 ...
61 @property
62 def priority(self) -> int:
63 """Ordering priority within its group (lower = first)."""
64 ...
66 @property
67 def version(self) -> str:
68 """Semantic version of this contributor (e.g. '1.2.3')."""
69 ...
71 @property
72 def package_source(self) -> str:
73 """Python package name that provides this contributor (e.g. 'lexigram-cache')."""
74 ...
76 @property
77 def contributor_id(self) -> str:
78 """Stable unique identifier used for lookup and RBAC keying (equals ``name``)."""
79 ...
81 @property
82 def required_permissions(self) -> frozenset[str]:
83 """Permissions a user must hold to execute any action on this contributor."""
84 ...
86 def get_resources(self) -> Sequence[type]:
87 """Return resource classes managed by this contributor."""
88 ...
90 def get_dashboard_widgets(self) -> Sequence[DashboardWidgetDefinition]:
91 """Return widget definitions for the main dashboard."""
92 ...
94 def get_navigation_items(self) -> Sequence[NavigationContribution]:
95 """Return navigation entries for the admin sidebar."""
96 ...
98 def get_management_pages(self) -> Sequence[ManagementPageDefinition]:
99 """Return full management page definitions."""
100 ...
102 def get_settings_panels(self) -> Sequence[SettingsPanelDefinition]:
103 """Return settings panel definitions."""
104 ...
106 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]:
107 """Return health check definitions to surface in the dashboard."""
108 ...
110 def get_actions(self) -> Sequence[AdminActionDefinition]:
111 """Return framework-level actions."""
112 ...
114 def get_routes(self) -> Sequence[AdminRouteSpec]:
115 """Return route specifications for the admin router."""
116 ...
118 async def on_admin_boot(self, container: ContainerResolverProtocol) -> None:
119 """Called when the admin dashboard boots."""
120 ...
122 async def on_admin_shutdown(self) -> None:
123 """Called when the admin dashboard shuts down."""
124 ...
126 async def render_widget(
127 self,
128 widget_name: str,
129 params: WidgetParams,
130 resolver: ContainerResolverProtocol | None = None,
131 ) -> Result[WidgetViewModel, AdminError]:
132 """Render a named widget to a typed WidgetViewModel.
134 Args:
135 widget_name: Name of the widget to render.
136 params: Typed, validated widget parameters.
137 resolver: Optional DI resolver for lazy dependency injection.
139 Returns:
140 Ok(WidgetViewModel) with structured content in ``content`` on success.
141 Err(WidgetNotFoundError) when the widget name is unknown.
142 Err(AdminError) for other expected domain failures.
143 Infrastructure exceptions propagate (not caught here).
144 """
145 ...
147 async def render_health_check(
148 self,
149 check_name: str,
150 ) -> Result[HealthCheckPayload, AdminError]:
151 """Run a health check and return a structured health-check payload.
153 Args:
154 check_name: Name of the health check to run (matches check ID
155 from ``get_health_definitions``).
157 Returns:
158 ``Ok(HealthCheckPayload)`` describing the check result on
159 success — the host renders it as HTML.
160 ``Err(HealthCheckNotFoundError)`` if *check_name* is not served
161 by this contributor.
162 ``Err(AdminError)`` if the check fails for any other reason.
163 """
164 ...
167@runtime_checkable
168class AdminContributorRegistryProtocol(Protocol):
169 """Registry that collects and manages admin contributors."""
171 def register(self, contributor: AdminContributorProtocol) -> None:
172 """Register a contributor."""
173 ...
175 def get(self, name: str) -> AdminContributorProtocol | None:
176 """Get contributor by name."""
177 ...
179 def get_all(self) -> Sequence[AdminContributorProtocol]:
180 """Get all registered contributors, ordered by priority."""
181 ...
183 def get_by_group(self, group: str) -> Sequence[AdminContributorProtocol]:
184 """Get contributors in a specific group."""
185 ...
188@runtime_checkable
189class AdminDashboardProtocol(Protocol):
190 """Protocol for the assembled admin dashboard service."""
192 async def get_all_widgets(self) -> Sequence[DashboardWidgetDefinition]:
193 """Collect widgets from all contributors."""
194 ...
196 async def get_all_navigation(self) -> Sequence[NavigationContribution]:
197 """Collect navigation from all contributors."""
198 ...
200 async def get_framework_health(self) -> dict[str, object]:
201 """Aggregate health from all contributors."""
202 ...
204 async def execute_action(
205 self,
206 contributor_id: str,
207 action_name: str,
208 params: dict[str, object],
209 user_permissions: frozenset[str],
210 ) -> object:
211 """Execute a framework-level action from a contributor.
213 Args:
214 contributor_id: Identifier of the target contributor.
215 action_name: Name of the action to execute.
216 params: Parameters forwarded to the action handler.
217 user_permissions: Permissions held by the requesting user.
219 Returns:
220 Whatever the action handler returns.
221 """
222 ...
225__all__ = [
226 "AdminContributorProtocol",
227 "AdminContributorRegistryProtocol",
228 "AdminDashboardProtocol",
229]