Coverage for src / augint_library / feature_flags.py: 97%

1"""Feature flags system for controlling feature rollout and A/B testing. 

2 

3This module provides a comprehensive feature flag system that enables: 

4- Gradual feature rollouts 

5- A/B testing 

6- Emergency kill switches 

7- User targeting 

8- Environment-specific configuration 

9 

10Common Use Cases: 

11 1. Rolling out new features gradually 

12 2. A/B testing different implementations 

13 3. Emergency feature disabling 

14 4. Beta testing with specific users 

15 5. Environment-specific features 

16 

17Examples: 

18 Basic feature flag usage: 

19 >>> from augint_library import FeatureFlags, feature_flag 

20 >>> 

21 >>> flags = FeatureFlags() 

22 >>> 

23 >>> @feature_flag("new_algorithm", default=False) 

24 >>> def process_data(data): 

25 ... if flags.is_enabled("new_algorithm"): 

26 ... return new_algorithm(data) 

27 ... else: 

28 ... return old_algorithm(data) 

29 

30 Percentage-based rollout: 

31 >>> flags.set_flag("dark_mode", { 

32 ... "enabled": True, 

33 ... "rollout_percentage": 25 # 25% of users 

34 ... }) 

35 >>> 

36 >>> # Check for specific user 

37 >>> if flags.is_enabled("dark_mode", user_id="user123"): 

38 ... render_dark_theme() 

39 

40 User targeting: 

41 >>> flags.set_flag("beta_feature", { 

42 ... "enabled": True, 

43 ... "allowed_users": ["alice@example.com", "bob@example.com"], 

44 ... "denied_users": ["charlie@example.com"] 

45 ... }) 

46 

47 Environment-specific flags: 

48 >>> # In production config 

49 >>> flags.set_flag("debug_mode", {"enabled": False}) 

50 >>> 

51 >>> # In development config 

52 >>> flags.set_flag("debug_mode", {"enabled": True}) 

53 

54 A/B testing with variants: 

55 >>> flags.set_flag("checkout_flow", { 

56 ... "enabled": True, 

57 ... "variants": { 

58 ... "control": {"weight": 50, "config": {"steps": 3}}, 

59 ... "test": {"weight": 50, "config": {"steps": 1}} 

60 ... } 

61 ... }) 

62 >>> 

63 >>> variant = flags.get_variant("checkout_flow", user_id="user123") 

64 >>> if variant == "test": 

65 ... show_single_page_checkout() 

66 

67Configuration Examples: 

68 JSON configuration file: 

69 ```json 

70 { 

71 "features": { 

72 "new_dashboard": { 

73 "enabled": true, 

74 "rollout_percentage": 10, 

75 "allowed_users": ["beta@example.com"] 

76 }, 

77 "experimental_api": { 

78 "enabled": false, 

79 "description": "New API endpoint - disabled due to bug" 

80 } 

81 } 

82 } 

83 ``` 

84 

85 Environment variables: 

86 ```bash 

87 export FEATURE_NEW_DASHBOARD=true 

88 export FEATURE_EXPERIMENTAL_API=false 

89 ``` 

90 

91Best Practices: 

92 1. Always provide meaningful defaults 

93 2. Document feature flags in code 

94 3. Clean up old flags regularly 

95 4. Use descriptive flag names 

96 5. Monitor flag usage 

97 

98Advanced Patterns: 

99 Context manager for temporary flags: 

100 >>> with flags.override("debug_mode", True): 

101 ... # Debug mode is enabled here 

102 ... debug_function() 

103 ... # Debug mode returns to previous state 

104 

105 Decorator with fallback: 

106 >>> @feature_flag("new_feature", fallback=legacy_function) 

107 >>> def new_function(data): 

108 ... return enhanced_processing(data) 

109 

110Note: 

111 Feature flags are powerful but can add complexity. Use them judiciously 

112 and have a process for removing flags once features are fully rolled out. 

113""" 

114 

115import hashlib 

116import json 

117import logging 

118import os 

119import random 

120import re 

121from collections.abc import Iterator 

122from contextlib import contextmanager 

123from pathlib import Path 

