Coverage for src/lexigram/web/admin/contributor.py: 58%

1"""Admin contributor for lexigram-web — surfaces web server widgets, navigation, 

2health checks, and actions into the Lexigram admin dashboard. 

3""" 

4 

5from __future__ import annotations 

6 

7from collections.abc import Sequence 

8from typing import TYPE_CHECKING, cast 

9 

10from lexigram.contracts.admin.contributor import BaseAdminContributor 

11from lexigram.contracts.admin.errors import AdminError, WidgetNotFoundError 

12from lexigram.contracts.admin.types import ( 

13 AdminActionDefinition, 

14 AdminHealthDefinition, 

15 AdminRouteSpec, 

16 DashboardWidgetDefinition, 

17 ManagementPageDefinition, 

18 NavigationContribution, 

19 PageCategory, 

20 WidgetCategory, 

21 WidgetKind, 

22 WidgetParams, 

23 WidgetSize, 

24 WidgetViewModel, 

25) 

26from lexigram.logging import get_logger 

27from lexigram.result import Err, Ok, Result 

28 

29if TYPE_CHECKING: 

30 from lexigram.contracts.admin.widget_protocols import WidgetHandlerProtocol 

31 from lexigram.contracts.core.di import ContainerResolverProtocol 

32 

33 

34async def _empty_handler(request: object) -> str: # noqa: ARG001 

35 """Placeholder handler for contributed routes.""" 

36 return "" 

37 

38 

39logger = get_logger(__name__) 

40 

41_WIDGETS: tuple[DashboardWidgetDefinition, ...] = ( 

42 DashboardWidgetDefinition( 

43 name="server_status", 

44 title="Server Status", 

45 contributor="web", 

46 render_endpoint="/admin/web/widgets/server_status", 

47 size=WidgetSize.SMALL, 

48 category=WidgetCategory.HEALTH, 

49 view_kind=WidgetKind.HEALTH, 

50 description="Current HTTP server status and uptime.", 

51 ), 

52 DashboardWidgetDefinition( 

53 name="active_connections", 

54 title="Active Connections", 

55 contributor="web", 

56 render_endpoint="/admin/web/widgets/active_connections", 

57 size=WidgetSize.SMALL, 

58 category=WidgetCategory.METRICS, 

59 view_kind=WidgetKind.STAT, 

60 description="Number of currently open HTTP connections.", 

61 ), 

62 DashboardWidgetDefinition( 

63 name="request_rate", 

64 title="Request Rate", 

65 contributor="web", 

66 render_endpoint="/admin/web/widgets/request_rate", 

67 size=WidgetSize.SMALL, 

68 category=WidgetCategory.METRICS, 

69 view_kind=WidgetKind.STAT, 

70 description="HTTP request rate over the last minute.", 

71 ), 

72) 

73 

74_NAV_ITEMS: tuple[NavigationContribution, ...] = ( 

75 NavigationContribution( 

76 label="Web", 

77 url="/admin/web", 

78 icon="globe", 

79 group="infrastructure", 

80 order=10, 

81 children=( 

82 NavigationContribution( 

83 label="Routes", 

84 url="/admin/web/routes", 

85 icon="map", 

86 group="infrastructure", 

87 order=10, 

88 ), 

89 NavigationContribution( 

90 label="Middleware", 

91 url="/admin/web/middleware", 

92 icon="layers", 

93 group="infrastructure", 

94 order=20, 

95 ), 

96 ), 

97 ), 

98) 

99 

100_HEALTH_DEFS: tuple[AdminHealthDefinition, ...] = ( 

101 AdminHealthDefinition( 

102 name="web.server", 

103 contributor="web", 

104 component="HTTP Server", 

105 check_endpoint="/admin/web/health/server", 

106 description="Verifies the ASGI server is accepting connections.", 

107 ), 

108) 

109 

110_ACTIONS: tuple[AdminActionDefinition, ...] = ( 

111 AdminActionDefinition( 

112 name="reload_routes", 

113 title="Reload Routes", 

114 contributor="web", 

115 handler="lexigram.web.admin.actions:reload_routes", 

116 icon="refresh-cw", 

117 category="operations", 

118 ), 

119) 

120 

121 

122class WebAdminContributor(BaseAdminContributor): 

123 """Admin contributor for the lexigram-web package. 

124 

125 Provides widget rendering with registry-based dispatch to handlers. 

126 Each widget is handled by a dedicated handler class that returns 

127 structured WidgetContent directly. 

128 Dependencies are resolved from the container in ``on_admin_boot``. 

129 """ 

130 

131 name = "web" 

132 display_name = "Web Server" 

133 group = "infrastructure" 

