Coverage for src/local_deep_research/settings/env_settings.py: 98%

154 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-06-03 23:15 +0000

1""" 

2Environment-only settings that are loaded early and never stored in database. 

3 

4These settings are: 

51. Required before database initialization 

62. Used for testing/CI configuration 

73. System bootstrap configuration 

8 

9They are accessed through SettingsManager but always read from environment variables. 

10 

11Why some settings must be environment-only: 

12- Bootstrap settings (paths, encryption keys) are needed to initialize the database itself 

13- Database configuration settings must be available before connecting to the database 

14- Testing flags need to be checked before any database operations occur 

15- CI/CD variables control build-time behavior before the application starts 

16 

17These settings cannot be stored in the database because they are prerequisites for 

18accessing the database. This creates a bootstrapping requirement where certain 

19configuration must come from the environment to establish the system state needed 

20to access persisted settings. 

21""" 

22 

23import os 

24from pathlib import Path 

25from typing import Any, Dict, Optional, List, Set, TypeVar, Generic 

26from abc import ABC, abstractmethod 

27from loguru import logger 

28 

29from .exceptions import ( 

30 EnvironmentPathNotFoundError, 

31 EnvironmentValueRangeError, 

32 InvalidEnvironmentValueError, 

33 MissingEnvironmentVariableError, 

34) 

35 

36T = TypeVar("T") 

37 

38 

39class EnvSetting(ABC, Generic[T]): 

40 """Base class for all environment settings.""" 

41 

42 def __init__( 

43 self, 

44 key: str, 

45 description: str, 

46 default: Optional[T] = None, 

47 required: bool = False, 

48 deprecated_env_var: Optional[str] = None, 

49 ): 

50 self.key = key 

51 # Auto-generate env_var from key 

52 # e.g., "testing.test_mode" -> "LDR_TESTING_TEST_MODE" 

53 self.env_var = "LDR_" + key.upper().replace(".", "_") 

54 self.description = description 

55 self.default = default 

56 self.required = required 

57 self.deprecated_env_var = deprecated_env_var 

58 

59 def get_value(self) -> Optional[T]: 

60 """Get the value from environment with type conversion.""" 

61 raw = self._get_raw_value() 

62 if raw is None: 

63 if self.required and self.default is None: 

64 raise MissingEnvironmentVariableError(self.env_var) 

65 return self.default 

66 return self._convert_value(raw) 

67 

68 @abstractmethod 

69 def _convert_value(self, raw: str) -> T: 

70 """Convert raw string value to the appropriate type.""" 

71 pass 

72 

73 def _get_raw_value(self) -> Optional[str]: 

74 """Get raw string value from environment. 

75 

76 Checks the canonical env var first. If not set and a deprecated 

77 alias is configured, falls back to the deprecated name with a 

78 warning guiding users to migrate. 

79 """ 

80 value = os.environ.get(self.env_var) 

81 if value is not None: 

82 return value 

83 

84 if self.deprecated_env_var: 

85 deprecated_value = os.environ.get(self.deprecated_env_var) 

86 if deprecated_value is not None: 

87 logger.warning( 

88 f"Environment variable '{self.deprecated_env_var}' is deprecated " 

89 f"and will be removed in a future release. " 

90 f"Please use '{self.env_var}' instead." 

91 ) 

92 return deprecated_value 

93 

94 return None 

95 

96 @property 

97 def is_set(self) -> bool: 

98 """Check if the environment variable is set.""" 

99 return self.env_var in os.environ 

100 

101 def __repr__(self) -> str: 

102 """String representation for debugging.""" 

103 return f"{self.__class__.__name__}(key='{self.key}', env_var='{self.env_var}')" 

104 

105 

106class BooleanSetting(EnvSetting[bool]): 

107 """Boolean environment setting.""" 

108 

109 def __init__( 

110 self, 

111 key: str, 

112 description: str, 

113 default: bool = False, 

114 deprecated_env_var: Optional[str] = None, 

115 ): 

116 super().__init__( 

117 key, description, default, deprecated_env_var=deprecated_env_var 

118 ) 

119 

120 def _convert_value(self, raw: str) -> bool: 

121 """Convert string to boolean.""" 

122 return raw.lower() in ("true", "1", "yes", "on", "enabled") 

123 

124 