124from typing import Any, Callable, Optional, Union 

125 

126from .constants import HASH_PREFIX_LENGTH, PERCENTAGE_MAX, PERCENTAGE_MIN, USER_ID_RANDOM_MAX 

127from .exceptions import ConfigurationError, ValidationError 

128 

129logger = logging.getLogger(__name__) 

130 

131 

132class FeatureFlag: 

133 """Represents a single feature flag with its configuration.""" 

134 

135 def __init__( 

136 self, 

137 name: str, 

138 default: str = "disabled", 

139 description: Optional[str] = None, 

140 metadata: Optional[dict[str, Any]] = None, 

141 evaluator: Optional[Callable[[dict[str, Any]], bool]] = None, 

142 ): 

143 """Initialize a feature flag. 

144 

145 Args: 

146 name: Flag identifier (alphanumeric + underscore/hyphen) 

147 default: Default state ("enabled", "disabled", "percentage:N", "conditional") 

148 description: Human-readable description 

149 metadata: Additional metadata for the flag 

150 evaluator: Function for conditional evaluation 

151 """ 

152 self.name = self._validate_name(name) 

153 self.default = self._validate_state(default) 

154 self.description = description 

155 self.metadata = metadata or {} 

156 self.evaluator = evaluator 

157 self.state = self.default 

158 self._override_state: Optional[str] = None 

159 

160 @staticmethod 

161 def _validate_name(name: str) -> str: 

162 """Validate flag name format.""" 

163 if not re.match(r"^[a-zA-Z0-9_-]+$", name): 

164 raise ValidationError( 

165 f"Invalid flag name '{name}'. Use only alphanumeric, underscore, and hyphen." 

166 ) 

167 return name 

168 

169 @staticmethod 

170 def _validate_state(state: str) -> str: 

171 """Validate state format.""" 

172 if state in ["enabled", "disabled", "conditional"]: 

173 return state 

174 if state.startswith("percentage:"): 

175 try: 

176 percentage = int(state.split(":")[1]) 

177 if PERCENTAGE_MIN <= percentage <= PERCENTAGE_MAX: 

178 return state 

179 except (IndexError, ValueError): 

180 pass 

181 raise ValidationError( 

182 f"Invalid state '{state}'. Use 'enabled', 'disabled', 'percentage:N' (0-100), or 'conditional'." # noqa: E501 

183 ) 

184 

185 def evaluate(self, context: Optional[dict[str, Any]] = None) -> bool: # noqa: PLR0911 

186 """Evaluate if the flag is enabled given the context.""" 

187 state = self._override_state if self._override_state else self.state 

188 

189 if state == "enabled": 

190 return True 

191 if state == "disabled": 

192 return False 

193 if state.startswith("percentage:"): 

194 percentage = int(state.split(":")[1]) 

195 return self._evaluate_percentage(percentage, context) 

196 if state == "conditional": 196 ↛ 207line 196 didn't jump to line 207 because the condition on line 196 was always true

197 if not self.evaluator: 

198 logger.warning( 

199 f"No evaluator for conditional flag '{self.name}', defaulting to disabled" 

200 ) 

201 return False 

202 try: 

203 return self.evaluator(context or {}) 

204 except Exception: 

205 logger.exception(f"Error evaluating conditional flag '{self.name}'") 

206 return False 

207 return False 

208 

209 def _evaluate_percentage( 

210 self, percentage: int, context: Optional[dict[str, Any]] = None 

211 ) -> bool: 

212 """Evaluate percentage-based rollout with consistent hashing.""" 

213 if percentage == 0: 

214 return False 

215 if percentage == 100: # noqa: PLR2004 

216 return True 

217 

218 context = context or {} 

219 user_id = context.get("user_id", str(random.randint(0, USER_ID_RANDOM_MAX))) # noqa: S311 

220 hash_input = f"{self.name}:{user_id}" 

221 hash_value = hashlib.md5(hash_input.encode(), usedforsecurity=False).hexdigest() 

222 user_bucket = int(hash_value[:HASH_PREFIX_LENGTH], 16) % PERCENTAGE_MAX 