134 icon = "globe" 

135 priority = 10 

136 

137 def __init__(self) -> None: 

138 self._handlers: dict[str, WidgetHandlerProtocol] = {} 

139 

140 async def on_admin_boot(self, container: ContainerResolverProtocol) -> None: 

141 """Resolve web admin dependencies from the DI container. 

142 

143 Args: 

144 container: The DI container resolver. 

145 """ 

146 from lexigram.web.admin.handlers.active_connections import ( 

147 ActiveConnectionsWidgetHandler, 

148 ) 

149 from lexigram.web.admin.handlers.request_rate import ( 

150 RequestRateWidgetHandler, 

151 ) 

152 from lexigram.web.admin.handlers.server_status import ( 

153 ServerStatusWidgetHandler, 

154 ) 

155 

156 try: 

157 self._handlers = { 

158 "server_status": await container.resolve(ServerStatusWidgetHandler), 

159 "active_connections": await container.resolve( 

160 ActiveConnectionsWidgetHandler 

161 ), 

162 "request_rate": await container.resolve(RequestRateWidgetHandler), 

163 } 

164 except Exception as exc: # noqa: BLE001 

165 logger.warning("web_contributor.handlers_unavailable", error=str(exc)) 

166 

167 def get_routes(self) -> Sequence[AdminRouteSpec]: 

168 return [ 

169 AdminRouteSpec( 

170 path=w.render_endpoint, 

171 method="GET", 

172 handler=_empty_handler, 

173 name=f"widgets.{w.name}", 

174 ) 

175 for w in _WIDGETS 

176 ] 

177 

178 def get_dashboard_widgets(self) -> Sequence[DashboardWidgetDefinition]: 

179 return list(_WIDGETS) 

180 

181 def get_navigation_items(self) -> Sequence[NavigationContribution]: 

182 return list(_NAV_ITEMS) 

183 

184 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]: 

185 return list(_HEALTH_DEFS) 

186 

187 def get_actions(self) -> Sequence[AdminActionDefinition]: 

188 return list(_ACTIONS) 

189 

190 def get_management_pages(self) -> Sequence[ManagementPageDefinition]: 

191 return [ 

192 ManagementPageDefinition( 

193 name="web_overview", 

194 title="Web Overview", 

195 contributor="web", 

196 route_path="/web", 

197 handler="lexigram.web.admin.pages.overview:WebOverviewPage", 

198 category=PageCategory.INFRASTRUCTURE, 

199 icon="globe", 

200 description="Web server overview and metrics", 

201 order=10, 

202 ), 

203 ManagementPageDefinition( 

204 name="web_routes", 

205 title="Web Routes", 

206 contributor="web", 

207 route_path="/web/routes", 

208 handler="lexigram.web.admin.pages.routes:WebRoutesPage", 

209 category=PageCategory.INFRASTRUCTURE, 

210 icon="map", 

211 description="View registered routes", 

212 order=20, 

213 ), 

214 ManagementPageDefinition( 

215 name="web_middleware", 

216 title="Web Middleware", 

217 contributor="web", 

218 route_path="/web/middleware", 

219 handler="lexigram.web.admin.pages.middleware:WebMiddlewarePage", 

220 category=PageCategory.INFRASTRUCTURE, 

221 icon="layers", 

222 description="View middleware and filter chains", 

223 order=30, 

224 ), 

225 ] 

226 

227 async def render_widget( 

228 self, 

229 widget_name: str, 

230 params: WidgetParams, 

231 resolver: ContainerResolverProtocol | None = None, 

232 ) -> Result[WidgetViewModel, AdminError]: 

233 """Render a widget by name using registry dispatch. 

234 

235 Registry dispatch — no if/elif. Infrastructure exceptions propagate. 

236 

237 Args: 

238 widget_name: Name of the widget to render (e.g. "server_status"). 

239 params: Widget request parameters. 

240 

241 Returns: 

242 Result containing a WidgetViewModel with structured content, 

243 or WidgetNotFoundError if the widget is not registered. 

244 """ 

245 handler = self._handlers.get(widget_name) 

246 if handler is None: 

247 return cast( 

248 "Result[WidgetViewModel, AdminError]", 

249 Err(WidgetNotFoundError(self.name, widget_name)), 

250 ) 

251 result = await handler.get_data(params) 

252 if result.is_err(): 

253 return cast("Result[WidgetViewModel, AdminError]", result) 

254 return cast( 

255 "Result[WidgetViewModel, AdminError]", 

256 Ok(WidgetViewModel(content=result.unwrap())), 

257 ) 

258 

259 

260__all__ = ["WebAdminContributor"]