Coverage for src / lexigram / admin / di / bundle_provider.py: 65%

1"""Admin bundle provider — orchestrates focused sub-providers. 

2 

3Follows the AuthBundleProvider pattern from lexigram-auth. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Any, Self 

9 

10from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

11from lexigram.contracts.core.provider import ProviderPriority 

12from lexigram.di.provider import Provider 

13from lexigram.logging import get_logger 

14 

15_log = get_logger(__name__) 

16 

17if TYPE_CHECKING: 

18 from collections.abc import AsyncGenerator 

19 

20 from lexigram.admin.config import AdminConfig 

21 from lexigram.contracts.core.di import ( 

22 ContainerRegistrarProtocol, 

23 ContainerResolverProtocol, 

24 ) 

25 

26 

27class AdminProvider(Provider): 

28 """Orchestrates admin sub-providers for the full admin panel. 

29 

30 Sub-providers are focused helper classes (not Provider subclasses). 

31 This follows the EventsProvider/AuthBundleProvider pattern. 

32 

33 Config is accepted only in __init__ and never mutated after construction. 

34 Sub-providers are instantiated in register() — not in __init__ — so that 

35 no DI work happens before the container is ready. 

36 """ 

37 

38 name = "admin" 

39 priority = ProviderPriority.APPLICATION 

40 config_key: str | None = None 

41 # Do NOT declare config_key/config_model — admin config is always set 

42 # programmatically via AdminModule.configure(), not from YAML auto-injection. 

43 

44 def __init__( 

45 self, 

46 config: AdminConfig | None = None, 

47 auth_provider: Any | None = None, 

48 resources: list[type] | None = None, 

49 controllers: list[type] | None = None, 

50 extra_providers: list[Any] | None = None, 

51 **kwargs: Any, 

52 ) -> None: 

53 super().__init__(name="admin", priority=ProviderPriority.APPLICATION) 

54 from lexigram.admin.config import AdminConfig as AdminConfigCls 

55 

56 self._config = config or AdminConfigCls() 

57 self._auth_provider = auth_provider 

58 self._resources = resources or [] 

59 self._controllers = controllers or [] 

60 self._extra_providers: list[Any] = extra_providers or [] 

61 self._kwargs = kwargs 

62 

63 if ( 

64 self._config.auth.env in {"production", "staging"} 

65 and not self._config.strict_resource_resolution 

66 ): 

67 _log.warning( 

68 "admin.strict_resource_resolution_disabled_in_production", 

69 message="strict_resource_resolution=False in production/staging can hide missing admin routes", 

70 ) 

71 

72 # Sub-providers are populated in register() — empty until then. 

73 self._sub_providers: list[Any] = [] 

74 self._resolved_resources: dict[str, Any] = {} 

75 self._nav_item_builder: Any | None = None 

76 self._admin_resolver: Any | None = None 

77 self._mount_failures: dict[str, str] = {} 

78 

79 @property # type: ignore[misc] 

80 def config(self) -> AdminConfig: 

81 """Return current admin config.""" 

82 return self._config 

83 

84 @classmethod 

85 def from_config(cls, config: AdminConfig, **context: Any) -> Self: 

86 """Create provider from typed config.""" 

87 return cls(config=config, **context) 

88 

89 async def register(self, container: ContainerRegistrarProtocol) -> None: 

90 """Register admin and all sub-providers. 

91 

92 Sub-providers are instantiated here (not in __init__) so that no DI 

93 resolution or heavyweight initialisation happens before the container 

94 lifecycle has started. No resolution is performed in this method — 

95 only bindings are registered. 