223 return user_bucket < percentage 

224 

225 

226class FeatureFlags: 

227 """Main class for managing feature flags.""" 

228 

229 _instance: Optional["FeatureFlags"] = None 

230 

231 def __new__(cls, *args: Any, **kwargs: Any) -> "FeatureFlags": # noqa: ARG004 

232 """Implement singleton pattern.""" 

233 if cls._instance is None: 

234 cls._instance = super().__new__(cls) 

235 return cls._instance 

236 

237 def __init__(self, config_file: Optional[str] = None, auto_load_env: bool = True): 

238 """Initialize feature flags manager. 

239 

240 Args: 

241 config_file: Path to JSON configuration file 

242 auto_load_env: Whether to automatically load from environment variables 

243 """ 

244 if hasattr(self, "_initialized"): 

245 return 

246 self._initialized = True 

247 

248 self._flags: dict[str, FeatureFlag] = {} 

249 self._evaluators: dict[str, Callable[[dict[str, Any]], bool]] = {} 

250 

251 if config_file: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true

252 self.load_from_file(config_file) 

253 

254 if auto_load_env: 

255 self.load_from_env() 

256 

257 def register( 

258 self, 

259 name: str, 

260 default: str = "disabled", 

261 description: Optional[str] = None, 

262 metadata: Optional[dict[str, Any]] = None, 

263 evaluator: Optional[str] = None, 

264 ) -> None: 

265 """Register a new feature flag. 

266 

267 Args: 

268 name: Flag identifier 

269 default: Default state 

270 description: Human-readable description 

271 metadata: Additional metadata 

272 evaluator: Name of registered evaluator function 

273 """ 

274 evaluator_func = None 

275 if evaluator: 

276 if evaluator not in self._evaluators: 

277 raise ConfigurationError(f"Evaluator '{evaluator}' not registered") 

278 evaluator_func = self._evaluators[evaluator] 

279 

280 flag = FeatureFlag(name, default, description, metadata, evaluator_func) 

281 self._flags[name] = flag 

282 logger.debug(f"Registered feature flag '{name}' with default state '{default}'") 

283 

284 def register_evaluator(self, name: str, func: Callable[[dict[str, Any]], bool]) -> None: 

285 """Register a custom evaluator function. 

286 

287 Args: 

288 name: Evaluator name 

289 func: Function that takes context and returns bool 

290 """ 

291 self._evaluators[name] = func 

292 logger.debug(f"Registered evaluator '{name}'") 

293 

294 def is_enabled(self, name: str, context: Optional[dict[str, Any]] = None) -> bool: 

295 """Check if a feature flag is enabled. 

296 

297 Args: 

298 name: Flag name 

299 context: Evaluation context (e.g., user_id, environment) 

300 

301 Returns: 

302 True if enabled, False otherwise 

303 """ 

304 if name not in self._flags: 

305 logger.warning(f"Unknown feature flag '{name}', defaulting to disabled") 

306 return False 

307 

308 return self._flags[name].evaluate(context) 

309 

310 def set_state(self, name: str, state: str) -> None: 

311 """Set the runtime state of a flag. 

312 

313 Args: 

314 name: Flag name 

315 state: New state 

316 """ 

317 if name not in self._flags: 

318 raise ValidationError(f"Unknown feature flag '{name}'") 

319 

320 self._flags[name].state = self._flags[name]._validate_state(state) 

321 logger.info(f"Set feature flag '{name}' to state '{state}'") 

322 

323 def get_state(self, name: str) -> str: 

324 """Get the current state of a flag.""" 

325 if name not in self._flags: 

326 raise ValidationError(f"Unknown feature flag '{name}'") 

327 return self._flags[name].state 

328 

329 def get_all_flags(self) -> dict[str, dict[str, Any]]: 

330 """Get all registered flags with their current state.""" 

331 return { 

332 name: { 

333 "state": flag.state, 

334 "default": flag.default, 

335 "description": flag.description, 

336 "metadata": flag.metadata, 

337 } 

338 for name, flag in self._flags.items() 

339 } 

340 

341 @contextmanager 

342 def override(self, name: str, state: str) -> Iterator[None]: 