125class StringSetting(EnvSetting[str]): 

126 """String environment setting.""" 

127 

128 def __init__( 

129 self, 

130 key: str, 

131 description: str, 

132 default: Optional[str] = None, 

133 required: bool = False, 

134 deprecated_env_var: Optional[str] = None, 

135 ): 

136 super().__init__( 

137 key, 

138 description, 

139 default, 

140 required, 

141 deprecated_env_var=deprecated_env_var, 

142 ) 

143 

144 def _convert_value(self, raw: str) -> str: 

145 """Return string value as-is.""" 

146 return raw 

147 

148 

149class IntegerSetting(EnvSetting[int]): 

150 """Integer environment setting.""" 

151 

152 def __init__( 

153 self, 

154 key: str, 

155 description: str, 

156 default: Optional[int] = None, 

157 min_value: Optional[int] = None, 

158 max_value: Optional[int] = None, 

159 deprecated_env_var: Optional[str] = None, 

160 ): 

161 super().__init__( 

162 key, description, default, deprecated_env_var=deprecated_env_var 

163 ) 

164 self.min_value = min_value 

165 self.max_value = max_value 

166 

167 def _convert_value(self, raw: str) -> Optional[int]: 

168 """Convert string to integer with validation.""" 

169 try: 

170 value = int(raw) 

171 except ValueError: 

172 logger.warning( 

173 f"Invalid integer value '{raw}' for {self.env_var}, using default: {self.default}" 

174 ) 

175 return self.default 

176 

177 if self.min_value is not None and value < self.min_value: 

178 raise EnvironmentValueRangeError( 

179 self.env_var, value, min_val=self.min_value 

180 ) 

181 if self.max_value is not None and value > self.max_value: 

182 raise EnvironmentValueRangeError( 

183 self.env_var, value, max_val=self.max_value 

184 ) 

185 return value 

186 

187 

188class PathSetting(StringSetting): 

189 """Path environment setting with validation.""" 

190 

191 def __init__( 

192 self, 

193 key: str, 

194 description: str, 

195 default: Optional[str] = None, 

196 must_exist: bool = False, 

197 create_if_missing: bool = False, 

198 ): 

199 super().__init__(key, description, default) 

200 self.must_exist = must_exist 

201 self.create_if_missing = create_if_missing 

202 

203 def get_value(self) -> Optional[str]: 

204 """Get path value with optional validation/creation.""" 

205 path_str = super().get_value() 

206 if path_str is None: 

207 return None 

208 

209 # Use pathlib for path operations 

210 path = Path(path_str).expanduser() 

211 # Expand environment variables manually since pathlib doesn't have expandvars 

212 # Note: os.path.expandvars is kept here as there's no pathlib equivalent 

213 # noqa: PLR0402 - Suppress pathlib check for this line 

214 path_str = os.path.expandvars(str(path)) 

215 path = Path(path_str).resolve() 

216 

217 if self.create_if_missing and not path.exists(): 

218 try: 

219 path.mkdir(parents=True, exist_ok=True) 

220 except OSError: 

221 logger.warning("Failed to create directory") 

222 elif self.must_exist and not path.exists(): 

223 # Only raise if explicitly required to exist 

224 raise EnvironmentPathNotFoundError(self.env_var, path) 

225 

226 return str(path) 

227 

228 

229class SecretSetting(StringSetting): 

230 """Secret/sensitive environment setting.""" 

231 

232 def __init__( 

233 self, 

234 key: str, 

235 description: str, 

236 default: Optional[str] = None, 

237 required: bool = False, 

238 ): 

239 super().__init__(key, description, default, required) 

240 

241 def __repr__(self) -> str: 

242 """Hide the value in string representation.""" 

243 return f"SecretSetting(key='{self.key}', value=***)" 

244 

245 def __str__(self) -> str: 

246 """Hide the value in string conversion.""" 

247 value = "SET" if self.is_set else "NOT SET" 

248 return f"{self.key}=<{value}>" 

249 

250 

251class EnumSetting(EnvSetting[str]): 

252 """Enum-style setting with allowed values.""" 

253 

254 def __init__( 

255 self, 

256 key: str, 

257 description: str, 

258 allowed_values: Set[str], 

259 default: Optional[str] = None, 

260 case_sensitive: bool = False, 

261 deprecated_env_var: Optional[str] = None, 

262 ): 