96 """ 

97 from lexigram.admin.controllers.dashboard import DashboardController 

98 from lexigram.admin.controllers.widgets import WidgetController 

99 from lexigram.admin.di.sub_providers.auth import AdminAuthSubProvider 

100 from lexigram.admin.di.sub_providers.contributor import ( 

101 AdminContributorSubProvider, 

102 ) 

103 from lexigram.admin.di.sub_providers.core import AdminCoreSubProvider 

104 from lexigram.admin.di.sub_providers.dashboard import AdminDashboardSubProvider 

105 from lexigram.admin.di.sub_providers.integrations import ( 

106 AdminIntegrationsSubProvider, 

107 ) 

108 from lexigram.admin.di.sub_providers.realtime import AdminRealtimeSubProvider 

109 from lexigram.admin.di.sub_providers.resource import AdminResourceSubProvider 

110 from lexigram.admin.di.sub_providers.tenancy import AdminTenancySubProvider 

111 from lexigram.admin.di.sub_providers.ui import AdminUISubProvider 

112 from lexigram.admin.navigation.nav_item_builder import NavItemBuilder 

113 

114 contributor = AdminContributorSubProvider( 

115 config=self._config, 

116 contributors=self._kwargs.get("contributors", []), 

117 ) 

118 tenancy = AdminTenancySubProvider(config=self._config) 

119 # Boot order is intentional: 

120 # 1. Core binds admin primitives used by later providers. 

121 # 2. Auth binds identity/session services before resources/actions are wired. 

122 # 3. Resource/UI/realtime bind user-facing admin surfaces. 

123 # 4. Tenancy decorates data access before dashboard aggregation. 

124 # 5. Dashboard consumes contributor registry state. 

125 # 6. Contributor boots last because extension packages may depend on every earlier seat. 

126 sub_providers: list[Any] = [ 

127 AdminCoreSubProvider(config=self._config, **self._kwargs), 

128 AdminAuthSubProvider( 

129 config=self._config, auth_provider=self._auth_provider 

130 ), 

131 AdminResourceSubProvider(config=self._config, resources=self._resources), 

132 AdminUISubProvider(config=self._config), 

133 AdminRealtimeSubProvider(config=self._config), 

134 tenancy, 

135 AdminDashboardSubProvider( 

136 config=self._config, 

137 contributor_registry=contributor.registry, 

138 ), 

139 contributor, 

140 AdminIntegrationsSubProvider(config=self._config.integrations), 

141 ] 

142 self._sub_providers = sub_providers 

143 

144 nav_item_builder = NavItemBuilder(config=self._config) 

145 self._nav_item_builder = nav_item_builder 

146 

147 container.singleton(AdminProvider, self) 

148 # Register with string key so web provider can discover without importing admin 

149 container.singleton("admin_bundle", self) 

150 # Register NavItemBuilder as a pre-built instance (config is not in container) 

151 container.singleton(NavItemBuilder, nav_item_builder) 

152 # Register built-in controllers 

153 container.singleton(WidgetController, WidgetController) 

154 container.singleton(DashboardController, DashboardController) 

155 # Register controller classes for DI resolution 

156 for controller_cls in self._controllers: 

157 try: 

158 container.singleton(controller_cls, controller_cls) 

159 except Exception: # noqa: BLE001 — re-registration is expected; continue loop 

160 _log.debug( 

161 "admin.controller_already_registered", 

162 controller=controller_cls.__name__, 

163 ) 

164 # Register resource classes so the container can inject their service dependencies 

165 for resource_cls in self._resources: 

166 try: 

167 container.singleton(resource_cls, resource_cls) 

168 except Exception: # noqa: BLE001 — re-registration is expected; continue loop 

169 _log.debug( 

170 "admin.resource_already_registered", resource=resource_cls.__name__ 

171 ) 

172 for ep in self._extra_providers: 

173 await ep.register(container) 

174 for sp in self._sub_providers: 

175 await sp.register(container) 

176 

177 # NOTE: TenantConfigProviderProtocol is not registered here — it is 

178 # constructed directly in mount_to_app() via AdminSettingsDbProvider. 

179 

180 async def mount_to_app( 

181 self, 

182 app: Any, 

183 container: ContainerResolverProtocol, 

184 ) -> None: 

185 """Build and mount the admin panel onto a Starlette application. 

186 

187 Called by the web provider during route setup, after the Starlette 

188 app is created and all providers have booted. 

189 

190 Args: 

191 app: The Starlette application to mount the admin panel on. 

192 container: The DI resolver for resolving controller dependencies. 

