Coverage for src/lexigram/admin/di/mount/contributors.py: 0%
143 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Mount phases for contributors, routing, SSE and app-state exposure."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.logging import get_logger
9if TYPE_CHECKING:
10 from lexigram.admin.di.bundle_provider import AdminProvider
11 from lexigram.admin.di.mount.context import MountContext
13_log = get_logger(__name__)
16class AdminMountContributorsMixin:
17 """Mount phases that wire contributors, the router and app state."""
19 # Host attributes provided by AdminProvider.
20 _config: Any
21 _resources: list[Any]
22 _controllers: list[Any]
23 _mount_failures: dict[str, str]
24 _authorizer_service: Any
25 _get_csrf_service: Any
28 async def _mount_contributors(
29 self, resolver: Any, ctx: MountContext
30 ) -> None:
31 """Discover contributor resources and wire their data sources.
33 Contributor discovery is best-effort; failures are recorded in
34 ``_mount_failures`` without aborting the mount.
36 Args:
37 resolver: The DI resolver for contributor resolution.
38 ctx: Mount pipeline state (resources/contributors populated).
39 """
40 from lexigram.admin.contributors.registry import ContributorRegistry
41 from lexigram.admin.contributors.resource_collector import ResourceCollector
42 from lexigram.admin.dashboard.naming_policy import NamingPolicy
44 contributors: list = []
45 contributor_registry: Any | None = None
46 try:
47 contributor_registry = await resolver.resolve(
48 ContributorRegistry,
49 bypass_visibility=True,
50 )
51 if contributor_registry is None:
52 raise ValueError("Contributor registry unavailable")
53 contributors = list(contributor_registry.get_all())
54 except Exception as exc:
55 _log.warning("admin.contributors_discovery_failed", exc_info=True)
56 self._mount_failures["contributor_discovery"] = str(exc)
58 ctx.contributors = contributors
59 ctx.contributor_registry = contributor_registry
61 try:
62 naming = NamingPolicy(mode=self._config.contributor_collision_mode)
63 collector = ResourceCollector(naming_policy=naming)
64 contributor_resources = collector.collect(contributors)
65 for resource_cls in contributor_resources:
66 name = (
67 getattr(resource_cls, "name", None)
68 or resource_cls.__name__.replace("Resource", "").lower()
69 )
70 try:
71 ctx.resources[name] = await resolver.resolve(
72 resource_cls,
73 bypass_visibility=True,
74 )
75 except Exception: # noqa: BLE001 — fall back to direct
76 try:
77 ctx.resources[name] = resource_cls()
78 except Exception as inner: # noqa: BLE001 — skip, log warning
79 _log.warning(
80 "admin.contributor_resource_resolution_failed",
81 resource=resource_cls.__name__,
82 contributor=name,
83 error=str(inner),
84 )
85 self._mount_failures[f"contributor_resource:{name}"] = str(
86 inner
87 )
89 # Wire data sources to resolved resources that declare _data_source_class
90 user_permission_inventory: Any = None
91 user_resource_cls: Any = None
92 try:
93 from lexigram.admin.rbac.inventory import PermissionInventoryService
94 from lexigram.admin.resources.users import UserResource
96 user_permission_inventory = await resolver.resolve(
97 PermissionInventoryService,
98 bypass_visibility=True,
99 )
100 user_resource_cls = UserResource
101 except Exception: # noqa: BLE001 — best-effort wiring
102 _log.debug("admin.user_permission_inventory_unavailable")
104 for name, resource in list(ctx.resources.items()):
105 if (
106 user_permission_inventory is not None
107 and user_resource_cls is not None
108 and isinstance(resource, user_resource_cls)
109 ):
110 try:
111 resource.permission_inventory = user_permission_inventory
112 _log.debug(
113 "admin.user_permission_inventory_wired", resource=name
114 )
115 except Exception: # noqa: BLE001 — best-effort wiring
116 _log.debug(
117 "admin.user_permission_inventory_wiring_failed",
118 resource=name,
119 )
120 dsc = getattr(type(resource), "_data_source_class", None)
121 if dsc is not None and hasattr(resource, "set_data_source"):
122 try:
123 ds = await resolver.resolve(dsc, bypass_visibility=True)
124 resource.set_data_source(ds)
125 _log.debug("admin.data_source_wired", resource=name)
126 except Exception:
127 _log.debug("admin.data_source_wiring_failed", resource=name)
129 # Wrap data source with search wrappers when the resource
130 # has a searchable spec and the search engine is available.
131 search_spec = resource.search_spec()
132 if search_spec and search_spec.index_name:
133 try:
134 from lexigram.contracts.search import SearchEngineProtocol
136 search_engine = await resolver.resolve(
137 SearchEngineProtocol,
138 bypass_visibility=True,
139 )
141 from lexigram.admin.integrations.search_query import (
142 SearchQueryDataSourceWrapper,
143 )
144 from lexigram.admin.integrations.search_sync import (
145 SearchSyncDataSourceWrapper,
146 )
148 fallback_to_like = getattr(
149 self._config.integrations.search,
150 "fallback_to_like",
151 True,
152 )
153 query_wrapped = SearchQueryDataSourceWrapper(
154 ds,
155 search_engine,
156 search_spec.index_name,
157 fallback_to_like=fallback_to_like,
158 )
159 wrapped = SearchSyncDataSourceWrapper(
160 query_wrapped, search_engine, search_spec
161 )
162 resource.set_data_source(wrapped)
163 _log.debug("admin.search_wired", resource=name)
164 except Exception:
165 _log.debug("admin.search_wiring_failed", resource=name)
167 except Exception: # noqa: BLE001 — resource collection is non-fatal
168 _log.warning("admin.contributors_resource_collection_failed", exc_info=True)
170 async def _mount_integration(
171 self, container: Any, ctx: MountContext
172 ) -> None:
173 """Build the admin router and integrate contributor routes.
175 Args:
176 container: The root DI resolver for route integration.
177 ctx: Mount pipeline state (``router`` populated).
178 """
179 from lexigram.admin.core.routing import AdminRouter
180 from lexigram.admin.dashboard.naming_policy import NamingPolicy
181 from lexigram.admin.dashboard.route_integrator import RouteIntegrator
183 router = AdminRouter(
184 config=self._config,
185 resources=ctx.resources,
186 controllers=ctx.controllers,
187 middleware_stack=ctx.middlewares,
188 authorizer=self._authorizer_service,
189 )
190 ctx.router = router
192 try:
193 naming = NamingPolicy(mode=self._config.contributor_collision_mode)
194 integrator = RouteIntegrator(
195 router=router,
196 naming_policy=naming,
197 route_prefix=self._config.prefix,
198 container=container,
199 )
200 integrator.register(ctx.contributors)
201 except Exception as exc: # noqa: BLE001 — route integration is non-fatal
202 _log.warning("admin.contributors_route_integration_failed", exc_info=True)
203 self._mount_failures["route_integrator"] = str(exc)
205 async def _mount_sse_widgets(
206 self, container: Any, ctx: MountContext
207 ) -> None:
208 """Register the SSE endpoint for live widget delivery.
210 Args:
211 container: The root DI resolver for hub/permission services.
212 ctx: Mount pipeline state (``router`` read).
213 """
214 router = ctx.router
215 if router is None:
216 return
217 try:
218 from lexigram.admin.dashboard.widget_stream import (
219 build_widget_event_stream_handler,
220 )
221 from lexigram.admin.rbac.service import PermissionService
222 from lexigram.admin.realtime.subject_hub import SubjectAdminEventHub
223 from lexigram.contracts.web.sse import ReactiveSseBridgeProtocol
225 widget_hub: SubjectAdminEventHub = await container.resolve(
226 SubjectAdminEventHub
227 )
228 permission_service: PermissionService = await container.resolve(
229 PermissionService
230 )
231 sse_bridge = await container.resolve(ReactiveSseBridgeProtocol)
233 router.add_route(
234 "/_sse/widgets",
235 "GET",
236 build_widget_event_stream_handler(
237 widget_hub, permission_service, sse_bridge=sse_bridge
238 ),
239 "admin_sse_widgets",
240 )
241 _log.info("admin.sse_widgets_route_registered", path="/admin/_sse/widgets")
242 except Exception as exc: # noqa: BLE001 — SSE is optional
243 _log.warning("admin.sse_widgets_route_skipped", reason=str(exc))
245 async def _mount_app_state(
246 self, app: Any, ctx: MountContext
247 ) -> None:
248 """Mount the router and expose nav/registry state on both apps.
250 The renderer looks up request.app.state.nav_builder; request.app is
251 the *inner* admin sub-app (not the outer Starlette app), so state is
252 set on both.
254 Args:
255 app: The outer Starlette application to mount the panel on.
256 ctx: Mount pipeline state (nav/registry state read).
257 """
258 router = ctx.router
259 if router is None:
260 return
261 admin_app = router.mount(app)
262 ctx.admin_app = admin_app
264 # Expose nav_builder on app state so AdminRenderer can build the sidebar.
265 if hasattr(app, "state"):
266 app.state.nav_builder = ctx.nav_builder
267 if admin_app is not None and hasattr(admin_app, "state"):
268 admin_app.state.nav_builder = ctx.nav_builder
270 # Build NavigationAssembler contributions and expose on app state.
271 assembler_nav_items: list[dict[str, object]] = []
272 assembler_groups: dict[str, list[Any]] | None = None
273 registry = ctx.contributor_registry
274 if registry is not None and ctx.contributors:
275 from lexigram.admin.navigation.assembler import (
276 NavigationAssembler,
277 contributions_to_flat_nav,
278 )
280 try:
281 assembler = NavigationAssembler(
282 contributor_registry=registry,
283 resource_items=[],
284 )
285 grouped = await assembler.build()
286 assembler_groups = grouped
287 assembler_nav_items = contributions_to_flat_nav(grouped)
288 except Exception: # noqa: BLE001 — non-fatal
289 _log.warning("admin.navigation_assembler_prebuild_failed")
290 if hasattr(app, "state"):
291 app.state.assembler_nav_items = assembler_nav_items
292 app.state.assembler_groups = assembler_groups or {}
293 if admin_app is not None and hasattr(admin_app, "state"):
294 admin_app.state.assembler_nav_items = assembler_nav_items
295 admin_app.state.assembler_groups = assembler_groups or {}
297 # Expose the cluster registry on app state so nav resolution and
298 # cluster centers resolve the active cluster per request.
299 if ctx.cluster_registry is not None:
300 if hasattr(app, "state"):
301 app.state.cluster_registry = ctx.cluster_registry
302 if admin_app is not None and hasattr(admin_app, "state"):
303 admin_app.state.cluster_registry = ctx.cluster_registry