Coverage for src/lexigram/features/di/provider.py: 95%

1"""DI provider for the feature-flag subsystem. 

2 

3:class:`FeatureFlagsProvider` registers a :class:`~lexigram.features.manager.FlagManager` 

4singleton in the container so application services can resolve it via DI 

5rather than constructing it manually. 

6 

7The default configuration uses a 

8:class:`~lexigram.features.backends.local.LocalProvider` for 

9the simple boolean contract and a 

10:class:`~lexigram.features.backends.local.LocalProvider` backed by 

11``config.initial_flags`` for the rich evaluation API. 

12""" 

13 

14from __future__ import annotations 

15 

16from typing import TYPE_CHECKING, Any, Self 

17 

18from lexigram.contracts.core import ProviderPriority 

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

20from lexigram.contracts.feature_flags import FlagProviderProtocol 

21from lexigram.di.provider import Provider 

22from lexigram.features.backends.local import LocalProvider 

23from lexigram.features.config import FeatureFlagsConfig 

24from lexigram.features.manager import FlagManager 

25from lexigram.features.types import Flag, FlagType 

26from lexigram.logging import get_logger 

27 

28if TYPE_CHECKING: 

29 from lexigram.contracts.core.di import ( 

30 BootContainerProtocol, 

31 ContainerRegistrarProtocol, 

32 ) 

33 

34 

35logger = get_logger(__name__) 

36 

37 

38class FeatureFlagsProvider(Provider): 

39 """Provider that registers feature-flag infrastructure in the container. 

40 

41 Registers: 

42 

43 * ``FlagProviderProtocol`` (simple boolean contract) as a singleton backed by 

44 :class:`~lexigram.features.backends.local.LocalProvider`. 

45 * ``FlagManager`` as a singleton wrapping a 

46 :class:`~lexigram.features.backends.local.LocalProvider` seeded from 

47 :attr:`~lexigram.features.config.FeatureFlagsConfig.initial_flags`. 

48 """ 

49 

50 name = "features" 

51 config_key: str | None = "features" 

52 config_model: type | None = FeatureFlagsConfig 

53 priority = ProviderPriority.INFRASTRUCTURE 

54 

55 def __init__(self, config: FeatureFlagsConfig | None = None) -> None: 

56 """Create the feature-flags provider. 

57 

58 Args: 

59 config: Optional feature-flag configuration. When omitted, 

60 defaults are used (all flags disabled, cache TTL 60 s). 

61 """ 

62 super().__init__() 

63 self._config = config or FeatureFlagsConfig() 

64 self._simple_provider: LocalProvider | None = None 

65 self._manager: FlagManager | None = None 

66 

67 @classmethod 

68 def from_config(cls, config: FeatureFlagsConfig, **context: Any) -> Self: 

69 """Create provider from config object.""" 

70 return cls(config=config) 

71 

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

73 """Register flag infrastructure in the container. 

74 

75 Registers ``FlagProviderProtocol`` (simple boolean API) and ``FlagManager`` 

76 (rich evaluation API) as singletons, seeded from the provider config. 

77 """ 

78 container.singleton(FeatureFlagsConfig, self._config) 

79 

80 if not self._config.enabled: 

81 logger.info("features_disabled", reason="FeatureFlagsConfig.enabled=False") 

82 return 

83 

84 # Simple boolean provider for the FlagProviderProtocol contract. 

85 simple = LocalProvider() 

86 for flag_name, enabled in self._config.initial_flags.items(): 

87 simple.set_flag_sync(flag_name, enabled) 

88 container.singleton(FlagProviderProtocol, simple) 

89 self._simple_provider = simple 

90 

91 # Rich provider + manager for full evaluation API. 

92 initial: dict[str, Flag] = { 

93 flag_name: Flag(name=flag_name, type=FlagType.BOOLEAN, enabled=enabled) 

94 for flag_name, enabled in self._config.initial_flags.items() 

95 } 

96 local = LocalProvider(initial) 

97 manager = FlagManager( 

98 local, 

99 cache_ttl=self._config.cache_ttl, 

100 default_enabled=self._config.default_enabled, 

101 ) 

102 container.singleton(FlagManager, manager) 

103 from lexigram.contracts.feature_flags.protocols import FlagManagerProtocol 

104 

105 container.singleton(FlagManagerProtocol, manager) 

106 self._manager = manager 

107 

108 async def boot(self, container: BootContainerProtocol) -> None: 

109 """Wire the event bus into the flag manager when one is available.""" 

110 if self._manager is None: 

111 return 

112 try: 

113 from lexigram.contracts.events.protocols import EventBusProtocol 

114 

115 event_bus = await container.resolve(EventBusProtocol) 

116 self._manager._event_bus = event_bus 

117 except Exception as e: # noqa: BLE001 

118 logger.debug( 

119 "event_bus_unavailable", 

120 error=str(e), 

121 ) # EventBusProtocol is optional; feature flags work without it 

122 

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

124 """Check health of feature flags. 

125 

126 Feature flags are in-memory and have no external dependencies, 

127 so this always returns HEALTHY. 

128 

129 Args: 

130 timeout: Not used - feature flags have no external dependencies. 

131 

132 Returns: 

133 HealthCheckResult showing healthy status. 

134 """ 

135 flag_count = ( 

136 len(self._config.initial_flags) if self._config.initial_flags else 0 

137 ) 

138 return HealthCheckResult( 

139 component="features", 

140 status=HealthStatus.HEALTHY, 

141 message=f"Feature flags operational with {flag_count} initial flags", 

142 details={"initial_flags": flag_count}, 

143 ) 

144 

145 async def shutdown(self) -> None: 

146 """No resources to release for the feature-flags module.""" 

147 

148 def get_simple_provider(self) -> LocalProvider | None: 

149 """Return the registered simple provider after registration. 

150 

151 Returns: 

152 The ``LocalProvider`` instance, or ``None`` before registration. 

153 """ 

154 return self._simple_provider 

155 

156 def get_manager(self) -> FlagManager | None: 

157 """Return the registered manager after registration. 

158 

159 Returns: 

160 The ``FlagManager`` instance, or ``None`` before registration. 

161 """ 

162 return self._manager 

163 

164 

165__all__ = ["FeatureFlagsProvider"]