Coverage for src/lexigram/auth/admin/contributor.py: 82%
50 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Admin contributor for lexigram-auth — surfaces session, token, and login
2widgets into the Lexigram admin dashboard.
3"""
5from __future__ import annotations
7from collections.abc import Sequence
8from typing import TYPE_CHECKING, Any, cast
10from lexigram.contracts import Result
11from lexigram.contracts.admin.contributor import BaseAdminContributor
12from lexigram.contracts.admin.errors import AdminError, WidgetNotFoundError
13from lexigram.contracts.admin.types import (
14 AdminActionDefinition,
15 AdminHealthDefinition,
16 DashboardWidgetDefinition,
17 ManagementPageDefinition,
18 NavigationContribution,
19 PageCategory,
20 WidgetCategory,
21 WidgetKind,
22 WidgetParams,
23 WidgetSize,
24 WidgetViewModel,
25)
26from lexigram.contracts.admin.widget_protocols import WidgetHandlerProtocol
27from lexigram.logging import get_logger
28from lexigram.result import Err, Ok
30if TYPE_CHECKING:
31 from lexigram.contracts.core.di import ContainerResolverProtocol
33logger = get_logger(__name__)
35_WIDGETS: tuple[DashboardWidgetDefinition, ...] = (
36 DashboardWidgetDefinition(
37 name="active_sessions",
38 title="Active Sessions",
39 contributor="auth",
40 render_endpoint="/admin/auth/widgets/active_sessions",
41 size=WidgetSize.LARGE,
42 category=WidgetCategory.ACTIVITY,
43 view_kind=WidgetKind.STAT,
44 description="Number of currently active authenticated sessions.",
45 ),
46 DashboardWidgetDefinition(
47 name="token_refresh_rate",
48 title="Token Refresh Rate",
49 contributor="auth",
50 render_endpoint="/admin/auth/widgets/token_refresh_rate",
51 size=WidgetSize.SMALL,
52 category=WidgetCategory.METRICS,
53 view_kind=WidgetKind.STAT,
54 description="JWT / OAuth2 token refresh requests per minute.",
55 ),
56 DashboardWidgetDefinition(
57 name="failed_logins",
58 title="Failed Logins",
59 contributor="auth",
60 render_endpoint="/admin/auth/widgets/failed_logins",
61 size=WidgetSize.SMALL,
62 category=WidgetCategory.HEALTH,
63 view_kind=WidgetKind.STAT,
64 description="Count of failed login attempts over the last hour.",
65 ),
66)
68_NAV_ITEMS: tuple[NavigationContribution, ...] = (
69 NavigationContribution(
70 label="Auth",
71 url="/admin/auth",
72 icon="lock-closed",
73 group="security",
74 order=25,
75 children=(
76 NavigationContribution(
77 label="Users",
78 url="/admin/auth/users",
79 icon="users",
80 group="security",
81 order=10,
82 ),
83 NavigationContribution(
84 label="Sessions",
85 url="/admin/auth/sessions",
86 icon="activity",
87 group="security",
88 order=20,
89 ),
90 NavigationContribution(
91 label="Tokens",
92 url="/admin/auth/tokens",
93 icon="key",
94 group="security",
95 order=30,
96 ),
97 ),
98 ),
99)
101_HEALTH_DEFS: tuple[AdminHealthDefinition, ...] = (
102 AdminHealthDefinition(
103 name="auth.token_store",
104 contributor="auth",
105 component="Token Store",
106 check_endpoint="/admin/auth/health/token_store",
107 description="Verifies the token storage backend is available.",
108 ),
109)
111_ACTIONS: tuple[AdminActionDefinition, ...] = (
112 AdminActionDefinition(
113 name="revoke_all_sessions",
114 title="Revoke All Sessions",
115 contributor="auth",
116 handler="lexigram.auth.admin.actions:revoke_all_sessions",
117 icon="shield-off",
118 confirmation_message="This will immediately log out all active users. Are you sure?",
119 destructive=True,
120 category="security",
121 ),
122)
125class AuthAdminContributor(BaseAdminContributor):
126 """Admin contributor for the lexigram-auth package.
128 Provides 3 widgets: active sessions, token refresh rate, and failed logins.
129 Each widget is handled by a dedicated handler class that returns
130 structured WidgetContent directly. Dependencies are resolved from the
131 container in ``on_admin_boot``.
132 """
134 name = "auth"
135 display_name = "Auth"
136 group = "security"
137 icon = "lock-closed"
138 priority = 25
140 def __init__(self) -> None:
141 """Initialize the contributor with no arguments.
143 All DI-dependent attributes are resolved in ``on_admin_boot``.
144 """
145 self._handlers: dict[str, WidgetHandlerProtocol] = {}
147 async def on_admin_boot(self, container: ContainerResolverProtocol) -> None:
148 """Resolve auth DI dependencies from the container.
150 Args:
151 container: The DI container resolver.
152 """
153 from lexigram.auth.admin.handlers.active_sessions import (
154 ActiveSessionsWidgetHandler,
155 )
156 from lexigram.auth.admin.handlers.failed_logins import (
157 FailedLoginsWidgetHandler,
158 )
159 from lexigram.auth.admin.handlers.token_refresh_rate import (
160 TokenRefreshRateWidgetHandler,
161 )
163 try:
164 self._handlers = {
165 "active_sessions": await container.resolve(ActiveSessionsWidgetHandler),
166 "token_refresh_rate": await container.resolve(
167 TokenRefreshRateWidgetHandler
168 ),
169 "failed_logins": await container.resolve(FailedLoginsWidgetHandler),
170 }
171 except Exception as exc: # noqa: BLE001
172 logger.warning("auth_contributor.handlers_unavailable", error=str(exc))
174 def get_dashboard_widgets(self) -> Sequence[DashboardWidgetDefinition]:
175 """Return the dashboard widget definitions for this contributor."""
176 return list(_WIDGETS)
178 def get_navigation_items(self) -> Sequence[NavigationContribution]:
179 """Return the navigation items for this contributor."""
180 return list(_NAV_ITEMS)
182 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]:
183 """Return the health check definitions for this contributor."""
184 return list(_HEALTH_DEFS)
186 def get_actions(self) -> Sequence[AdminActionDefinition]:
187 """Return the action definitions for this contributor."""
188 return list(_ACTIONS)
190 def get_management_pages(self) -> Sequence[ManagementPageDefinition]:
191 """Return the management page definitions for this contributor."""
192 return [
193 ManagementPageDefinition(
194 name="auth_overview",
195 title="Auth Overview",
196 contributor=self.name,
197 route_path="/auth",
198 handler="lexigram.auth.admin.pages.overview:AuthOverviewPage",
199 icon="lock-closed",
200 category=PageCategory.SECURITY,
201 ),
202 ManagementPageDefinition(
203 name="auth_users",
204 title="Auth Users",
205 contributor=self.name,
206 route_path="/auth/users",
207 handler="lexigram.auth.admin.pages.users:AuthUsersPage",
208 icon="users",
209 category=PageCategory.SECURITY,
210 ),
211 ManagementPageDefinition(
212 name="auth_sessions",
213 title="Auth Sessions",
214 contributor=self.name,
215 route_path="/auth/sessions",
216 handler="lexigram.auth.admin.pages.sessions:AuthSessionsPage",
217 icon="activity",
218 category=PageCategory.SECURITY,
219 ),
220 ManagementPageDefinition(
221 name="auth_tokens",
222 title="Auth Tokens",
223 contributor=self.name,
224 route_path="/auth/tokens",
225 handler="lexigram.auth.admin.pages.tokens:AuthTokensPage",
226 icon="key",
227 category=PageCategory.SECURITY,
228 ),
229 ]
231 async def render_widget(
232 self,
233 widget_name: str,
234 params: WidgetParams,
235 resolver: ContainerResolverProtocol | None = None,
236 ) -> Result[WidgetViewModel, AdminError]:
237 """Render a widget by name using registry dispatch.
239 Registry dispatch — no if/elif. Infrastructure exceptions propagate.
241 Args:
242 widget_name: Name of the widget to render.
243 params: Widget parameters.
245 Returns:
246 Result containing a WidgetViewModel with structured content,
247 or WidgetNotFoundError if the widget is not registered.
248 """
249 handler: Any = self._handlers.get(widget_name)
250 if handler is None:
251 return Err(cast("AdminError", WidgetNotFoundError(self.name, widget_name)))
253 result = await handler.get_data(params)
254 if result.is_err():
255 return Err(result.unwrap_err())
256 return Ok(WidgetViewModel(content=result.unwrap()))
259__all__ = ["AuthAdminContributor"]