Coverage for src/lexigram/admin/di/sub_providers/resource.py: 96%

1"""Admin resource sub-provider — resource registry, CRUD routing, handlers.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

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

8 

9if TYPE_CHECKING: 

10 from lexigram.admin.config import AdminConfig 

11 from lexigram.contracts.core.di import ( 

12 ContainerRegistrarProtocol, 

13 ContainerResolverProtocol, 

14 ) 

15 

16 

17class AdminResourceSubProvider: 

18 """Manages admin resource infrastructure: registry, CRUD routing, handlers. 

19 

20 Handles resource registration, CRUD action dispatch, and resource discovery. 

21 Accepts a list of Resource classes to register on initialization. 

22 """ 

23 

24 def __init__( 

25 self, 

26 config: AdminConfig, 

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

28 **kwargs: object, 

29 ) -> None: 

30 self._config = config 

31 self._resources = resources or [] 

32 self._kwargs = kwargs 

33 self._initialized = False 

34 

35 @property 

36 def config(self) -> AdminConfig: 

37 """Return current admin config.""" 

38 return self._config 

39 

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

41 """Register resource infrastructure: registry, manager, action handler registry.""" 

42 from lexigram.admin.core.registry import AdminRegistry 

43 from lexigram.admin.resources.handler import ActionHandlerRegistry 

44 from lexigram.admin.services.resource_manager_factory import ( 

45 ResourceManagerFactory, 

46 ) 

47 

48 # Register the class (container will instantiate via DI) 

49 container.singleton(ActionHandlerRegistry, ActionHandlerRegistry) 

50 container.singleton(ResourceManagerFactory, ResourceManagerFactory()) 

51 # Register the class (container will instantiate via DI with provider argument) 

52 container.singleton(AdminRegistry, AdminRegistry) 

53 

54 async def boot(self, container: ContainerResolverProtocol) -> None: 

55 """Boot resource infrastructure: initialize registries and discovery.""" 

56 self._initialized = True 

57 

58 async def shutdown(self) -> None: 

59 """Shut down resource infrastructure.""" 

60 self._initialized = False 

61 

62 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

63 """Return resource infrastructure health status.""" 

64 return HealthCheckResult( 

65 component="admin_resource", 

66 status=HealthStatus.HEALTHY if self._initialized else HealthStatus.UNKNOWN, 

67 message=f"Admin resources operational ({len(self._resources)} registered)" 

68 if self._initialized 

69 else "Not yet initialized", 

70 ) 

71 

72 

73__all__ = ["AdminResourceSubProvider"]