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

155 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-06 15:42 +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 # Lazy import: this module is loaded very early (bootstrap 

220 # env-only settings), before the database and app are 

221 # initialized, so importing security at module load time 

222 # risks an import cycle. 

223 from ..security.directory_creation import create_directory 

224 

225 create_directory( 

226 path, 

227 context=f"env-configured path setting '{self.key}'", 

228 ) 

229 except OSError: 

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

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

232 # Only raise if explicitly required to exist 

233 raise EnvironmentPathNotFoundError(self.env_var, path) 

234 

235 return str(path) 

236 

237 

238class SecretSetting(StringSetting): 

239 """Secret/sensitive environment setting.""" 

240 

241 def __init__( 

242 self, 

243 key: str, 

244 description: str, 

245 default: Optional[str] = None, 

246 required: bool = False, 

247 ): 

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

249 

250 def __repr__(self) -> str: 

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

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

253 

254 def __str__(self) -> str: 

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

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

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

258 

259 

260class EnumSetting(EnvSetting[str]): 

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

262 

263 def __init__( 

264 self, 

265 key: str, 

266 description: str, 

267 allowed_values: Set[str], 

268 default: Optional[str] = None, 

269 case_sensitive: bool = False, 

270 deprecated_env_var: Optional[str] = None, 

271 ): 

272 super().__init__( 

273 key, description, default, deprecated_env_var=deprecated_env_var 

274 ) 

275 self.allowed_values = allowed_values 

276 self.case_sensitive = case_sensitive 

277 

278 # Store lowercase versions for case-insensitive comparison 

279 if not case_sensitive: 

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

281 # Create a mapping from lowercase to original case 

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

283 

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

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

286 if self.case_sensitive: 

287 if raw not in self.allowed_values: 

288 raise InvalidEnvironmentValueError( 

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

290 ) 

291 return raw 

292 # Case-insensitive matching 

293 raw_lower = raw.lower() 

294 if raw_lower not in self._allowed_lower: 

295 raise InvalidEnvironmentValueError( 

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

297 ) 

298 # Return the canonical version (from allowed_values) 

299 return self._canonical_map[raw_lower] 

300 

301 

302class SettingsRegistry: 

303 """Registry for all environment settings.""" 

304 

305 def __init__(self): 

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

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

308 

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

310 """Register a category of settings.""" 

311 self._categories[category] = settings 

312 for setting in settings: 

313 self._settings[setting.key] = setting 

314 

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

316 """ 

317 Get a setting value. 

318 

319 Args: 

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

321 default: Default value if not set or on error 

322 

323 Returns: 

324 Setting value or default 

325 """ 

326 setting = self._settings.get(key) 

327 if not setting: 

328 return default 

329 

330 try: 

331 value = setting.get_value() 

332 # Use provided default if setting returns None 

333 return value if value is not None else default 

334 except ValueError: 

335 logger.warning( 

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

337 ) 

338 return default 

339 

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

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

342 return self._settings.get(key) 

343 

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

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

346 return key in self._settings 

347 

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

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

350 setting = self._settings.get(key) 

351 return setting.env_var if setting else None 

352 

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

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

355 return { 

356 setting.env_var: setting.description 

357 for setting in self._settings.values() 

358 } 

359 

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

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

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

363 

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

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

366 result = {} 

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

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

369 result[setting.env_var] = setting.description 

370 return result 

371 

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

373 """Get testing environment variables.""" 

374 result = {} 

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

376 result[setting.env_var] = setting.description 

377 return result 

378 

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

380 """List all registered setting keys.""" 

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

382 

383 

384# Export list for better IDE discovery 

385__all__ = [ 

386 "EnvSetting", 

387 "BooleanSetting", 

388 "StringSetting", 

389 "IntegerSetting", 

390 "PathSetting", 

391 "SecretSetting", 

392 "EnumSetting", 

393 "SettingsRegistry", 

394]