Coverage for src / local_deep_research / settings / env_settings.py: 98%
153 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1"""
2Environment-only settings that are loaded early and never stored in database.
4These settings are:
51. Required before database initialization
62. Used for testing/CI configuration
73. System bootstrap configuration
9They are accessed through SettingsManager but always read from environment variables.
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
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"""
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
30T = TypeVar("T")
33class EnvSetting(ABC, Generic[T]):
34 """Base class for all environment settings."""
36 def __init__(
37 self,
38 key: str,
39 description: str,
40 default: Optional[T] = None,
41 required: bool = False,
42 deprecated_env_var: Optional[str] = None,
43 ):
44 self.key = key
45 # Auto-generate env_var from key
46 # e.g., "testing.test_mode" -> "LDR_TESTING_TEST_MODE"
47 self.env_var = "LDR_" + key.upper().replace(".", "_")
48 self.description = description
49 self.default = default
50 self.required = required
51 self.deprecated_env_var = deprecated_env_var
53 def get_value(self) -> Optional[T]:
54 """Get the value from environment with type conversion."""
55 raw = self._get_raw_value()
56 if raw is None:
57 if self.required and self.default is None:
58 raise ValueError(
59 f"Required environment variable {self.env_var} is not set"
60 )
61 return self.default
62 return self._convert_value(raw)
64 @abstractmethod
65 def _convert_value(self, raw: str) -> T:
66 """Convert raw string value to the appropriate type."""
67 pass
69 def _get_raw_value(self) -> Optional[str]:
70 """Get raw string value from environment.
72 Checks the canonical env var first. If not set and a deprecated
73 alias is configured, falls back to the deprecated name with a
74 warning guiding users to migrate.
75 """
76 value = os.environ.get(self.env_var)
77 if value is not None:
78 return value
80 if self.deprecated_env_var:
81 deprecated_value = os.environ.get(self.deprecated_env_var)
82 if deprecated_value is not None:
83 logger.warning(
84 f"Environment variable '{self.deprecated_env_var}' is deprecated "
85 f"and will be removed in a future release. "
86 f"Please use '{self.env_var}' instead."
87 )
88 return deprecated_value
90 return None
92 @property
93 def is_set(self) -> bool:
94 """Check if the environment variable is set."""
95 return self.env_var in os.environ
97 def __repr__(self) -> str:
98 """String representation for debugging."""
99 return f"{self.__class__.__name__}(key='{self.key}', env_var='{self.env_var}')"
102class BooleanSetting(EnvSetting[bool]):
103 """Boolean environment setting."""
105 def __init__(
106 self,
107 key: str,
108 description: str,
109 default: bool = False,
110 deprecated_env_var: Optional[str] = None,
111 ):
112 super().__init__(
113 key, description, default, deprecated_env_var=deprecated_env_var
114 )
116 def _convert_value(self, raw: str) -> bool:
117 """Convert string to boolean."""
118 return raw.lower() in ("true", "1", "yes", "on", "enabled")
121class StringSetting(EnvSetting[str]):
122 """String environment setting."""
124 def __init__(
125 self,
126 key: str,
127 description: str,
128 default: Optional[str] = None,
129 required: bool = False,
130 deprecated_env_var: Optional[str] = None,
131 ):
132 super().__init__(
133 key,
134 description,
135 default,
136 required,
137 deprecated_env_var=deprecated_env_var,
138 )
140 def _convert_value(self, raw: str) -> str:
141 """Return string value as-is."""
142 return raw
145class IntegerSetting(EnvSetting[int]):
146 """Integer environment setting."""
148 def __init__(
149 self,
150 key: str,
151 description: str,
152 default: Optional[int] = None,
153 min_value: Optional[int] = None,
154 max_value: Optional[int] = None,
155 deprecated_env_var: Optional[str] = None,
156 ):
157 super().__init__(
158 key, description, default, deprecated_env_var=deprecated_env_var
159 )
160 self.min_value = min_value
161 self.max_value = max_value
163 def _convert_value(self, raw: str) -> Optional[int]:
164 """Convert string to integer with validation."""
165 try:
166 value = int(raw)
167 except ValueError:
168 logger.warning(
169 f"Invalid integer value '{raw}' for {self.env_var}, using default: {self.default}"
170 )
171 return self.default
173 if self.min_value is not None and value < self.min_value:
174 raise ValueError(
175 f"{self.env_var} value {value} is below minimum {self.min_value}"
176 )
177 if self.max_value is not None and value > self.max_value:
178 raise ValueError(
179 f"{self.env_var} value {value} is above maximum {self.max_value}"
180 )
181 return value
184class PathSetting(StringSetting):
185 """Path environment setting with validation."""
187 def __init__(
188 self,
189 key: str,
190 description: str,
191 default: Optional[str] = None,
192 must_exist: bool = False,
193 create_if_missing: bool = False,
194 ):
195 super().__init__(key, description, default)
196 self.must_exist = must_exist
197 self.create_if_missing = create_if_missing
199 def get_value(self) -> Optional[str]:
200 """Get path value with optional validation/creation."""
201 path_str = super().get_value()
202 if path_str is None:
203 return None
205 # Use pathlib for path operations
206 path = Path(path_str).expanduser()
207 # Expand environment variables manually since pathlib doesn't have expandvars
208 # Note: os.path.expandvars is kept here as there's no pathlib equivalent
209 # noqa: PLR0402 - Suppress pathlib check for this line
210 path_str = os.path.expandvars(str(path))
211 path = Path(path_str).resolve()
213 if self.create_if_missing and not path.exists():
214 try:
215 path.mkdir(parents=True, exist_ok=True)
216 except OSError:
217 logger.warning("Failed to create directory")
218 elif self.must_exist and not path.exists():
219 # Only raise if explicitly required to exist
220 raise ValueError(
221 f"Path {path} specified in {self.env_var} does not exist"
222 )
224 return str(path)
227class SecretSetting(StringSetting):
228 """Secret/sensitive environment setting."""
230 def __init__(
231 self,
232 key: str,
233 description: str,
234 default: Optional[str] = None,
235 required: bool = False,
236 ):
237 super().__init__(key, description, default, required)
239 def __repr__(self) -> str:
240 """Hide the value in string representation."""
241 return f"SecretSetting(key='{self.key}', value=***)"
243 def __str__(self) -> str:
244 """Hide the value in string conversion."""
245 value = "SET" if self.is_set else "NOT SET"
246 return f"{self.key}=<{value}>"
249class EnumSetting(EnvSetting[str]):
250 """Enum-style setting with allowed values."""
252 def __init__(
253 self,
254 key: str,
255 description: str,
256 allowed_values: Set[str],
257 default: Optional[str] = None,
258 case_sensitive: bool = False,
259 deprecated_env_var: Optional[str] = None,
260 ):
261 super().__init__(
262 key, description, default, deprecated_env_var=deprecated_env_var
263 )
264 self.allowed_values = allowed_values
265 self.case_sensitive = case_sensitive
267 # Store lowercase versions for case-insensitive comparison
268 if not case_sensitive:
269 self._allowed_lower = {v.lower() for v in allowed_values}
270 # Create a mapping from lowercase to original case
271 self._canonical_map = {v.lower(): v for v in allowed_values}
273 def _convert_value(self, raw: str) -> str:
274 """Convert and validate value against allowed values."""
275 if self.case_sensitive:
276 if raw not in self.allowed_values:
277 raise ValueError(
278 f"{self.env_var} value '{raw}' not in allowed values: {self.allowed_values}"
279 )
280 return raw
281 # Case-insensitive matching
282 raw_lower = raw.lower()
283 if raw_lower not in self._allowed_lower:
284 raise ValueError(
285 f"{self.env_var} value '{raw}' not in allowed values: {self.allowed_values}"
286 )
287 # Return the canonical version (from allowed_values)
288 return self._canonical_map[raw_lower]
291class SettingsRegistry:
292 """Registry for all environment settings."""
294 def __init__(self):
295 self._settings: Dict[str, EnvSetting] = {}
296 self._categories: Dict[str, List[EnvSetting]] = {}
298 def register_category(self, category: str, settings: List[EnvSetting]):
299 """Register a category of settings."""
300 self._categories[category] = settings
301 for setting in settings:
302 self._settings[setting.key] = setting
304 def get(self, key: str, default: Optional[Any] = None) -> Any:
305 """
306 Get a setting value.
308 Args:
309 key: Setting key (e.g., "testing.test_mode")
310 default: Default value if not set or on error
312 Returns:
313 Setting value or default
314 """
315 setting = self._settings.get(key)
316 if not setting:
317 return default
319 try:
320 value = setting.get_value()
321 # Use provided default if setting returns None
322 return value if value is not None else default
323 except ValueError:
324 logger.warning(
325 "Validation error for setting '{}', using default", key
326 )
327 return default
329 def get_setting_object(self, key: str) -> Optional[EnvSetting]:
330 """Get the setting object itself for introspection."""
331 return self._settings.get(key)
333 def is_env_only(self, key: str) -> bool:
334 """Check if a key is an env-only setting."""
335 return key in self._settings
337 def get_env_var(self, key: str) -> Optional[str]:
338 """Get the environment variable name for a key."""
339 setting = self._settings.get(key)
340 return setting.env_var if setting else None
342 def get_all_env_vars(self) -> Dict[str, str]:
343 """Get all environment variables and descriptions."""
344 return {
345 setting.env_var: setting.description
346 for setting in self._settings.values()
347 }
349 def get_category_settings(self, category: str) -> List[EnvSetting]:
350 """Get all settings in a category."""
351 return self._categories.get(category, [])
353 def get_bootstrap_vars(self) -> Dict[str, str]:
354 """Get bootstrap environment variables (bootstrap + db_config)."""
355 result = {}
356 for category in ["bootstrap", "db_config"]:
357 for setting in self._categories.get(category, []):
358 result[setting.env_var] = setting.description
359 return result
361 def get_testing_vars(self) -> Dict[str, str]:
362 """Get testing environment variables."""
363 result = {}
364 for setting in self._categories.get("testing", []):
365 result[setting.env_var] = setting.description
366 return result
368 def list_all_settings(self) -> List[str]:
369 """List all registered setting keys."""
370 return list(self._settings.keys())
373# Export list for better IDE discovery
374__all__ = [
375 "EnvSetting",
376 "BooleanSetting",
377 "StringSetting",
378 "IntegerSetting",
379 "PathSetting",
380 "SecretSetting",
381 "EnumSetting",
382 "SettingsRegistry",
383]