193 """ 

194 from lexigram.admin.core.routing import AdminRouter 

195 from lexigram.admin.navigation.nav_item_builder import NavItemBuilder 

196 

197 admin_resolver = self._admin_resolver or container 

198 

199 # Build resources dict from resource classes {name: instance} 

200 resources_dict: dict[str, Any] = {} 

201 for resource_cls in self._resources: 

202 name = ( 

203 getattr(resource_cls, "name", None) 

204 or resource_cls.__name__.replace("Resource", "").lower() 

205 ) 

206 try: 

207 resources_dict[name] = await admin_resolver.resolve( 

208 resource_cls, 

209 bypass_visibility=True, 

210 ) 

211 except Exception as exc: 

212 _log.error( 

213 "admin.resource_resolution_failed", 

214 resource=resource_cls.__name__, 

215 error=str(exc), 

216 strict=self._config.strict_resource_resolution, 

217 ) 

218 self._mount_failures[f"resource:{resource_cls.__name__}"] = str(exc) 

219 if self._config.strict_resource_resolution: 

220 raise 

221 

222 # Resolve controller instances from container (best-effort) 

223 controller_instances: list[Any] = [] 

224 

225 # Create shared settings_service for runtime theme overrides (best-effort) 

226 admin_settings_service: Any = None 

227 try: 

228 from lexigram.admin.services.settings_service import ( 

229 AdminSettingsDbProvider, 

230 AdminSettingsService, 

231 ) 

232 from lexigram.contracts.data import DatabaseProviderProtocol 

233 

234 db_provider = await admin_resolver.resolve( 

235 DatabaseProviderProtocol, 

236 bypass_visibility=True, 

237 ) 

238 config_provider = AdminSettingsDbProvider(db=db_provider) 

239 # Ensure the tenant_configs table is created eagerly at startup 

240 try: 

241 await config_provider._ensure_table() 

242 except Exception: 

243 _log.warning("admin.tenant_config_table_creation_failed") 

244 admin_settings_service = AdminSettingsService( 

245 config_provider=config_provider, 

246 ) 

247 from lexigram.admin.settings.panel.registry import ConfigRegistry 

248 from lexigram.admin.settings.store import TenantConfigStore 

249 

250 try: 

251 registry = await admin_resolver.resolve( 

252 ConfigRegistry, 

253 bypass_visibility=True, 

254 ) 

255 registry.register_store("db", TenantConfigStore(admin_settings_service)) 

256 except Exception: 

257 _log.warning("admin.settings_store_registration_failed") 

258 except Exception: 

259 try: 

260 admin_settings_service = AdminSettingsService() 

261 except Exception: 

262 pass 

263 

264 for controller_cls in self._controllers: 

265 try: 

266 instance = await admin_resolver.resolve( 

267 controller_cls, 

268 bypass_visibility=True, 

269 ) 

270 controller_instances.append(instance) 

271 if admin_settings_service is not None and hasattr( 

272 instance, "_settings_service" 

273 ): 

274 instance._settings_service = admin_settings_service 

275 except Exception as exc: 

276 _log.error( 

277 "admin.controller_resolution_failed", 

278 controller=controller_cls.__name__, 

279 error=str(exc), 

280 strict=self._config.strict_resource_resolution, 

281 ) 

282 self._mount_failures[f"controller:{controller_cls.__name__}"] = str(exc) 

283 if self._config.strict_resource_resolution: 

284 raise 

285 

286 # Resolve built-in WidgetController (best-effort) 

287 try: 

288 from lexigram.admin.auth.protocols import ( 

289 AdminAuditLogServiceProtocol, 

290 AdminCsrfServiceProtocol, 

291 ) 

292 from lexigram.admin.controllers.widgets import WidgetController 

293 

294 widget_controller = await admin_resolver.resolve( 

295 WidgetController, 

296 bypass_visibility=True, 

297 ) 

298 controller_instances.append(widget_controller) 

299 if admin_settings_service is not None and hasattr( 

300 widget_controller, "_settings_service" 

301 ): 

302 widget_controller._settings_service = admin_settings_service 

303 try: 

304 audit_service = await admin_resolver.resolve( 

305 AdminAuditLogServiceProtocol, 

306 bypass_visibility=True, 

307 ) 

308 except Exception: 

309 audit_service = None 

310 if audit_service is not None and hasattr( 

311 widget_controller, "_audit_service" 

312 ): 

313 widget_controller._audit_service = audit_service 

314 try: 

315 csrf_service = await admin_resolver.resolve( 

316 AdminCsrfServiceProtocol, 

317 bypass_visibility=True, 

318 ) 

319 except Exception: 

320 csrf_service = None 

321 if csrf_service is not None and hasattr(widget_controller, "_csrf_service"): 

322 widget_controller._csrf_service = csrf_service 

323 except Exception as exc: 

324 _log.error( 

325 "admin.widget_controller_resolution_failed", 

326 error=str(exc), 

327 strict=self._config.strict_resource_resolution, 

328 ) 

329 self._mount_failures["controller:WidgetController"] = str(exc) 

330 if self._config.strict_resource_resolution: 

331 raise 

332 

333 # Resolve built-in DashboardController (best-effort) 

334 try: 

335 from lexigram.admin.controllers.dashboard import DashboardController 

336 

337 dashboard_controller = await admin_resolver.resolve( 

338 DashboardController, 

339 bypass_visibility=True, 

340 ) 

341 controller_instances.append(dashboard_controller) 

342 if admin_settings_service is not None: 

343 dashboard_controller._settings_service = admin_settings_service 

344 except Exception as exc: 

345 _log.error( 

346 "admin.dashboard_controller_resolution_failed", 

347 error=str(exc), 

348 strict=self._config.strict_resource_resolution, 

349 ) 

350 self._mount_failures["controller:DashboardController"] = str(exc) 

351 if self._config.strict_resource_resolution: 

352 raise 

353 

354 # Resolve built-in AuthController (login, logout) 

355 try: 

356 from lexigram.admin.controllers.auth import AuthController 

357 

358 auth_controller = await admin_resolver.resolve( 

359 AuthController, 

360 bypass_visibility=True, 

361 ) 

362 controller_instances.append(auth_controller) 

363 if admin_settings_service is not None: 

364 auth_controller._settings_service = admin_settings_service 

365 except Exception as exc: 

366 _log.error( 

367 "admin.auth_controller_resolution_failed", 

368 error=str(exc), 

369 strict=self._config.strict_resource_resolution, 

370 ) 

371 self._mount_failures["controller:AuthController"] = str(exc) 

372 if self._config.strict_resource_resolution: 

373 raise 

374 

375 # Resolve built-in SetupController (first-run wizard) 

376 try: 

377 from lexigram.admin.controllers.setup import SetupController 

378 

379 setup_controller = await admin_resolver.resolve( 

380 SetupController, 

381 bypass_visibility=True, 

382 ) 

383 controller_instances.append(setup_controller) 

384 if admin_settings_service is not None: 

385 setup_controller._settings_service = admin_settings_service 

386 except Exception as exc: 

387 _log.error( 

388 "admin.setup_controller_resolution_failed", 

389 error=str(exc), 

390 strict=self._config.strict_resource_resolution, 

391 ) 

392 self._mount_failures["controller:SetupController"] = str(exc) 

393 if self._config.strict_resource_resolution: 

394 raise 

395 

396 # Mount ErrorController (styled error pages) — best-effort 

397 try: 

398 from lexigram.admin.controllers.error import ErrorController 

399 

400 error_controller = await admin_resolver.resolve( 

401 ErrorController, 

402 bypass_visibility=True, 

403 ) 

404 controller_instances.append(error_controller) 

405 except Exception as exc: 

406 _log.error( 

407 "admin.error_controller_resolution_failed", 

408 error=str(exc), 

409 strict=self._config.strict_resource_resolution, 

410 ) 

411 self._mount_failures["controller:ErrorController"] = str(exc) 

412 if self._config.strict_resource_resolution: 

413 raise 

414 

415 # Mount PoolHealthController (connection pool monitoring) — best-effort. 

416 # pool_manager/task_manager are optional: without them the endpoints 

417 # respond 503 instead of failing resolution. 

418 try: 

419 from lexigram.admin.controllers.pool_health import PoolHealthController 

420 

421 pool_health_controller = await admin_resolver.resolve( 

422 PoolHealthController, 

423 bypass_visibility=True, 

424 ) 

425 controller_instances.append(pool_health_controller) 

426 except Exception as exc: 

427 _log.error( 

428 "admin.pool_health_controller_resolution_failed", 

429 error=str(exc), 

430 strict=self._config.strict_resource_resolution, 

431 ) 

432 self._mount_failures["controller:PoolHealthController"] = str(exc) 

433 if self._config.strict_resource_resolution: 

434 raise 

435 

436 # Mount ProgressController (SSE/status progress tracking) — best-effort. 

437 # Tries an integrator-registered tracker first; falls back to the 

438 # admin-owned LocalProgressTracker (no dependency on optional 

439 # integration packages — without one the controller mounts with the 

440 # in-process tracker instead of being skipped). 

441 try: 

442 from lexigram.admin.controllers.progress import ( 

443 LocalProgressTracker, 

444 ProgressController, 

445 ) 

446 

447 try: 

448 progress_controller = await admin_resolver.resolve( 

449 ProgressController, 

450 bypass_visibility=True, 

451 ) 

452 except Exception: 

453 progress_controller = ProgressController(tracker=LocalProgressTracker()) 

454 controller_instances.append(progress_controller) 

455 except ModuleNotFoundError as exc: 

456 _log.info( 

457 "admin.progress_controller_skipped", 

458 reason="progress_controller_unavailable", 

459 error=str(exc), 

460 ) 

461 except Exception as exc: 

462 _log.error( 

463 "admin.progress_controller_resolution_failed", 

464 error=str(exc), 

465 strict=self._config.strict_resource_resolution, 

466 ) 

467 self._mount_failures["controller:ProgressController"] = str(exc) 

468 if self._config.strict_resource_resolution: 

469 raise 

470 

471 # Mount SettingsController (theme & branding settings) 

472 try: 

473 from lexigram.admin.auth.protocols import ( 

474 AdminAuditLogServiceProtocol, 

475 AdminCsrfServiceProtocol, 

476 ) 

477 from lexigram.admin.controllers.settings import SettingsController 

478 from lexigram.admin.engine.renderer import AdminRenderer 

479 from lexigram.admin.settings.panel.registry import ConfigRegistry 

480 

481 csrf_service: AdminCsrfServiceProtocol | None = None 

482 try: 

483 csrf_service = await admin_resolver.resolve( 

484 AdminCsrfServiceProtocol, 

485 bypass_visibility=True, 

486 ) 

487 except Exception: 

488 pass 

489 

490 registry: ConfigRegistry | None = None 

491 try: 

492 registry = await admin_resolver.resolve( 

493 ConfigRegistry, 

494 bypass_visibility=True, 

495 ) 

496 except Exception: 

497 pass 

498 

499 audit_service: AdminAuditLogServiceProtocol | None = None 

500 try: 

501 audit_service = await admin_resolver.resolve( 

502 AdminAuditLogServiceProtocol, 

503 bypass_visibility=True, 

504 ) 

505 except Exception: 

506 pass 

507 

508 renderer = await admin_resolver.resolve( 

509 AdminRenderer, 

510 bypass_visibility=True, 

511 ) 

512 settings_controller = SettingsController( 

513 renderer=renderer, 

514 settings_service=admin_settings_service, 

515 csrf_service=csrf_service, 

516 audit_service=audit_service, 

517 registry=registry, 

518 ) 

519 controller_instances.append(settings_controller) 

520 except Exception as exc: 

521 _log.warning( 

522 "admin.settings_controller_skipped", 

523 error=str(exc), 

524 ) 

525 

526 # Mount InfrastructureController (cluster landing page) 

527 try: 

528 from lexigram.admin.controllers.infrastructure import ( 

529 InfrastructureController, 

530 ) 

531 from lexigram.admin.engine.renderer import AdminRenderer 

532 

533 infra_renderer = await admin_resolver.resolve( 

534 AdminRenderer, 

535 bypass_visibility=True, 

536 ) 

537 controller_instances.append( 

538 InfrastructureController(renderer=infra_renderer) 

539 ) 

540 except Exception as exc: 

541 _log.warning( 

542 "admin.infrastructure_controller_skipped", 

543 error=str(exc), 

544 ) 

545 

546 # Populate NavItemBuilder with resolved resource instances 

547 self._resolved_resources = resources_dict 

548 nav_builder = self._nav_item_builder 

549 if nav_builder is None: 

550 nav_builder = NavItemBuilder(config=self._config) 

551 self._nav_item_builder = nav_builder 

552 nav_builder.set_resources(resources_dict) 

553 

554 # Resolve SetupMiddleware's dependency: the admin user store. 

555 # Best-effort — if the store is not registered (e.g. custom auth setups) 

556 # SetupMiddleware is simply not added and no setup redirect occurs. 

557 middleware_stack: list[tuple[type, dict[str, Any]]] = [] 

558 try: 

559 from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol 

560 from lexigram.admin.middleware.setup import SetupMiddleware 

561 

562 admin_user_store = await admin_resolver.resolve( 

563 AdminUserStoreProtocol, 

564 bypass_visibility=True, 

565 ) 

566 middleware_stack.append( 

567 (SetupMiddleware, {"admin_user_store": admin_user_store}) 

568 ) 

569 _log.debug("admin.setup_middleware_wired") 

570 except Exception as exc: # noqa: BLE001 — SetupMiddleware is optional 

571 _log.warning( 

572 "admin.setup_middleware_skipped", 

573 reason=str(exc), 

574 ) 

575 

576 # Wire AdminCsrfMiddleware when the CSRF service is available. 

577 try: 

578 from lexigram.admin.auth.protocols import ( 

579 AdminAuditLogServiceProtocol, 

580 AdminCsrfServiceProtocol, 

581 ) 

582 from lexigram.admin.middleware.csrf import AdminCsrfMiddleware 

583 

584 csrf_service = await admin_resolver.resolve( 

585 AdminCsrfServiceProtocol, 

586 bypass_visibility=True, 

587 ) 

588 csrf_audit_service = None 

589 try: 

590 csrf_audit_service = await admin_resolver.resolve( 

591 AdminAuditLogServiceProtocol, 

592 bypass_visibility=True, 

593 ) 

594 except Exception: # noqa: BLE001 — CSRF audit is optional 

595 pass 

596 middleware_stack.append( 

597 ( 

598 AdminCsrfMiddleware, 

599 { 

600 "csrf_service": csrf_service, 

601 "audit_service": csrf_audit_service, 

602 }, 

603 ) 

604 ) 

605 _log.debug("admin.csrf_middleware_wired") 

606 except Exception as exc: # noqa: BLE001 — CSRF middleware is optional 

607 _log.warning("admin.csrf_middleware_skipped", reason=str(exc)) 

608 

609 # Wire session-based auth guard — redirects unauthenticated requests to login. 

610 if self._config.require_auth: 

611 try: 

612 from lexigram.admin.middleware.auth_guard import ( 

613 AdminAuthGuardMiddleware, 

614 ) 

615 

616 middleware_stack.append((AdminAuthGuardMiddleware, {})) 

617 _log.debug("admin.auth_guard_middleware_wired") 

618 except Exception as exc: # noqa: BLE001 — guard middleware is optional 

619 _log.warning("admin.auth_guard_middleware_skipped", reason=str(exc)) 

620 else: 

621 _log.debug("admin.auth_guard_middleware_skipped_require_auth_unset") 

622 

623 # Wire auth middleware — loads user from session into request.state.user 

624 # so that downstream authorization middleware can enforce RBAC. 

625 try: 

626 from lexigram.admin.auth.protocols import AdminSessionServiceProtocol 

627 from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol 

628 from lexigram.admin.middleware.auth import AdminAuthMiddleware 

629 

630 user_store = await admin_resolver.resolve( 

631 AdminUserStoreProtocol, 

632 bypass_visibility=True, 

633 ) 

634 session_service = await admin_resolver.resolve( 

635 AdminSessionServiceProtocol, 

636 bypass_visibility=True, 

637 ) 

638 middleware_stack.append( 

639 ( 

640 AdminAuthMiddleware, 

641 { 

642 "user_store": user_store, 

643 "session_service": session_service, 

644 "require_auth": False, 

645 }, 

646 ) 

647 ) 

648 _log.debug("admin.auth_middleware_wired") 

649 except Exception as exc: # noqa: BLE001 — auth middleware is optional 

650 _log.warning("admin.auth_middleware_skipped", reason=str(exc)) 

651 

652 # Wire request-entry RBAC middleware — checks authorization before 

653 # dispatching to handlers (AUTH-09, AUTH-18). Placed after the auth 

654 # guard so request.state.user is populated. 

655 try: 

656 from lexigram.admin.middleware.authorization import ( 

657 AdminAuthorizationMiddleware, 

658 ) 

659 from lexigram.contracts.admin.authorizer import ( 

660 AdminAuthorizerProtocol, 

661 ) 

662 

663 authorizer = await admin_resolver.resolve( 

664 AdminAuthorizerProtocol, 

665 bypass_visibility=True, 

666 ) 

667 middleware_stack.append( 

668 (AdminAuthorizationMiddleware, {"authorizer": authorizer}) 

669 ) 

670 _log.debug("admin.authorization_middleware_wired") 

671 except Exception as exc: # noqa: BLE001 — authorization middleware is optional 

672 _log.warning("admin.authorization_middleware_skipped", reason=str(exc)) 

673 

674 # Wire tenant middleware when tenancy is enabled (before auth guard 

675 # so request.state.tenant_id is available during auth checks). 

676 if self._config.tenancy.enabled: 

677 from lexigram.admin.middleware.tenant import AdminTenantMiddleware 

678 

679 middleware_stack.insert( 

680 0, (AdminTenantMiddleware, {"config": self._config.tenancy}) 

681 ) 

682 _log.debug("admin.tenant_middleware_wired") 

683 

684 # Wire AdminErrorMiddleware innermost so it catches exceptions from 

685 # all other middleware and controllers. Best-effort — if the 

686 # middleware fails to resolve, admin runs without custom error pages. 

687 try: 

688 from lexigram.admin.middleware.error import AdminErrorMiddleware 

689 

690 middleware_stack.append( 

691 ( 

692 AdminErrorMiddleware, 

693 { 

694 "debug": self._config.debug, 

695 "login_url": f"{self._config.prefix}/login", 

696 }, 

697 ) 

698 ) 

699 _log.debug("admin.error_middleware_wired") 

700 except Exception as exc: # noqa: BLE001 — error middleware is optional 

701 _log.warning("admin.error_middleware_skipped", reason=str(exc)) 

702 

703 # Wire HX-Push-Url for body-targeted htmx GETs so client-side 

704 # navigation (htmx.ajax with target "body") keeps the address bar 

705 # in sync and htmx history (back/forward) works. 

706 from lexigram.admin.middleware.nav_push import AdminNavPushMiddleware 

707 

708 middleware_stack.append((AdminNavPushMiddleware, {})) 

709 _log.debug("admin.nav_push_middleware_wired") 

710 

711 # Wrap resource data sources with tenant scoping when enabled 

712 if self._config.tenancy.enabled: 

713 from lexigram.admin.multitenancy.data_source import TenantScopedDataSource 

714 

715 tenant_id = self._config.tenancy.default_tenant_id 

716 tenant_field = self._config.tenancy.tenant_field 

717 for resource in resources_dict.values(): 

718 ds = getattr(resource, "data_source", None) or getattr( 

719 resource, "_data_source", None 

720 ) 

721 if ds is not None: 

722 scoped = TenantScopedDataSource( 

723 data_source=ds, 

724 tenant_id=tenant_id, 

725 tenant_field=tenant_field, 

726 ) 

727 resource.data_source = scoped 

728 _log.debug("admin.resource_data_sources_tenant_scoped") 

729 

730 from lexigram.admin.contributors.registry import ContributorRegistry 

731 from lexigram.admin.contributors.resource_collector import ResourceCollector 

732 from lexigram.admin.dashboard.naming_policy import NamingPolicy 

733 from lexigram.admin.dashboard.route_integrator import RouteIntegrator 

734 

735 contributors: list = [] 

736 try: 

737 registry = await admin_resolver.resolve( 

738 ContributorRegistry, 

739 bypass_visibility=True, 

740 ) 

741 contributors = list(registry.get_all()) 

742 except Exception as exc: 

743 _log.warning("admin.contributors_discovery_failed", exc_info=True) 

744 self._mount_failures["contributor_discovery"] = str(exc) 

745 

746 try: 

747 naming = NamingPolicy(mode=self._config.contributor_collision_mode) 

748 collector = ResourceCollector(naming_policy=naming) 

749 contributor_resources = collector.collect(contributors) 

750 for resource_cls in contributor_resources: 

751 name = ( 

752 getattr(resource_cls, "name", None) 

753 or resource_cls.__name__.replace("Resource", "").lower() 

754 ) 

755 try: 

756 resources_dict[name] = await admin_resolver.resolve( 

757 resource_cls, 

758 bypass_visibility=True, 

759 ) 

760 except Exception: # noqa: BLE001 — fall back to direct 

761 try: 

762 resources_dict[name] = resource_cls() 

763 except Exception as inner: # noqa: BLE001 — skip, log warning 

764 _log.warning( 

765 "admin.contributor_resource_resolution_failed", 

766 resource=resource_cls.__name__, 

767 contributor=name, 

768 error=str(inner), 

769 ) 

770 self._mount_failures[f"contributor_resource:{name}"] = str( 

771 inner 

772 ) 

773 

774 # Wire data sources to resolved resources that declare _data_source_class 

775 for name, resource in list(resources_dict.items()): 

776 dsc = getattr(type(resource), "_data_source_class", None) 

777 if dsc is not None and hasattr(resource, "set_data_source"): 

778 try: 

779 ds = await admin_resolver.resolve(dsc, bypass_visibility=True) 

780 resource.set_data_source(ds) 

781 _log.debug("admin.data_source_wired", resource=name) 

782 except Exception: 

783 _log.debug("admin.data_source_wiring_failed", resource=name) 

784 

785 # Wrap data source with search wrappers when the resource 

786 # has a searchable spec and the search engine is available. 

787 search_spec = resource.search_spec() 

788 if search_spec and search_spec.index_name: 

789 try: 

790 from lexigram.search.engine import SearchEngine 

791 

792 search_engine = await admin_resolver.resolve( 

793 SearchEngine, 

794 bypass_visibility=True, 

795 ) 

796 

797 from lexigram.admin.integrations.search_query import ( 

798 SearchQueryDataSourceWrapper, 

799 ) 

800 from lexigram.admin.integrations.search_sync import ( 

801 SearchSyncDataSourceWrapper, 

802 ) 

803 

804 fallback_to_like = getattr( 

805 self._config.integrations.search, 

806 "fallback_to_like", 

807 True, 

808 ) 

809 query_wrapped = SearchQueryDataSourceWrapper( 

810 ds, 

811 search_engine, 

812 search_spec.index_name, 

813 fallback_to_like=fallback_to_like, 

814 ) 

815 wrapped = SearchSyncDataSourceWrapper( 

816 query_wrapped, search_engine, search_spec 

817 ) 

818 resource.set_data_source(wrapped) 

819 _log.debug("admin.search_wired", resource=name) 

820 except Exception: 

821 _log.debug("admin.search_wiring_failed", resource=name) 

822 

823 except Exception: # noqa: BLE001 — resource collection is non-fatal 

824 _log.warning("admin.contributors_resource_collection_failed", exc_info=True) 

825 

826 router = AdminRouter( 

827 config=self._config, 

828 resources=resources_dict, 

829 controllers=controller_instances, 

830 middleware_stack=middleware_stack, 

831 ) 

832 

833 try: 

834 naming = NamingPolicy(mode=self._config.contributor_collision_mode) 

835 integrator = RouteIntegrator( 

836 router=router, 

837 naming_policy=naming, 

838 route_prefix=self._config.prefix, 

839 container=container, 

840 ) 

841 integrator.register(contributors) 

842 except Exception as exc: # noqa: BLE001 — route integration is non-fatal 

843 _log.warning("admin.contributors_route_integration_failed", exc_info=True) 

844 self._mount_failures["route_integrator"] = str(exc) 

845 

846 # Register SSE endpoint for real-time notification streaming 

847 try: 

848 from lexigram.admin.realtime.sse import AdminEventHub, AdminEventsHandler # noqa: I001 

849 from lexigram.serialization import dumps_str 

850 from starlette.responses import StreamingResponse 

851 

852 hub: AdminEventHub = await container.resolve(AdminEventHub) 

853 

854 async def sse_event_stream(request: Any) -> StreamingResponse: 

855 handler = AdminEventsHandler(hub) 

856 

857 async def event_generator() -> AsyncGenerator[str, None]: