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
« 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.
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
29from .exceptions import (
30 EnvironmentPathNotFoundError,
31 EnvironmentValueRangeError,
32 InvalidEnvironmentValueError,
33 MissingEnvironmentVariableError,
34)
36T = TypeVar("T")
39class EnvSetting(ABC, Generic[T]):
40 """Base class for all environment settings."""
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
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)
68 @abstractmethod
69 def _convert_value(self, raw: str) -> T:
70 """Convert raw string value to the appropriate type."""
71 pass
73 def _get_raw_value(self) -> Optional[str]:
74 """Get raw string value from environment.
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
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
94 return None
96 @property
97 def is_set(self) -> bool:
98 """Check if the environment variable is set."""
99 return self.env_var in os.environ
101 def __repr__(self) -> str:
102 """String representation for debugging."""
103 return f"{self.__class__.__name__}(key='{self.key}', env_var='{self.env_var}')"
106class BooleanSetting(EnvSetting[bool]):
107 """Boolean environment setting."""
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 )
120 def _convert_value(self, raw: str) -> bool:
121 """Convert string to boolean."""
122 return raw.lower() in ("true", "1", "yes", "on", "enabled")
125class StringSetting(EnvSetting[str]):
126 """String environment setting."""
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 )
144 def _convert_value(self, raw: str) -> str:
145 """Return string value as-is."""
146 return raw
149class IntegerSetting(EnvSetting[int]):
150 """Integer environment setting."""
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
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
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
188class PathSetting(StringSetting):
189 """Path environment setting with validation."""
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
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
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()
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)
226 return str(path)
229class SecretSetting(StringSetting):
230 """Secret/sensitive environment setting."""
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)
241 def __repr__(self) -> str:
242 """Hide the value in string representation."""
243 return f"SecretSetting(key='{self.key}', value=***)"
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}>"
251class EnumSetting(EnvSetting[str]):
252 """Enum-style setting with allowed values."""
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
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}
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]
293class SettingsRegistry:
294 """Registry for all environment settings."""
296 def __init__(self):
297 self._settings: Dict[str, EnvSetting] = {}
298 self._categories: Dict[str, List[EnvSetting]] = {}
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
306 def get(self, key: str, default: Optional[Any] = None) -> Any:
307 """
308 Get a setting value.
310 Args:
311 key: Setting key (e.g., "testing.test_mode")
312 default: Default value if not set or on error
314 Returns:
315 Setting value or default
316 """
317 setting = self._settings.get(key)
318 if not setting:
319 return default
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
331 def get_setting_object(self, key: str) -> Optional[EnvSetting]:
332 """Get the setting object itself for introspection."""
333 return self._settings.get(key)
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
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
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 }
351 def get_category_settings(self, category: str) -> List[EnvSetting]:
352 """Get all settings in a category."""
353 return self._categories.get(category, [])
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
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
370 def list_all_settings(self) -> List[str]:
371 """List all registered setting keys."""
372 return list(self._settings.keys())
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]