343 """Temporarily override a flag's state. 

344 

345 Args: 

346 name: Flag name 

347 state: Temporary state 

348 """ 

349 if name not in self._flags: 

350 raise ValidationError(f"Unknown feature flag '{name}'") 

351 

352 flag = self._flags[name] 

353 validated_state = flag._validate_state(state) 

354 original_override = flag._override_state 

355 

356 try: 

357 flag._override_state = validated_state 

358 logger.debug(f"Overriding flag '{name}' to '{state}'") 

359 yield 

360 finally: 

361 flag._override_state = original_override 

362 logger.debug(f"Restored flag '{name}' override") 

363 

364 def load_from_file(self, file_path: Union[str, Path]) -> None: 

365 """Load flag configuration from JSON file. 

366 

367 Args: 

368 file_path: Path to JSON configuration file 

369 """ 

370 path = Path(file_path) 

371 if not path.exists(): 

372 raise ConfigurationError(f"Configuration file not found: {file_path}") 

373 

374 try: 

375 with path.open() as f: 

376 config = json.load(f) 

377 except json.JSONDecodeError as e: 

378 raise ConfigurationError(f"Invalid JSON in configuration file: {e}") from e 

379 

380 for flag_name, flag_config in config.items(): 

381 if isinstance(flag_config, str): 

382 # Simple state configuration 

383 self.register(flag_name, default=flag_config) 

384 elif isinstance(flag_config, dict): 384 ↛ 380line 384 didn't jump to line 380 because the condition on line 384 was always true

385 # Detailed configuration 

386 self.register( 

387 flag_name, 

388 default=flag_config.get("state", "disabled"), 

389 description=flag_config.get("description"), 

390 metadata=flag_config.get("metadata"), 

391 evaluator=flag_config.get("evaluator"), 

392 ) 

393 # Set current state if different from default 

394 if "state" in flag_config: 394 ↛ 380line 394 didn't jump to line 380 because the condition on line 394 was always true

395 self.set_state(flag_name, flag_config["state"]) 

396 

397 def load_from_env(self) -> None: 

398 """Load flag states from environment variables. 

399 

400 Environment variables should be in format: FEATURE_FLAG_{NAME}=state 

401 """ 

402 prefix = "FEATURE_FLAG_" 

403 for key, value in os.environ.items(): 

404 if key.startswith(prefix): 

405 flag_name = key[len(prefix) :].lower() 

406 try: 

407 if flag_name in self._flags: 

408 self.set_state(flag_name, value) 

409 else: 

410 self.register(flag_name, default=value) 

411 except ValidationError as e: 

412 logger.warning(f"Invalid environment variable {key}: {e}") 

413 

414 

415# Decorator for conditional execution 

416def feature_flag( 

417 flag_name: str, 

418 fallback: Any = None, 

419 flags: Optional[FeatureFlags] = None, 

420) -> Callable[[Callable[..., Any]], Callable[..., Any]]: 

421 """Decorator for conditional feature execution. 

422 

423 Args: 

424 flag_name: Name of the feature flag 

425 fallback: Value to return if flag is disabled 

426 flags: FeatureFlags instance (uses singleton if not provided) 

427 """ 

428 

429 def decorator(func: Callable[..., Any]) -> Callable[..., Any]: 

430 def wrapper(*args: Any, **kwargs: Any) -> Any: 

431 _flags = flags or FeatureFlags() 

432 

433 # Extract context from kwargs if provided 

434 context = kwargs.pop("_feature_context", None) 

435 

436 if _flags.is_enabled(flag_name, context): 

437 return func(*args, **kwargs) 

438 

439 if callable(fallback): 

440 return fallback(*args, **kwargs) 

441 return fallback 

442 

443 wrapper.__name__ = func.__name__ 

444 wrapper.__doc__ = func.__doc__ 

445 wrapper.__wrapped__ = func # type: ignore[attr-defined] 

446 return wrapper 

447 

448 return decorator 

449 

450 

451def get_flags() -> FeatureFlags: 

452 """Get the global FeatureFlags instance.""" 

453 return FeatureFlags()