Coverage for src/local_deep_research/settings/env_settings.py: 98%
211 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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 (the plaintext-SQLite fallback policy) 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
24import re
25from pathlib import Path
26from typing import Any, Dict, Optional, List, Set, TypeVar, Generic
27from abc import ABC, abstractmethod
28from loguru import logger
30from .exceptions import (
31 EnvironmentPathNotFoundError,
32 EnvironmentValueRangeError,
33 InvalidEnvironmentValueError,
34 MissingEnvironmentVariableError,
35 SettingsValueError,
36)
38T = TypeVar("T")
41class EnvSetting(ABC, Generic[T]):
42 """Base class for all environment settings."""
44 def __init__(
45 self,
46 key: str,
47 description: str,
48 default: Optional[T] = None,
49 required: bool = False,
50 deprecated_env_var: Optional[str] = None,
51 ):
52 self.key = key
53 # Auto-generate env_var from key
54 # e.g., "testing.test_mode" -> "LDR_TESTING_TEST_MODE"
55 self.env_var = "LDR_" + key.upper().replace(".", "_")
56 self.description = description
57 self.default = default
58 self.required = required
59 self.deprecated_env_var = deprecated_env_var
61 def get_value(
62 self, *, strict: bool = False, check_env: bool = True
63 ) -> Optional[T]:
64 """Get the value from environment with type conversion."""
65 raw = self._get_raw_value() if check_env else None
66 if raw is None:
67 if self.required and self.default is None:
68 raise MissingEnvironmentVariableError(self.env_var)
69 return self.default
70 if strict:
71 return self._convert_value(raw, strict=True)
72 return self._convert_value(raw)
74 @abstractmethod
75 def _convert_value(self, raw: str, *, strict: bool = False) -> T:
76 """Convert raw string value to the appropriate type."""
77 pass
79 def _get_raw_value(self) -> Optional[str]:
80 """Get raw string value from environment.
82 Checks the canonical env var first. If not set and a deprecated
83 alias is configured, falls back to the deprecated name with a
84 warning guiding users to migrate.
85 """
86 value = os.environ.get(self.env_var)
87 if value is not None:
88 return value
90 if self.deprecated_env_var:
91 deprecated_value = os.environ.get(self.deprecated_env_var)
92 if deprecated_value is not None:
93 logger.warning(
94 f"Environment variable '{self.deprecated_env_var}' is deprecated "
95 f"and will be removed in a future release. "
96 f"Please use '{self.env_var}' instead."
97 )
98 return deprecated_value
100 return None
102 @property
103 def is_set(self) -> bool:
104 """Check if the environment variable is set."""
105 return self.env_var in os.environ
107 def __repr__(self) -> str:
108 """String representation for debugging."""
109 return f"{self.__class__.__name__}(key='{self.key}', env_var='{self.env_var}')"
112class BooleanSetting(EnvSetting[bool]):
113 """Boolean environment setting."""
115 def __init__(
116 self,
117 key: str,
118 description: str,
119 default: bool = False,
120 deprecated_env_var: Optional[str] = None,
121 ):
122 super().__init__(
123 key, description, default, deprecated_env_var=deprecated_env_var
124 )
126 def _convert_value(self, raw: str, *, strict: bool = False) -> bool:
127 """Convert string to boolean."""
128 if strict:
129 normalized = raw.strip().lower()
130 if normalized in ("true", "1", "yes", "on", "enabled"):
131 return True
132 if normalized in ("false", "0", "no", "off", "disabled", ""):
133 return False
134 raise ValueError("Expected a boolean token")
135 return raw.lower() in ("true", "1", "yes", "on", "enabled")
138class StringSetting(EnvSetting[str]):
139 """String environment setting."""
141 def __init__(
142 self,
143 key: str,
144 description: str,
145 default: Optional[str] = None,
146 required: bool = False,
147 deprecated_env_var: Optional[str] = None,
148 ):
149 super().__init__(
150 key,
151 description,
152 default,
153 required,
154 deprecated_env_var=deprecated_env_var,
155 )
157 def _convert_value(self, raw: str, *, strict: bool = False) -> str:
158 """Return string value as-is."""
159 return raw
162#: Plain ASCII decimal integer, optionally negative. Deliberately NOT what
163#: ``int()`` accepts: ``int`` also takes underscore separators (``"1_0"``
164#: -> 10), a leading ``"+"``, surrounding whitespace, and any Unicode
165#: decimal digit (``"٣"`` -> 3). For a setting that defines a trust
166#: boundary those spellings are a review hazard -- ``"1_0"`` reads as one
167#: hop and means ten -- so ``strict_digits`` settings reject them. The
168#: optional ``-`` is matched (rather than rejected outright) so a negative
169#: value still reaches the ``min_value`` check and gets the specific
170#: "below minimum" range error instead of a generic parse error.
171_STRICT_INTEGER_LITERAL = re.compile(r"-?[0-9]+")
174class IntegerSetting(EnvSetting[int]):
175 """Integer environment setting.
177 ``raise_on_validation_error`` makes malformed and out-of-range values
178 propagate instead of falling back through ``SettingsRegistry``.
180 ``strict_digits`` restricts accepted spellings to a plain ASCII
181 decimal literal (see ``_STRICT_INTEGER_LITERAL``).
183 An empty value is treated as UNSET in every mode: ``LDR_X=""`` falls
184 back to the default rather than failing, because orchestrators
185 (Compose, k8s manifests, Unraid templates) routinely emit an empty
186 string for an unconfigured variable -- the repo-wide contract pinned
187 by ``tests/test_config_env_var_contracts.py::
188 test_empty_env_var_is_treated_as_unset_not_as_false``.
189 """
191 def __init__(
192 self,
193 key: str,
194 description: str,
195 default: Optional[int] = None,
196 min_value: Optional[int] = None,
197 max_value: Optional[int] = None,
198 deprecated_env_var: Optional[str] = None,
199 raise_on_validation_error: bool = False,
200 strict_digits: bool = False,
201 ):
202 super().__init__(
203 key,
204 description,
205 default,
206 deprecated_env_var=deprecated_env_var,
207 )
208 self.min_value = min_value
209 self.max_value = max_value
210 self.raise_on_validation_error = raise_on_validation_error
211 self.strict_digits = strict_digits
213 def _convert_value(
214 self, raw: str, *, strict: bool = False
215 ) -> Optional[int]:
216 """Convert string to integer with validation."""
217 candidate = raw.strip()
218 if not candidate:
219 logger.warning(
220 f"Empty value for {self.env_var}, treating it as unset and using default: {self.default}"
221 )
222 return self.default
224 value: Optional[int] = None
225 parse_error: Optional[ValueError] = None
226 if self.strict_digits and not _STRICT_INTEGER_LITERAL.fullmatch(
227 candidate
228 ):
229 parse_error = ValueError(
230 f"invalid literal for int() with base 10: {raw!r}"
231 )
232 else:
233 try:
234 value = int(candidate)
235 except ValueError as exc:
236 parse_error = exc
238 if parse_error is not None:
239 if strict:
240 raise parse_error
241 if self.raise_on_validation_error:
242 raise InvalidEnvironmentValueError(
243 self.env_var, raw, expected_type="integer"
244 ) from parse_error
245 logger.warning(
246 f"Invalid integer value '{raw}' for {self.env_var}, using default: {self.default}"
247 )
248 return self.default
250 if self.min_value is not None and value < self.min_value:
251 raise EnvironmentValueRangeError(
252 self.env_var, value, min_val=self.min_value
253 )
254 if self.max_value is not None and value > self.max_value:
255 raise EnvironmentValueRangeError(
256 self.env_var, value, max_val=self.max_value
257 )
258 return value
261class PathSetting(StringSetting):
262 """Path environment setting with validation."""
264 def __init__(
265 self,
266 key: str,
267 description: str,
268 default: Optional[str] = None,
269 must_exist: bool = False,
270 create_if_missing: bool = False,
271 ):
272 super().__init__(key, description, default)
273 self.must_exist = must_exist
274 self.create_if_missing = create_if_missing
276 def get_value(
277 self, *, strict: bool = False, check_env: bool = True
278 ) -> Optional[str]:
279 """Get path value with optional validation/creation."""
280 path_str = super().get_value(strict=strict, check_env=check_env)
281 if path_str is None or not check_env:
282 return path_str
283 if strict and not path_str.strip():
284 raise ValueError("Expected a nonempty path")
286 # Use pathlib for path operations
287 path = Path(path_str).expanduser()
288 # Expand environment variables manually since pathlib doesn't have expandvars
289 # Note: os.path.expandvars is kept here as there's no pathlib equivalent
290 # noqa: PLR0402 - Suppress pathlib check for this line
291 path_str = os.path.expandvars(str(path))
292 path = Path(path_str).resolve()
294 if (
295 strict
296 and self.create_if_missing
297 and path.exists()
298 and not path.is_dir()
299 ):
300 raise ValueError("Expected a directory")
301 if self.create_if_missing and not path.exists():
302 # Lazy import: this module is loaded very early (bootstrap
303 # env-only settings), before the database and app are
304 # initialized, so importing security at module load time
305 # risks an import cycle.
306 from ..security.directory_creation import (
307 DirectoryCreationSecurityError,
308 create_directory,
309 )
311 try:
312 create_directory(
313 path,
314 context=f"env-configured path setting '{self.key}'",
315 )
316 except DirectoryCreationSecurityError:
317 if strict: 317 ↛ 321line 317 didn't jump to line 321 because the condition on line 317 was always true
318 raise ValueError(
319 "Could not prepare configured path"
320 ) from None
321 raise
322 except OSError:
323 if strict: 323 ↛ 325line 323 didn't jump to line 325 because the condition on line 323 was always true
324 raise
325 logger.warning("Failed to create directory")
326 elif self.must_exist and not path.exists():
327 # Only raise if explicitly required to exist
328 raise EnvironmentPathNotFoundError(self.env_var, path)
330 return str(path)
333class SecretSetting(StringSetting):
334 """Secret/sensitive environment setting."""
336 def __init__(
337 self,
338 key: str,
339 description: str,
340 default: Optional[str] = None,
341 required: bool = False,
342 ):
343 super().__init__(key, description, default, required)
345 def __repr__(self) -> str:
346 """Hide the value in string representation."""
347 return f"SecretSetting(key='{self.key}', value=***)"
349 def __str__(self) -> str:
350 """Hide the value in string conversion."""
351 value = "SET" if self.is_set else "NOT SET"
352 return f"{self.key}=<{value}>"
355class EnumSetting(EnvSetting[str]):
356 """Enum-style setting with allowed values."""
358 def __init__(
359 self,
360 key: str,
361 description: str,
362 allowed_values: Set[str],
363 default: Optional[str] = None,
364 case_sensitive: bool = False,
365 deprecated_env_var: Optional[str] = None,
366 ):
367 super().__init__(
368 key, description, default, deprecated_env_var=deprecated_env_var
369 )
370 self.allowed_values = allowed_values
371 self.case_sensitive = case_sensitive
373 # Store lowercase versions for case-insensitive comparison
374 if not case_sensitive:
375 self._allowed_lower = {v.lower() for v in allowed_values}
376 # Create a mapping from lowercase to original case
377 self._canonical_map = {v.lower(): v for v in allowed_values}
379 def _convert_value(self, raw: str, *, strict: bool = False) -> str:
380 """Convert and validate value against allowed values."""
381 if self.case_sensitive:
382 if raw not in self.allowed_values:
383 raise InvalidEnvironmentValueError(
384 self.env_var, raw, list(self.allowed_values)
385 )
386 return raw
387 # Case-insensitive matching
388 raw_lower = raw.lower()
389 if raw_lower not in self._allowed_lower:
390 raise InvalidEnvironmentValueError(
391 self.env_var, raw, list(self.allowed_values)
392 )
393 # Return the canonical version (from allowed_values)
394 return self._canonical_map[raw_lower]
397class SettingsRegistry:
398 """Registry for all environment settings."""
400 def __init__(self):
401 self._settings: Dict[str, EnvSetting] = {}
402 self._categories: Dict[str, List[EnvSetting]] = {}
404 def register_category(self, category: str, settings: List[EnvSetting]):
405 """Register a category of settings."""
406 self._categories[category] = settings
407 for setting in settings:
408 self._settings[setting.key] = setting
410 def get(self, key: str, default: Optional[Any] = None) -> Any:
411 """
412 Get a setting value.
414 Args:
415 key: Setting key (e.g., "testing.test_mode")
416 default: Default value if not set or on error
418 Returns:
419 Setting value or default
420 """
421 setting = self._settings.get(key)
422 if not setting:
423 return default
425 try:
426 value = setting.get_value()
427 # Use provided default if setting returns None
428 return value if value is not None else default
429 except ValueError:
430 if (
431 isinstance(setting, IntegerSetting)
432 and setting.raise_on_validation_error
433 ):
434 raise
435 logger.warning(
436 "Validation error for setting '{}', using default", key
437 )
438 return default
440 def get_strict(
441 self,
442 key: str,
443 default: Optional[Any] = None,
444 *,
445 check_env: bool = True,
446 ) -> Any:
447 """Read for a settings manager without hiding invalid configuration.
449 Missing optional settings retain their registered or caller default.
450 Disabling environment lookup also disables path resolution/creation.
451 The legacy bootstrap ``get`` API retains its historical behavior.
452 """
453 setting = self._settings.get(key)
454 if setting is None:
455 return default
457 source = "default"
458 if check_env and (
459 setting.env_var in os.environ
460 or (
461 setting.deprecated_env_var is not None
462 and setting.deprecated_env_var in os.environ
463 )
464 ):
465 source = "environment"
467 # A custom converter has no strict-read contract. Do not call it and
468 # accidentally accept a fallback or suppress its validation failure.
469 reasons = {
470 BooleanSetting: "expected a boolean token",
471 IntegerSetting: "expected an integer within the allowed range",
472 EnumSetting: "expected one of the declared choices",
473 PathSetting: "could not resolve or prepare the configured path",
474 StringSetting: "required value is missing",
475 SecretSetting: "required value is missing",
476 }
477 reason = reasons.get(type(setting))
478 if reason is None:
479 raise SettingsValueError(
480 key,
481 source=source,
482 reason="unsupported environment setting type",
483 )
485 try:
486 value = setting.get_value(strict=True, check_env=check_env)
487 except (ValueError, TypeError, OSError, RuntimeError):
488 # Converters and filesystem errors can contain configured secrets
489 # or paths. Raise outside the handler so no raw exception is kept
490 # as the exception context or emitted in the traceback.
491 pass
492 else:
493 return value if value is not None else default
494 raise SettingsValueError(key, source=source, reason=reason)
496 def get_setting_object(self, key: str) -> Optional[EnvSetting]:
497 """Get the setting object itself for introspection."""
498 return self._settings.get(key)
500 def is_env_only(self, key: str) -> bool:
501 """Check if a key is an env-only setting."""
502 return key in self._settings
504 def get_env_var(self, key: str) -> Optional[str]:
505 """Get the environment variable name for a key."""
506 setting = self._settings.get(key)
507 return setting.env_var if setting else None
509 def get_all_env_vars(self) -> Dict[str, str]:
510 """Get all environment variables and descriptions."""
511 return {
512 setting.env_var: setting.description
513 for setting in self._settings.values()
514 }
516 def get_category_settings(self, category: str) -> List[EnvSetting]:
517 """Get all settings in a category."""
518 return self._categories.get(category, [])
520 def get_bootstrap_vars(self) -> Dict[str, str]:
521 """Get bootstrap environment variables (bootstrap + db_config)."""
522 result = {}
523 for category in ["bootstrap", "db_config"]:
524 for setting in self._categories.get(category, []):
525 result[setting.env_var] = setting.description
526 return result
528 def get_testing_vars(self) -> Dict[str, str]:
529 """Get testing environment variables."""
530 result = {}
531 for setting in self._categories.get("testing", []):
532 result[setting.env_var] = setting.description
533 return result
535 def list_all_settings(self) -> List[str]:
536 """List all registered setting keys."""
537 return list(self._settings.keys())
540# Export list for better IDE discovery
541__all__ = [
542 "EnvSetting",
543 "BooleanSetting",
544 "StringSetting",
545 "IntegerSetting",
546 "PathSetting",
547 "SecretSetting",
548 "EnumSetting",
549 "SettingsRegistry",
550]