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

145 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2026-01-11 00:51 +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 

29 

30T = TypeVar("T") 

31 

32 

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

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

35 

36 def __init__( 

37 self, 

38 key: str, 

39 description: str, 

40 default: Optional[T] = None, 

41 required: bool = False, 

42 ): 

43 self.key = key 

44 # Auto-generate env_var from key 

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

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

47 self.description = description 

48 self.default = default 

49 self.required = required 

50 

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

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

53 raw = self._get_raw_value() 

54 if raw is None: 

55 if self.required and self.default is None: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 raise ValueError( 

57 f"Required environment variable {self.env_var} is not set" 

58 ) 

59 return self.default 

60 return self._convert_value(raw) 

61 

62 @abstractmethod 

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

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

65 pass 

66 

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

68 """Get raw string value from environment.""" 

69 return os.environ.get(self.env_var) 

70 

71 @property 

72 def is_set(self) -> bool: 

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

74 return self.env_var in os.environ 

75 

76 def __repr__(self) -> str: 

77 """String representation for debugging.""" 

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

79 

80 

81class BooleanSetting(EnvSetting[bool]): 

82 """Boolean environment setting.""" 

83 

84 def __init__( 

85 self, 

86 key: str, 

87 description: str, 

88 default: bool = False, 

89 ): 

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

91 

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

93 """Convert string to boolean.""" 

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

95 

96 

97class StringSetting(EnvSetting[str]): 

98 """String environment setting.""" 

99 

100 def __init__( 

101 self, 

102 key: str, 

103 description: str, 

104 default: Optional[str] = None, 

105 required: bool = False, 

106 ): 

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

108 

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

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

111 return raw 

112 

113 

114class IntegerSetting(EnvSetting[int]): 

115 """Integer environment setting.""" 

116 

117 def __init__( 

118 self, 

119 key: str, 

120 description: str, 

121 default: Optional[int] = None, 

122 min_value: Optional[int] = None, 

123 max_value: Optional[int] = None, 

124 ): 

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

126 self.min_value = min_value 

127 self.max_value = max_value 

128 

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

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

131 try: 

132 value = int(raw) 

133 if self.min_value is not None and value < self.min_value: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 raise ValueError( 

135 f"{self.env_var} value {value} is below minimum {self.min_value}" 

136 ) 

137 if self.max_value is not None and value > self.max_value: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true

138 raise ValueError( 

139 f"{self.env_var} value {value} is above maximum {self.max_value}" 

140 ) 

141 return value 

142 except ValueError as e: 

143 if "invalid literal" in str(e): 143 ↛ 148line 143 didn't jump to line 148 because the condition on line 143 was always true

144 logger.warning( 

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

146 ) 

147 return self.default 

148 raise 

149 

150 

151class PathSetting(StringSetting): 

152 """Path environment setting with validation.""" 

153 

154 def __init__( 

155 self, 

156 key: str, 

157 description: str, 

158 default: Optional[str] = None, 

159 must_exist: bool = False, 

160 create_if_missing: bool = False, 

161 ): 

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

163 self.must_exist = must_exist 

164 self.create_if_missing = create_if_missing 

165 

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

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

168 path_str = super().get_value() 

169 if path_str is None: 

170 return None 

171 

172 # Use pathlib for path operations 

173 path = Path(path_str).expanduser() 

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

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

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

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

178 path = Path(path_str) 

179 

180 if self.create_if_missing and not path.exists(): 180 ↛ 186line 180 didn't jump to line 186 because the condition on line 180 was always true

181 try: 

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

183 except OSError: 

184 # Silently fail if we can't create - let app handle it 

185 pass 

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

187 # Only raise if explicitly required to exist 

188 raise ValueError( 

189 f"Path {path} specified in {self.env_var} does not exist" 

190 ) 

191 

192 return str(path) 

193 

194 

195class SecretSetting(StringSetting): 

196 """Secret/sensitive environment setting.""" 

197 

198 def __init__( 

199 self, 

200 key: str, 

201 description: str, 

202 default: Optional[str] = None, 

203 required: bool = False, 

204 ): 

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

206 

207 def __repr__(self) -> str: 

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

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

210 

211 def __str__(self) -> str: 

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

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

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

215 

216 

217class EnumSetting(EnvSetting[str]): 

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

219 

220 def __init__( 

221 self, 

222 key: str, 

223 description: str, 

224 allowed_values: Set[str], 

225 default: Optional[str] = None, 

226 case_sensitive: bool = False, 

227 ): 

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

229 self.allowed_values = allowed_values 

230 self.case_sensitive = case_sensitive 

231 

232 # Store lowercase versions for case-insensitive comparison 

233 if not case_sensitive: 233 ↛ exitline 233 didn't return from function '__init__' because the condition on line 233 was always true

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

235 # Create a mapping from lowercase to original case 

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

237 

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

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

240 if self.case_sensitive: 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true

241 if raw not in self.allowed_values: 

242 raise ValueError( 

243 f"{self.env_var} value '{raw}' not in allowed values: {self.allowed_values}" 

244 ) 

245 return raw 

246 else: 

247 # Case-insensitive matching 

248 raw_lower = raw.lower() 

249 if raw_lower not in self._allowed_lower: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 raise ValueError( 

251 f"{self.env_var} value '{raw}' not in allowed values: {self.allowed_values}" 

252 ) 

253 # Return the canonical version (from allowed_values) 

254 return self._canonical_map[raw_lower] 

255 

256 

257class SettingsRegistry: 

258 """Registry for all environment settings.""" 

259 

260 def __init__(self): 

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

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

263 

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

265 """Register a category of settings.""" 

266 self._categories[category] = settings 

267 for setting in settings: 

268 self._settings[setting.key] = setting 

269 

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

271 """ 

272 Get a setting value. 

273 

274 Args: 

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

276 default: Default value if not set or on error 

277 

278 Returns: 

279 Setting value or default 

280 """ 

281 setting = self._settings.get(key) 

282 if not setting: 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true

283 return default 

284 

285 try: 

286 value = setting.get_value() 

287 # Use provided default if setting returns None 

288 return value if value is not None else default 

289 except ValueError: 

290 # On validation error, return default 

291 return default 

292 

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

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

295 return self._settings.get(key) 

296 

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

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

299 return key in self._settings 

300 

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

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

303 setting = self._settings.get(key) 

304 return setting.env_var if setting else None 

305 

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

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

308 return { 

309 setting.env_var: setting.description 

310 for setting in self._settings.values() 

311 } 

312 

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

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

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

316 

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

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

319 result = {} 

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

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

322 result[setting.env_var] = setting.description 

323 return result 

324 

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

326 """Get testing environment variables.""" 

327 result = {} 

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

329 result[setting.env_var] = setting.description 

330 return result 

331 

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

333 """List all registered setting keys.""" 

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

335 

336 

337# Export list for better IDE discovery 

338__all__ = [ 

339 "EnvSetting", 

340 "BooleanSetting", 

341 "StringSetting", 

342 "IntegerSetting", 

343 "PathSetting", 

344 "SecretSetting", 

345 "EnumSetting", 

346 "SettingsRegistry", 

347]