Coverage for src/lexigram/admin/di/sub_providers/contributor.py: 90%
136 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Admin contributor sub-provider — discovers and manages admin contributors."""
3from __future__ import annotations
5import importlib.metadata
6from typing import TYPE_CHECKING, Any
8from lexigram.admin.contributors.core import CoreAdminContributor
9from lexigram.admin.contributors.registry import ContributorRegistry
10from lexigram.admin.contributors.resource_collector import ResourceCollector
11from lexigram.admin.dashboard.naming_policy import NamingPolicy
12from lexigram.contracts.admin.dependencies import sort_contributors
13from lexigram.contracts.admin.protocols import AdminContributorRegistryProtocol
14from lexigram.contracts.core.health import HealthCheckResult, HealthStatus
15from lexigram.logging import get_logger
17if TYPE_CHECKING:
18 from lexigram.admin.config import AdminConfig
19 from lexigram.contracts.admin.protocols import AdminContributorProtocol
20 from lexigram.contracts.core.di import (
21 ContainerRegistrarProtocol,
22 ContainerResolverProtocol,
23 )
25logger = get_logger(__name__)
27ENTRY_POINT_GROUP = "lexigram.admin.contributors"
30def _validate_widget_endpoints(
31 contributors: list[Any],
32 known_paths: set[str],
33) -> None:
34 """Log a warning for each widget whose render_endpoint has no matching route."""
35 for c in contributors:
36 try:
37 widgets = c.get_dashboard_widgets()
38 except Exception: # noqa: BLE001
39 logger.warning(
40 "admin.widget_validation_get_widgets_failed",
41 contributor=getattr(c, "name", repr(c)),
42 exc_info=True,
43 )
44 continue
45 for widget in widgets:
46 ep = getattr(widget, "render_endpoint", None)
47 if ep and ep not in known_paths:
48 logger.warning(
49 "admin.widget_endpoint_not_registered",
50 contributor=getattr(c, "name", repr(c)),
51 widget=getattr(widget, "name", repr(widget)),
52 render_endpoint=ep,
53 )
56class AdminContributorSubProvider:
57 """Discovers and manages admin contributors via entry points.
59 Discovery order:
60 1. Built-in CoreAdminContributor
61 2. Entry points from ``lexigram.admin.contributors`` group
62 3. Boot all enabled contributors
64 When constructed with ``contributors``, the sub-provider operates in
65 *direct mode*: ``boot_all()`` iterates those contributors, tracks any
66 failures in ``_boot_failures``, and ``health_check()`` reflects that state.
67 """
69 def __init__(
70 self,
71 config: AdminConfig | None = None,
72 contributors: list[AdminContributorProtocol] | None = None,
73 ) -> None:
74 self._config = config
75 self._registry = ContributorRegistry()
76 if config is not None:
77 self._registry.register(CoreAdminContributor())
78 self._direct_contributors: list[AdminContributorProtocol] = [
79 c() if isinstance(c, type) else c for c in (contributors or [])
80 ]
81 self._boot_failures: dict[str, str] = {}
82 self._entry_point_failures: dict[str, str] = {}
83 self._collected_resources: list[type] = []
84 self._collected_routes: list[Any] = []
85 # Register direct contributors in the registry before entry-point
86 # discovery so that entry points with the same name are skipped.
87 for dc in self._direct_contributors:
88 self._registry.register(dc)
89 # Discover entry-point contributors here so that sub-providers
90 # consuming the registry (e.g. AdminDashboardSubProvider) see
91 # the full set during their own register() phase.
92 self._discover_from_entry_points()
94 @property
95 def config(self) -> AdminConfig | None:
96 """Return current admin config."""
97 return self._config
99 @property
100 def registry(self) -> ContributorRegistry:
101 """Return the contributor registry."""
102 return self._registry
104 @property
105 def collected_resources(self) -> list[type]:
106 """Return Resource classes collected from all contributors."""
107 return list(self._collected_resources)
109 @property
110 def collected_routes(self) -> list[Any]:
111 """Return AdminRouteSpec instances collected from all contributors at boot."""
112 return list(self._collected_routes)
114 async def register(self, container: ContainerRegistrarProtocol) -> None:
115 """Register the contributor registry in the DI container."""
116 container.singleton(AdminContributorRegistryProtocol, self._registry)
117 container.singleton(ContributorRegistry, self._registry)
119 async def boot(self, container: ContainerResolverProtocol) -> None:
120 """Boot all discovered contributors, sorted by dependency order."""
121 contributors = sort_contributors(self._registry.get_all()) # type: ignore[type-var]
122 for contributor in contributors:
123 if self._is_enabled(contributor.name):
124 try:
125 await contributor.on_admin_boot(container)
126 except Exception as exc: # noqa: BLE001 — continue booting other contributors
127 logger.warning(
128 "admin.contributor_on_boot_failed",
129 contributor=contributor.name,
130 error=str(exc),
131 exc_info=True,
132 )
133 self._boot_failures[contributor.name] = str(exc)
135 async def boot_all(self) -> None:
136 """Boot directly-supplied contributors, tracking failures.
138 Iterates every contributor passed to the constructor, calls
139 ``await contributor.boot()``, and records the ``contributor_id``
140 of any contributor whose boot raises. Remaining contributors are
141 always attempted — failures do not abort the loop.
143 After booting, collects Resource classes from all contributors
144 via ``ResourceCollector`` and stores them in ``_collected_resources``.
145 """
146 for contributor in self._direct_contributors:
147 try:
148 await contributor.on_admin_boot(None) # type: ignore[arg-type]
149 except Exception as exc: # noqa: BLE001 — track failure and continue
150 logger.error(
151 "admin.contributor_boot_failed",
152 contributor_id=contributor.contributor_id,
153 error=str(exc),
154 )
155 self._boot_failures[contributor.contributor_id] = str(exc)
157 # Collect resources from all contributors (both direct and entry-point)
158 try:
159 all_contributors: list[AdminContributorProtocol] = list(
160 self._registry.get_all()
161 )
162 all_contributors.extend(self._direct_contributors)
164 naming = NamingPolicy(mode="warn")
165 collector = ResourceCollector(naming_policy=naming)
166 self._collected_resources = collector.collect(all_contributors)
167 except Exception: # noqa: BLE001 — resource collection is non-fatal
168 logger.warning(
169 "admin.contributor_resource_collection_failed", exc_info=True
170 )
172 # Collect routes from all contributors
173 try:
174 all_c: list[AdminContributorProtocol] = list(self._registry.get_all())
175 all_c.extend(self._direct_contributors)
176 for c in all_c:
177 try:
178 for spec in c.get_routes():
179 self._collected_routes.append(spec)
180 except Exception: # noqa: BLE001 — skip broken contributors
181 logger.warning(
182 "admin.contributor_get_routes_failed",
183 contributor=getattr(c, "name", repr(c)),
184 exc_info=True,
185 )
186 except Exception: # noqa: BLE001 — route collection is non-fatal
187 logger.warning("admin.contributor_route_collection_failed", exc_info=True)
189 # Validate widget render_endpoint against collected routes
190 try:
191 known_paths = {spec.path for spec in self._collected_routes}
192 all_contributors_for_validation: list = list(self._registry.get_all())
193 all_contributors_for_validation.extend(self._direct_contributors)
194 _validate_widget_endpoints(all_contributors_for_validation, known_paths)
195 except Exception: # noqa: BLE001 — validation is non-fatal
196 logger.warning("admin.widget_endpoint_validation_failed", exc_info=True)
198 async def shutdown(self) -> None:
199 """Shut down all contributors in reverse priority order."""
200 for contributor in reversed(list(self._registry.get_all())):
201 try:
202 await contributor.on_admin_shutdown()
203 except Exception: # noqa: BLE001 — continue shutting down other contributors
204 logger.warning(
205 "admin.contributor_shutdown_failed",
206 contributor=contributor.name,
207 exc_info=True,
208 )
210 def health_check(self, timeout: float = 5.0) -> HealthCheckResult: # noqa: ARG002
211 """Return contributor health, reflecting any boot/entry-point failures.
213 Args:
214 timeout: Accepted for interface compatibility; unused by this
215 synchronous implementation.
217 Returns:
218 ``DEGRADED`` with a failure summary when one or more contributors
219 failed to boot; ``HEALTHY`` otherwise.
220 """
221 details: dict[str, object] = {}
222 if self._entry_point_failures:
223 details["entry_point_failures"] = dict(self._entry_point_failures)
224 if self._boot_failures:
225 details["boot_failures"] = dict(self._boot_failures)
226 if self._entry_point_failures or self._boot_failures:
227 return HealthCheckResult(
228 component="admin_contributors",
229 status=HealthStatus.DEGRADED,
230 message=f"Boot failures: {self._boot_failures}, entry point failures: {self._entry_point_failures}",
231 details=details,
232 )
233 count = len(list(self._registry.get_all())) + len(self._direct_contributors)
234 return HealthCheckResult(
235 component="admin_contributors",
236 status=HealthStatus.HEALTHY,
237 message=f"{count} contributor(s) registered",
238 details={"count": count},
239 )
241 def _discover_from_entry_points(self) -> None:
242 """Load contributors from the entry point group.
244 Skips any contributor class that is already registered
245 in the registry (i.e., was set up by a provider's factory).
246 Only performs direct instantiation for zero-argument contributors
247 that have no provider-based registration.
248 """
249 for ep in importlib.metadata.entry_points(group=ENTRY_POINT_GROUP):
250 try:
251 contributor_cls = ep.load()
252 contributor = contributor_cls()
253 # Skip if this contributor is already registered by a provider.
254 # The provider's factory already constructed and registered it.
255 if self._registry.get(contributor.name) is not None:
256 logger.debug(
257 "admin.contributor_already_registered",
258 name=contributor.name,
259 )
260 continue
261 if self._is_enabled(contributor.name):
262 self._registry.register(contributor)
263 logger.info("contributor_discovered", name=contributor.name)
264 except Exception as exc: # noqa: BLE001 — skip bad entry points; continue discovery
265 logger.warning(
266 "admin.entry_point_load_failed",
267 name=ep.name,
268 error=str(exc),
269 exc_info=True,
270 )
271 self._entry_point_failures[ep.name] = str(exc)
273 def _is_enabled(self, name: str) -> bool:
274 """Check if a contributor is enabled via config.
276 Always returns ``True`` when no config was supplied (direct mode).
277 """
278 if self._config is None:
279 return True
280 contributors_config = getattr(self._config, "contributors", {})
281 if isinstance(contributors_config, dict) and name in contributors_config:
282 contrib_cfg = contributors_config[name]
283 if isinstance(contrib_cfg, dict):
284 return contrib_cfg.get("enabled", True)
285 return getattr(contrib_cfg, "enabled", True)
286 return True
289__all__ = ["AdminContributorSubProvider"]