263 super().__init__( 

264 key, description, default, deprecated_env_var=deprecated_env_var 

265 ) 

266 self.allowed_values = allowed_values 

267 self.case_sensitive = case_sensitive 

268 

269 # Store lowercase versions for case-insensitive comparison 

270 if not case_sensitive: 

271 self._allowed_lower = {v.lower() for v in allowed_values} 

272 # Create a mapping from lowercase to original case 

273 self._canonical_map = {v.lower(): v for v in allowed_values} 

274 

275 def _convert_value(self, raw: str) -> str: 

276 """Convert and validate value against allowed values.""" 

277 if self.case_sensitive: 

278 if raw not in self.allowed_values: 

279 raise InvalidEnvironmentValueError( 

280 self.env_var, raw, list(self.allowed_values) 

281 ) 

282 return raw 

283 # Case-insensitive matching 

284 raw_lower = raw.lower() 

285 if raw_lower not in self._allowed_lower: 

286 raise InvalidEnvironmentValueError( 

287 self.env_var, raw, list(self.allowed_values) 

288 ) 

289 # Return the canonical version (from allowed_values) 

290 return self._canonical_map[raw_lower] 

291 

292 

293class SettingsRegistry: 

294 """Registry for all environment settings.""" 

295 

296 def __init__(self): 

297 self._settings: Dict[str, EnvSetting] = {} 

298 self._categories: Dict[str, List[EnvSetting]] = {} 

299 

300 def register_category(self, category: str, settings: List[EnvSetting]): 

301 """Register a category of settings.""" 

302 self._categories[category] = settings 

303 for setting in settings: 

304 self._settings[setting.key] = setting 

305 

306 def get(self, key: str, default: Optional[Any] = None) -> Any: 

307 """ 

308 Get a setting value. 

309 

310 Args: 

311 key: Setting key (e.g., "testing.test_mode") 

312 default: Default value if not set or on error 

313 

314 Returns: 

315 Setting value or default 

316 """ 

317 setting = self._settings.get(key) 

318 if not setting: 

319 return default 

320 

321 try: 

322 value = setting.get_value() 

323 # Use provided default if setting returns None 

324 return value if value is not None else default 

325 except ValueError: 

326 logger.warning( 

327 "Validation error for setting '{}', using default", key 

328 ) 

329 return default 

330 

331 def get_setting_object(self, key: str) -> Optional[EnvSetting]: 

332 """Get the setting object itself for introspection.""" 

333 return self._settings.get(key) 

334 

335 def is_env_only(self, key: str) -> bool: 

336 """Check if a key is an env-only setting.""" 

337 return key in self._settings 

338 

339 def get_env_var(self, key: str) -> Optional[str]: 

340 """Get the environment variable name for a key.""" 

341 setting = self._settings.get(key) 

342 return setting.env_var if setting else None 

343 

344 def get_all_env_vars(self) -> Dict[str, str]: 

345 """Get all environment variables and descriptions.""" 

346 return { 

347 setting.env_var: setting.description 

348 for setting in self._settings.values() 

349 } 

350 

351 def get_category_settings(self, category: str) -> List[EnvSetting]: 

352 """Get all settings in a category.""" 

353 return self._categories.get(category, []) 

354 

355 def get_bootstrap_vars(self) -> Dict[str, str]: 

356 """Get bootstrap environment variables (bootstrap + db_config).""" 

357 result = {} 

358 for category in ["bootstrap", "db_config"]: 

359 for setting in self._categories.get(category, []): 

360 result[setting.env_var] = setting.description 

361 return result 

362 

363 def get_testing_vars(self) -> Dict[str, str]: 

364 """Get testing environment variables.""" 

365 result = {} 

366 for setting in self._categories.get("testing", []): 

367 result[setting.env_var] = setting.description 

368 return result 

369 

370 def list_all_settings(self) -> List[str]: 

371 """List all registered setting keys.""" 

372 return list(self._settings.keys()) 

373 

374 

375# Export list for better IDE discovery 

376__all__ = [ 

377 "EnvSetting", 

378 "BooleanSetting", 

379 "StringSetting", 

380 "IntegerSetting", 

381 "PathSetting", 

382 "SecretSetting", 

383 "EnumSetting", 

384 "SettingsRegistry", 

385]