Coverage for src/local_deep_research/settings/manager.py: 92%
532 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1import functools
2import json
3import os
4import threading
5import time
6from pathlib import Path
7from typing import Any, Callable, Dict, List, Optional, Union
9from loguru import logger
10from sqlalchemy import and_, func, or_
11from sqlalchemy.exc import SQLAlchemyError
12from sqlalchemy.orm import Session
14from .. import defaults
15from ..__version__ import __version__ as package_version
16from ..database.models import Setting, SettingType
17from ..utilities.sql_utils import escape_like
18from ..utilities.type_utils import to_bool
19from ..web.models.settings import (
20 AppSetting,
21 BaseSetting,
22 ChatSetting,
23 LLMSetting,
24 ReportSetting,
25 SearchSetting,
26)
27from .base import ISettingsManager
28from .env_registry import registry as env_registry
31def parse_boolean(value: Any) -> bool:
32 """
33 Convert various representations to boolean using HTML checkbox semantics.
35 This function handles form values, JSON booleans, and environment variables,
36 ensuring consistent behavior across client and server.
38 **HTML Checkbox Semantics** (INTENTIONAL DESIGN):
39 - **Any value present (except explicit false) = checked = True**
40 - This matches standard HTML form behavior where checkbox presence indicates checked state
41 - In HTML forms, checkboxes send a value when checked, nothing when unchecked
43 **Examples**:
44 parse_boolean("on") # True - standard HTML checkbox value
45 parse_boolean("true") # True - explicit true
46 parse_boolean("1") # True - numeric true
47 parse_boolean("enabled") # True - any non-empty string
48 parse_boolean("disabled") # True - INTENTIONAL: any string = checkbox was checked!
49 parse_boolean("custom") # True - custom checkbox value
51 parse_boolean("false") # False - explicit false
52 parse_boolean("off") # False - explicit false
53 parse_boolean("0") # False - explicit false
54 parse_boolean("") # False - empty string = unchecked
55 parse_boolean(None) # False - missing = unchecked
57 **Why "disabled" returns True**:
58 This is NOT a bug! If a checkbox sends the value "disabled", it means the checkbox
59 was checked (present in form data). The actual string content doesn't matter for
60 HTML checkboxes - only presence vs absence matters.
62 Args:
63 value: Value to convert to boolean. Accepts strings, booleans, or None.
65 Returns:
66 bool: True for truthy values (any non-empty string except explicit false);
67 False for falsy values ('off', 'false', '0', '', 'no', False, None)
69 Note:
70 This function implements HTML form semantics, NOT generic boolean parsing.
71 See tests/settings/test_boolean_parsing.py for comprehensive test coverage.
72 """
73 # Constants for boolean value parsing
74 FALSY_VALUES = ("off", "false", "0", "", "no")
76 # Handle already-boolean values
77 if isinstance(value, bool):
78 return value
80 # Handle None (missing values)
81 if value is None:
82 return False
84 # Handle string values
85 if isinstance(value, str):
86 value_lower = value.lower().strip()
87 # Explicitly falsy values (empty string, false-like values)
88 if value_lower in FALSY_VALUES:
89 return False
90 # Any other non-empty string = True (HTML checkbox semantics)
91 return True
93 # For other types (numbers, lists, etc.), use Python's bool conversion
94 return bool(value)
97def _parse_number(x):
98 """Parse number, returning int if it's a whole number, otherwise float."""
99 f = float(x)
100 if f.is_integer():
101 return int(f)
102 return f
105def _parse_json_value(x):
106 """Parse JSON ui_element values.
108 DB values (via SQLAlchemy JSON column) arrive as Python objects already.
109 Form POST and env var overrides arrive as raw strings and need parsing.
110 For example, a textarea containing ``["general"]`` arrives as the string
111 ``'[\\r\\n "general"\\r\\n]'`` which must be decoded into a list.
112 """
113 if isinstance(x, str):
114 stripped = x.strip()
115 if stripped:
116 try:
117 return json.loads(stripped)
118 except (json.JSONDecodeError, ValueError, RecursionError):
119 logger.warning("Failed to parse JSON value, returning raw")
120 return x
121 return x
124def _parse_multiselect(x):
125 """Parse multiselect value, handling both lists and strings.
127 DB values (via SQLAlchemy JSON column) arrive as Python lists already.
128 Env var overrides arrive as strings and need parsing — either as JSON
129 arrays (e.g. '["markdown","latex"]') or comma-separated values
130 (e.g. 'markdown,latex').
131 """
132 if isinstance(x, list):
133 return x
134 if isinstance(x, str):
135 stripped = x.strip()
136 if stripped.startswith("["):
137 try:
138 parsed = json.loads(stripped)
139 if isinstance(parsed, list): 139 ↛ 144line 139 didn't jump to line 144 because the condition on line 139 was always true
140 return parsed
141 except (json.JSONDecodeError, ValueError):
142 pass
143 # Comma-separated fallback
144 return [item.strip() for item in stripped.split(",") if item.strip()]
145 return x
148def _filter_setting_columns(data: dict) -> dict:
149 """Filter a dict to only keys that are valid Setting model columns.
151 Prevents crashes when default_settings.json contains keys not present
152 as columns on the Setting model (e.g. future flags).
153 """
154 valid_columns = {c.name for c in Setting.__table__.columns}
155 return {k: v for k, v in data.items() if k in valid_columns}
158_POLICY_AUDIT_KEYS = frozenset(
159 {
160 "llm.require_local_endpoint",
161 "llm.allowed_local_hostnames",
162 "embeddings.require_local",
163 }
164)
167def _is_policy_setting(key: str) -> bool:
168 """Return True for security-relevant setting keys that need an
169 audit-log entry on change. Scope is intentionally narrow so this
170 audit hook doesn't widen into general settings-change logging.
171 """
172 if key.startswith("policy."):
173 return True
174 return key in _POLICY_AUDIT_KEYS
177def is_valid_setting_key(key: Any) -> bool:
178 """Return True if *key* is a well-formed setting key.
180 A valid key is a non-empty string of dot-separated segments where each
181 segment is non-empty and contains no whitespace. This rejects the
182 malformed keys that corrupt prefix/namespace lookups (see #4840) — a
183 trailing dot ``"foo."``, a leading dot ``".foo"``, an empty segment
184 ``"foo..bar"``, a blank key — as well as keys carrying whitespace
185 (``" foo"``, ``"foo. bar"``, ``"llm.o k"``), which would also break the
186 ``key.split(".")`` → ``LDR_...`` env-var mapping. A malformed row makes
187 ``get_setting("foo")`` return a ``{"": value}`` wrapper dict, which the
188 UI renders as ``[object Object]``.
190 Used on both sides: writes reject new malformed keys, and both read paths
191 (``SettingsManager.get_setting`` on the DB and ``get_setting_from_snapshot``
192 on a settings snapshot) filter malformed rows out of a prefix read so a
193 stray legacy row can never wrap a leaf value into an ``[object Object]``
194 dict.
195 """
196 if not isinstance(key, str) or not key:
197 return False
198 return all(
199 seg and not any(c.isspace() for c in seg) for seg in key.split(".")
200 )
203def _infer_ui_element(value: Any, current: str = "text") -> str:
204 """Infer the appropriate ui_element string from a Python value's type.
206 Args:
207 value: The value to infer the ui_element from.
208 current: The existing ui_element. If it is already something more
209 specific than ``"text"``, it is kept as-is.
210 """
211 if current != "text":
212 return current
213 if isinstance(value, bool):
214 return "checkbox"
215 if isinstance(value, (int, float)):
216 return "number"
217 if isinstance(value, (list, dict)):
218 return "json"
219 return "text"
222# Default categories for each typed setting prefix, used by the self-heal
223# block in set_setting() when a row's type column doesn't match its key
224# prefix. The values mirror the canonical strings in
225# web/routes/settings_routes.py: a legacy chat.* row with type=APP and a
226# stale category gets repointed to type=CHAT + category="chat" on next
227# save. We use the most-general category per prefix here — sub-classifying
228# llm_general vs llm_parameters depends on the specific key, but the
229# self-heal only fires when the row was already mis-typed, so over-
230# generalizing the category is preferable to leaving it stale.
231_INFERRED_CATEGORY: Dict[str, str] = {
232 "llm.": "llm_general",
233 "search.": "search_general",
234 "report.": "report_parameters",
235 "database.": "database_parameters",
236 "chat.": "chat",
237}
240UI_ELEMENT_TO_SETTING_TYPE: Dict[str, Callable[..., Any]] = {
241 "text": str,
242 "json": _parse_json_value,
243 "password": str,
244 "select": str,
245 "number": _parse_number,
246 "range": _parse_number, # Same behavior as number for consistency
247 "checkbox": parse_boolean,
248 "textarea": str,
249 "multiselect": _parse_multiselect,
250}
253def get_typed_setting_value(
254 key: str,
255 value: Any,
256 ui_element: str,
257 default: Any = None,
258 check_env: bool = True,
259) -> Any:
260 """
261 Extracts the value for a particular setting, ensuring that it has the
262 correct type.
264 Args:
265 key: The setting key.
266 value: The setting value from the database.
267 ui_element: The setting UI element ID.
268 default: Default value to return if the value of the setting is
269 invalid.
270 check_env: If true, it will check the environment variable for
271 this setting before reading from the DB.
273 Returns:
274 The value of the setting.
276 """
277 setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None)
278 if setting_type is None:
279 logger.warning(
280 "Got unknown type {} for setting {}, returning default value.",
281 ui_element,
282 key,
283 )
284 return default
286 # Check environment variable first (highest priority).
287 if check_env:
288 env_value = check_env_setting(key)
289 if env_value is not None:
290 try:
291 return setting_type(env_value)
292 except ValueError:
293 logger.warning(
294 "Setting {} has invalid value {}. Falling back to DB.",
295 key,
296 env_value,
297 )
299 # If value is None (not in database), return default.
300 if value is None:
301 return default
303 # Read from the database.
304 try:
305 return setting_type(value)
306 except (ValueError, TypeError):
307 logger.warning(
308 "Setting {} has invalid value {}. Returning default.",
309 key,
310 value,
311 )
312 return default
315def check_env_setting(key: str) -> str | None:
316 """
317 Checks environment variables for a particular setting.
319 Args:
320 key: The database key for the setting.
322 Returns:
323 The setting from the environment variables, or None if the variable
324 is not set or is empty.
326 Note:
327 Empty environment variables ("") are treated as unset. This is standard
328 practice across the ecosystem — see CPython's official docs (PYTHON*
329 env vars require "a non-empty string"), botocore PR #1687, Pallets/Click
330 PR #2223, and Vercel Turborepo PR #6929. Orchestration tools like Unraid,
331 Terraform, and Kubernetes manifests often cannot conditionally omit env
332 var declarations, so they pass "" for unconfigured values. Treating ""
333 as unset prevents these empty strings from overriding database defaults.
334 See: https://github.com/LearningCircuit/local-deep-research/pull/3362
336 """
337 env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}"
338 env_value = os.getenv(env_variable_name)
339 # Treat empty string as unset — orchestration tools (Unraid, Terraform, K8s)
340 # often cannot omit env var declarations and pass "" for unconfigured values.
341 if env_value is not None and env_value != "":
342 logger.debug(f"Overriding {key} setting from environment variable.")
343 return env_value
344 if env_value == "":
345 logger.warning(
346 "Environment variable {} is set but empty — "
347 "ignoring it and falling back to DB/default for setting '{}'. "
348 "This is expected on Unraid or Docker templates that create "
349 "all variables even when left blank. To suppress this warning, "
350 "remove the variable from your environment or set a value.",
351 env_variable_name,
352 key,
353 )
354 return None
357class SettingsManager(ISettingsManager):
358 """
359 Manager for handling application settings with database storage and file fallback.
360 Provides methods to get and set settings, with the ability to override settings in memory.
361 """
363 def __init__(
364 self,
365 db_session: Optional[Session] = None,
366 owns_session: bool = False,
367 ):
368 """
369 Initialize the settings manager
371 Args:
372 db_session: SQLAlchemy session for database operations
373 owns_session: If True, close() will close the session.
374 Defaults to False (safe for borrowed sessions). Set to True
375 only when this manager created/owns the session — currently
376 only get_settings_manager() in db_utils.py does this.
377 """
378 self.db_session = db_session
379 self._owns_session = owns_session
380 self._closed = False
381 self.db_first = True # Always prioritize DB settings
383 # Store the thread ID this instance was created in
384 self._creation_thread_id = threading.get_ident()
386 # Initialize settings lock as None - will be checked lazily
387 self.__settings_locked: Optional[bool] = None
389 # Auto-initialize settings if database is empty
390 if self.db_session:
391 self._ensure_settings_initialized()
393 def close(self):
394 """Close the DB session if this manager owns it.
396 Borrowed sessions (owns_session=False) are left open for their
397 owner to close (e.g. Flask teardown closes g.db_session).
398 Safe to call multiple times — subsequent calls are no-ops.
399 """
400 if self._owns_session and self.db_session is not None: 400 ↛ 401line 400 didn't jump to line 401 because the condition on line 400 was never true
401 try:
402 logger.debug("Closing owned DB session in SettingsManager")
403 self.db_session.close()
404 except Exception:
405 logger.warning(
406 "Failed to close SettingsManager DB session — "
407 "connection may leak",
408 )
409 self._closed = True
410 self.db_session = None
412 def _ensure_settings_initialized(self):
413 """Ensure settings are initialized in the database."""
414 # Check if we have any settings at all
415 from ..database.models import Setting
417 if self.db_session is None: 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 raise RuntimeError("Database session is not initialized")
419 settings_count = self.db_session.query(Setting).count()
421 if settings_count == 0:
422 logger.info("No settings found in database, loading defaults")
423 self.load_from_defaults_file(commit=True)
424 logger.info("Default settings loaded successfully")
426 def _check_thread_safety(self):
427 """Check if this instance is being used in the same thread it was created in."""
428 current_thread_id = threading.get_ident()
429 if self.db_session and current_thread_id != self._creation_thread_id:
430 raise RuntimeError(
431 f"SettingsManager instance created in thread {self._creation_thread_id} "
432 f"is being used in thread {current_thread_id}. This is not thread-safe! "
433 f"Create a new SettingsManager instance within the current thread context."
434 )
436 @property
437 def settings_locked(self) -> bool:
438 """Check if settings are locked (lazy evaluation)."""
439 if self.__settings_locked is None:
440 try:
441 self.__settings_locked = self.get_setting(
442 "app.lock_settings", False
443 )
444 if self.settings_locked:
445 logger.info(
446 "Settings are locked. Disabling all settings changes."
447 )
448 except Exception:
449 logger.warning(
450 "Failed to check settings lock status, assuming not locked"
451 )
452 self.__settings_locked = False
453 return bool(self.__settings_locked)
455 @functools.cached_property
456 def default_settings(self) -> Dict[str, Any]:
457 """
458 Returns:
459 The default settings, loaded from JSON files and merged.
460 Automatically discovers and loads all .json files in the defaults
461 directory and its subdirectories.
462 Theme options are dynamically injected from the theme registry.
464 """
465 settings: Dict[str, Any] = {}
467 try:
468 # Get the defaults package path
469 defaults_path = Path(defaults.__file__).parent
471 # Find all JSON files recursively in the defaults directory
472 json_files = sorted(defaults_path.rglob("*.json"))
474 logger.debug(f"Found {len(json_files)} JSON settings files")
476 # Load and merge all JSON files
477 for json_file in json_files:
478 try:
479 with open(json_file, "r", encoding="utf-8-sig") as f:
480 file_settings = json.load(f)
482 # Get relative path for logging
483 relative_path = json_file.relative_to(defaults_path)
485 # Warn about key conflicts
486 conflicts = set(settings.keys()) & set(file_settings.keys())
487 if conflicts:
488 logger.warning(
489 f"Keys {conflicts} from {relative_path} "
490 f"override existing values"
491 )
493 settings.update(file_settings)
494 logger.debug(f"Loaded {relative_path}")
496 except json.JSONDecodeError:
497 logger.exception(f"Invalid JSON in {json_file}")
498 except Exception:
499 logger.warning(f"Could not load {json_file}")
501 except Exception:
502 logger.warning("Error loading settings files")
504 # Inject dynamic theme options from theme registry
505 if "app.theme" in settings:
506 try:
507 from local_deep_research.web.themes import theme_registry
509 settings["app.theme"]["options"] = (
510 theme_registry.get_settings_options()
511 )
512 except ImportError:
513 # Theme registry not available, use static options from JSON
514 pass
516 # Inject search strategy options from code (single source of truth)
517 if "search.search_strategy" in settings:
518 from local_deep_research.constants import get_available_strategies
520 strategies = get_available_strategies()
521 settings["search.search_strategy"]["options"] = [
522 {"label": s["label"], "value": s["name"]} for s in strategies
523 ]
525 logger.debug(f"Loaded {len(settings)} total settings")
526 return settings
528 def __get_typed_setting_value(
529 self,
530 setting: Setting,
531 default: Any = None,
532 check_env: bool = True,
533 ) -> Any:
534 """
535 Extracts the value for a particular setting, ensuring that it has the
536 correct type.
538 Args:
539 setting: The setting to get the value for.
540 default: Default value to return if the value of the setting is
541 invalid.
542 check_env: If true, it will check the environment variable for
543 this setting before reading from the DB.
545 Returns:
546 The value of the setting.
548 """
549 return get_typed_setting_value(
550 str(setting.key),
551 setting.value,
552 str(setting.ui_element),
553 default=default,
554 check_env=check_env,
555 )
557 def __query_settings(self, key: str | None = None) -> List[Setting]:
558 """
559 Abstraction for querying settings that also transparently handles
560 reading the default settings file if the DB is not enabled.
562 Args:
563 key: The key to read. If None, it will read everything.
565 Returns:
566 The settings it queried.
568 """
569 if self.db_session:
570 self._check_thread_safety()
571 query = self.db_session.query(Setting)
572 if key is not None:
573 # This will find exact matches and any subkeys.
574 #
575 # The ``startswith`` arm is intentionally restricted to
576 # subkeys that have AT LEAST one character past the dot
577 # (``key LIKE 'foo.%' AND key != 'foo.'``). The bare
578 # ``LIKE 'foo.%'`` form silently also matches ``'foo.'``
579 # (because ``%`` matches zero or more chars), which would
580 # treat a malformed trailing-dot duplicate row as a
581 # subkey of ``foo`` and cause ``get_setting`` to return
582 # a 2-key dict (``{"": value, "foo": value}``) instead
583 # of the primitive. See embedding-settings page bug.
584 # escape_like: the requested key is user-supplied (e.g. via
585 # /settings/api/bulk?keys[]=...), so escape its LIKE wildcards
586 # or ``keys[]=%`` would match every dotted key and dump the
587 # whole table. The ``!= f"{key}."`` guard compares against the
588 # raw stored key, so it stays unescaped.
589 query = query.filter(
590 or_(
591 Setting.key == key,
592 and_(
593 Setting.key.like(
594 f"{escape_like(key)}.%", escape="\\"
595 ),
596 Setting.key != f"{key}.",
597 ),
598 )
599 )
600 return query.all()
602 logger.debug(
603 "DB is disabled, reading setting '{}' from defaults file.", key
604 )
606 settings = []
607 for candidate_key, setting in self.default_settings.items():
608 if key is None or (
609 candidate_key == key or candidate_key.startswith(f"{key}.")
610 ):
611 settings.append(
612 Setting(
613 key=candidate_key, # gitleaks:allow
614 **_filter_setting_columns(setting),
615 )
616 )
618 return settings
620 def get_setting(
621 self, key: str, default: Any = None, check_env: bool = True
622 ) -> Any:
623 """
624 Get a setting value
626 Args:
627 key: Setting key
628 default: Default value if setting is not found
629 check_env: If true, it will check the environment variable for
630 this setting before reading from the DB.
632 Returns:
633 Setting value or default if not found
634 """
635 if self._closed: 635 ↛ 636line 635 didn't jump to line 636 because the condition on line 635 was never true
636 logger.error(
637 "SettingsManager.get_setting('{}') called after close() — "
638 "this is a bug; the caller should not reuse a closed manager",
639 key,
640 )
641 raise RuntimeError(
642 "SettingsManager has been closed. "
643 "Create a new instance or call close() only at end of lifecycle."
644 )
646 # First check if this is an env-only setting
647 if env_registry.is_env_only(key):
648 return env_registry.get(key, default)
650 # If using database first approach and session available, check database
651 try:
652 settings = self.__query_settings(key)
653 # Drop malformed subkey rows (e.g. legacy "foo." / "foo.." keys)
654 # that slip past the SQL prefix filter. #4852 excludes only the
655 # exact single-trailing-dot row; a "foo.." / "foo. " row would
656 # otherwise turn a leaf read into a `{".": v}`-style wrapper dict
657 # rendered as `[object Object]` (#4840). The exact-match row is
658 # always kept so a direct read of a malformed key still works.
659 settings = [
660 s
661 for s in settings
662 if str(s.key) == key or is_valid_setting_key(str(s.key))
663 ]
664 if len(settings) == 1:
665 # This is a bottom-level key.
666 return self.__get_typed_setting_value(
667 settings[0], default, check_env
668 )
669 # Cache the result
670 if len(settings) > 1:
671 # This is a higher-level key.
672 settings_map = {}
673 for setting in settings:
674 output_key = str(setting.key).removeprefix(f"{key}.")
675 settings_map[output_key] = self.__get_typed_setting_value(
676 setting, default, check_env
677 )
678 return settings_map
679 except SQLAlchemyError:
680 logger.exception(f"Error retrieving setting {key} from database")
682 # Check env var before returning default (setting not in DB)
683 if check_env:
684 env_value = check_env_setting(key)
685 if env_value is not None:
686 default_meta = self.default_settings.get(key)
687 if default_meta and isinstance(default_meta, dict):
688 ui_element = default_meta.get("ui_element", "text")
689 return get_typed_setting_value(
690 key,
691 None,
692 ui_element,
693 default=default,
694 check_env=True,
695 )
696 logger.warning(
697 "Setting '{}' has env var override but is not in "
698 "defaults — returning raw string without type "
699 "conversion. Add this setting to a defaults JSON "
700 "file with a ui_element type to enable proper "
701 "type conversion.",
702 key,
703 )
704 return env_value
706 # Return default if not found
707 return default
709 def get_bool_setting(
710 self, key: str, default: bool = False, check_env: bool = True
711 ) -> bool:
712 """
713 Get a setting value as a boolean, handling string conversion.
715 Args:
716 key: Setting key
717 default: Default boolean value if setting is not found
718 check_env: If true, it will check the environment variable for
719 this setting before reading from the DB.
721 Returns:
722 Boolean value of the setting
723 """
724 value = self.get_setting(key, default, check_env)
725 return to_bool(value, default)
727 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
728 """
729 Set a setting value
731 Args:
732 key: Setting key
733 value: Setting value
734 commit: Whether to commit the change
736 Returns:
737 True if successful, False otherwise
738 """
739 if self._closed: 739 ↛ 740line 739 didn't jump to line 740 because the condition on line 739 was never true
740 logger.error(
741 "SettingsManager.set_setting('{}') called after close() — "
742 "this is a bug; the caller should not reuse a closed manager",
743 key,
744 )
745 raise RuntimeError(
746 "SettingsManager has been closed. "
747 "Create a new instance or call close() only at end of lifecycle."
748 )
749 if not self.db_session:
750 logger.error(
751 "Cannot edit setting {} because no DB was provided.", key
752 )
753 return False
754 if self.settings_locked:
755 logger.error("Cannot edit setting {} because they are locked.", key)
756 return False
758 # Always update database if available
759 try:
760 self._check_thread_safety()
761 setting = (
762 self.db_session.query(Setting)
763 .filter(Setting.key == key)
764 .first()
765 )
766 # Capture old value for the policy-change audit log below.
767 old_value = setting.value if setting is not None else None
768 if setting:
769 if not setting.editable:
770 logger.error(
771 "Cannot change setting '{}' because it "
772 "is marked as non-editable.",
773 key,
774 )
775 return False
777 setting.value = value # type: ignore[assignment]
778 setting.updated_at = ( # type: ignore[assignment]
779 func.now()
780 ) # Explicitly set the current timestamp
782 # Self-heal stale ui_element from before inference was added
783 setting.ui_element = _infer_ui_element(
784 value, setting.ui_element
785 )
787 # Self-heal stale type from before the prefix dispatch was
788 # added (e.g. legacy chat.* rows created with type=APP).
789 # Also re-points category to the canonical per-prefix
790 # value, since a row with the wrong type column was
791 # almost certainly created before category dispatch was
792 # in place either.
793 inferred_type: Optional[SettingType] = None
794 inferred_category: Optional[str] = None
795 for prefix, category in _INFERRED_CATEGORY.items():
796 if key.startswith(prefix):
797 if prefix == "llm.":
798 inferred_type = SettingType.LLM
799 elif prefix == "search.": 799 ↛ 800line 799 didn't jump to line 800 because the condition on line 799 was never true
800 inferred_type = SettingType.SEARCH
801 elif prefix == "report.": 801 ↛ 802line 801 didn't jump to line 802 because the condition on line 801 was never true
802 inferred_type = SettingType.REPORT
803 elif prefix == "database.": 803 ↛ 804line 803 didn't jump to line 804 because the condition on line 803 was never true
804 inferred_type = SettingType.DATABASE
805 elif prefix == "chat.": 805 ↛ 807line 805 didn't jump to line 807 because the condition on line 805 was always true
806 inferred_type = SettingType.CHAT
807 inferred_category = category
808 break
809 # Only self-heal when the key matches a known prefix. Keys
810 # outside the dispatch map (e.g. focused_iteration.*,
811 # langgraph_agent.* which ship as type=SEARCH) must keep their
812 # shipped type — defaulting to APP here would wrongly demote
813 # them on every edit.
814 if inferred_type is not None and setting.type != inferred_type:
815 setting.type = inferred_type # type: ignore[assignment]
816 if inferred_category is not None: 816 ↛ 856line 816 didn't jump to line 856 because the condition on line 816 was always true
817 setting.category = inferred_category # type: ignore[assignment]
818 else:
819 # Refuse to CREATE a new row for a malformed key (e.g. a
820 # trailing-dot key like "foo."). Existing rows are still
821 # updatable above; this only blocks minting new corruption.
822 if not is_valid_setting_key(key):
823 logger.error(
824 "Refusing to create setting with malformed key {!r}",
825 key,
826 )
827 return False
829 # Determine setting type from key
830 setting_type = SettingType.APP
831 if key.startswith("llm."):
832 setting_type = SettingType.LLM
833 elif key.startswith("search."):
834 setting_type = SettingType.SEARCH
835 elif key.startswith("report."):
836 setting_type = SettingType.REPORT
837 elif key.startswith("database."): 837 ↛ 838line 837 didn't jump to line 838 because the condition on line 837 was never true
838 setting_type = SettingType.DATABASE
839 elif key.startswith("chat."): 839 ↛ 840line 839 didn't jump to line 840 because the condition on line 839 was never true
840 setting_type = SettingType.CHAT
842 # Infer ui_element from the value type
843 ui_element = _infer_ui_element(value)
845 # Create a new setting
846 new_setting = Setting(
847 key=key,
848 value=value,
849 type=setting_type,
850 name=key.split(".")[-1].replace("_", " ").title(),
851 ui_element=ui_element,
852 description=f"Setting for {key}",
853 )
854 self.db_session.add(new_setting)
856 if commit:
857 self.db_session.commit()
858 # Emit WebSocket event for settings change
859 self._emit_settings_changed([key])
861 # N16: audit log on policy.* / llm.require_local_endpoint /
862 # embeddings.require_local changes. Targeted scope — only
863 # security-relevant settings are logged, to avoid widening
864 # this PR into a general audit-log refactor.
865 if _is_policy_setting(key): 865 ↛ 866line 865 didn't jump to line 866 because the condition on line 865 was never true
866 logger.bind(policy_audit=True).warning(
867 "policy setting changed | key={} old={} new={}",
868 key,
869 old_value,
870 value,
871 )
873 return True
874 except SQLAlchemyError:
875 logger.exception(f"Error setting value for key: {key}")
876 self.db_session.rollback()
877 return False
879 def clear_cache(self):
880 """Clear the settings cache."""
881 self.__dict__.pop("default_settings", None)
882 logger.debug("Settings cache cleared")
884 def get_all_settings(self, bypass_cache: bool = False) -> Dict[str, Any]:
885 """
886 Get all settings, merging defaults with database values.
888 This ensures that new settings added to defaults.json automatically
889 appear in the UI without requiring a database reset.
891 Args:
892 bypass_cache: If True, bypass the cache and read directly from database
894 Returns:
895 Dictionary of all settings
896 """
897 if self._closed: 897 ↛ 898line 897 didn't jump to line 898 because the condition on line 897 was never true
898 logger.error(
899 "SettingsManager.get_all_settings() called after close() — "
900 "this is a bug; the caller should not reuse a closed manager",
901 )
902 raise RuntimeError(
903 "SettingsManager has been closed. "
904 "Create a new instance or call close() only at end of lifecycle."
905 )
907 result = {}
909 # Start with defaults so new settings are always included
910 for key, default_setting in self.default_settings.items():
911 result[key] = dict(default_setting)
913 # Check env var override for defaults not yet in DB
914 env_value = check_env_setting(key)
915 if env_value is not None: 915 ↛ 916line 915 didn't jump to line 916 because the condition on line 915 was never true
916 ui_element = default_setting.get("ui_element", "text")
917 typed_value = get_typed_setting_value(
918 key,
919 None,
920 ui_element,
921 default=env_value,
922 check_env=True,
923 )
924 result[key]["value"] = typed_value
925 result[key]["editable"] = False
927 # Override with database settings
928 try:
929 db_settings = self.__query_settings()
930 except (SQLAlchemyError, LookupError):
931 # LookupError fires when a row's `type` column holds an enum
932 # value that's no longer in `SettingType` (e.g. legacy 'CHAT'-
933 # typed rows from removed features). The previous handler only
934 # caught SQLAlchemyError, so a single stale row would crash the
935 # whole snapshot — and every caller of get_all_settings (the
936 # /settings/api endpoint, benchmark start, research start, MCP
937 # entry points, …) downstream of it. Falling back to defaults-
938 # only is strictly safer than crashing.
939 logger.exception(
940 "Error querying settings from database in get_all_settings"
941 )
942 db_settings = []
944 for setting in db_settings:
945 # Handle type field - it might be a string or an enum
946 setting_type = setting.type
947 if hasattr(setting_type, "name"):
948 setting_type = setting_type.name
950 # Log if this is a custom setting not in defaults
951 if str(setting.key) not in result:
952 logger.debug(
953 f"Database contains custom setting not in "
954 f"defaults: {setting.key} (type={setting_type}, "
955 f"category={setting.category})"
956 )
958 # Override the default with the full database row — value AND
959 # schema (options, description, name, constraints).
960 #
961 # This is deliberate, not a bug: the DB row is a self-contained
962 # snapshot. Schema is NOT read live from JSON on every call; it is
963 # reconciled from the JSON defaults only at version bump, via
964 # `import_settings(overwrite=False)` (see `load_from_defaults_file`
965 # and its callers in `database/initialize.py` and post-login in
966 # `web/auth/routes.py`). Every release bumps the package version, so
967 # JSON metadata changes reach a user on their next login after
968 # upgrade. Do NOT change this to overlay defaults schema on every
969 # read — it bypasses that version gate and breaks the snapshot
970 # invariant (clean export/import round-trip, no mid-session drift).
971 # See closed PR #2474 for the rejected schema/value-separation
972 # approach and the reasoning.
973 result[str(setting.key)] = {
974 "value": setting.value,
975 "type": setting_type,
976 "name": setting.name,
977 "description": setting.description,
978 "category": setting.category,
979 "ui_element": setting.ui_element,
980 "options": setting.options,
981 "min_value": setting.min_value,
982 "max_value": setting.max_value,
983 "step": setting.step,
984 "visible": setting.visible,
985 "editable": False if self.settings_locked else setting.editable,
986 }
988 # Override from the environment variables if needed.
989 env_value = check_env_setting(str(setting.key))
990 if env_value is not None:
991 ui_element = result[str(setting.key)].get(
992 "ui_element", setting.ui_element
993 )
994 typed_value = get_typed_setting_value(
995 str(setting.key),
996 None,
997 ui_element,
998 default=env_value,
999 check_env=True,
1000 )
1001 result[str(setting.key)]["value"] = typed_value
1002 # Mark it as non-editable, because changes to the DB
1003 # value have no effect as long as the environment
1004 # variable is set.
1005 result[str(setting.key)]["editable"] = False
1007 # Re-inject search strategy options from code after DB merge,
1008 # since the DB stores options=null for this setting.
1009 if "search.search_strategy" in result:
1010 from local_deep_research.constants import get_available_strategies
1012 strategies = get_available_strategies()
1013 result["search.search_strategy"]["options"] = [
1014 {"label": s["label"], "value": s["name"]} for s in strategies
1015 ]
1017 return result
1019 def get_settings_snapshot(self) -> Dict[str, Any]:
1020 """
1021 Get a simplified settings snapshot with just key-value pairs.
1022 This is useful for passing settings to background threads or storing in metadata.
1024 Returns:
1025 Dictionary with setting keys mapped to their values
1026 """
1027 if self._closed: 1027 ↛ 1028line 1027 didn't jump to line 1028 because the condition on line 1027 was never true
1028 logger.error(
1029 "SettingsManager.get_settings_snapshot() called after close() — "
1030 "this is a bug; the caller should not reuse a closed manager",
1031 )
1032 raise RuntimeError(
1033 "SettingsManager has been closed. "
1034 "Create a new instance or call close() only at end of lifecycle."
1035 )
1037 all_settings = self.get_all_settings()
1038 settings_snapshot = {}
1040 for key, setting in all_settings.items():
1041 if isinstance(setting, dict) and "value" in setting: 1041 ↛ 1044line 1041 didn't jump to line 1044 because the condition on line 1041 was always true
1042 settings_snapshot[key] = setting["value"]
1043 else:
1044 settings_snapshot[key] = setting
1046 return settings_snapshot
1048 def create_or_update_setting(
1049 self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True
1050 ) -> Optional[Setting]:
1051 """
1052 Create or update a setting
1054 Args:
1055 setting: Setting object or dictionary
1056 commit: Whether to commit the change
1058 Returns:
1059 The created or updated Setting model, or None if failed
1060 """
1061 if not self.db_session:
1062 logger.warning(
1063 "No database session available, cannot create/update setting"
1064 )
1065 return None
1066 if self.settings_locked:
1067 logger.error("Cannot edit settings because they are locked.")
1068 return None
1070 # Convert dict to BaseSetting if needed
1071 if isinstance(setting, dict): 1071 ↛ 1098line 1071 didn't jump to line 1098 because the condition on line 1071 was always true
1072 # Determine type from key if not specified
1073 if "type" not in setting and "key" in setting:
1074 setting_obj: BaseSetting
1075 key = setting["key"]
1076 if key.startswith("llm."):
1077 setting_obj = LLMSetting(**setting)
1078 elif key.startswith("search."):
1079 setting_obj = SearchSetting(**setting)
1080 elif key.startswith("report."):
1081 setting_obj = ReportSetting(**setting)
1082 elif key.startswith("chat."): 1082 ↛ 1083line 1082 didn't jump to line 1083 because the condition on line 1082 was never true
1083 setting_obj = ChatSetting(**setting)
1084 elif key.startswith("app."):
1085 setting_obj = AppSetting(**setting)
1086 else:
1087 # Keys outside the four buckets (e.g. local_search_*,
1088 # embeddings.*, rag.*) live in their own namespaces.
1089 # Use BaseSetting so the key is written verbatim —
1090 # AppSetting's validator would otherwise prepend
1091 # `app.` and silently relocate the row away from
1092 # where every reader looks it up. See #4208.
1093 setting_obj = BaseSetting(type=SettingType.APP, **setting)
1094 else:
1095 # Use generic BaseSetting
1096 setting_obj = BaseSetting(**setting)
1097 else:
1098 setting_obj = setting
1100 try:
1101 # Check if setting exists
1102 db_setting = (
1103 self.db_session.query(Setting)
1104 .filter(Setting.key == setting_obj.key)
1105 .first()
1106 )
1108 if db_setting:
1109 # Update existing setting
1110 if not db_setting.editable:
1111 logger.error(
1112 "Cannot change setting '{}' because it "
1113 "is marked as non-editable.",
1114 setting_obj.key,
1115 )
1116 return None
1118 db_setting.value = setting_obj.value # type: ignore[assignment]
1119 db_setting.name = setting_obj.name # type: ignore[assignment]
1120 db_setting.description = setting_obj.description # type: ignore[assignment]
1121 db_setting.category = setting_obj.category # type: ignore[assignment]
1122 db_setting.type = setting_obj.type # type: ignore[assignment]
1123 db_setting.ui_element = setting_obj.ui_element # type: ignore[assignment]
1124 db_setting.options = setting_obj.options # type: ignore[assignment]
1125 db_setting.min_value = setting_obj.min_value # type: ignore[assignment]
1126 db_setting.max_value = setting_obj.max_value # type: ignore[assignment]
1127 db_setting.step = setting_obj.step # type: ignore[assignment]
1128 db_setting.visible = setting_obj.visible # type: ignore[assignment]
1129 db_setting.editable = setting_obj.editable # type: ignore[assignment]
1130 db_setting.updated_at = ( # type: ignore[assignment]
1131 func.now()
1132 ) # Explicitly set the current timestamp
1133 else:
1134 # Refuse to CREATE a new row for a malformed key (see
1135 # is_valid_setting_key / #4840). Updates to existing rows
1136 # above are unaffected.
1137 if not is_valid_setting_key(setting_obj.key):
1138 logger.error(
1139 "Refusing to create setting with malformed key {!r}",
1140 setting_obj.key,
1141 )
1142 return None
1144 # Create new setting
1145 db_setting = Setting(
1146 key=setting_obj.key,
1147 value=setting_obj.value,
1148 type=setting_obj.type,
1149 name=setting_obj.name,
1150 description=setting_obj.description,
1151 category=setting_obj.category,
1152 ui_element=setting_obj.ui_element,
1153 options=setting_obj.options,
1154 min_value=setting_obj.min_value,
1155 max_value=setting_obj.max_value,
1156 step=setting_obj.step,
1157 visible=setting_obj.visible,
1158 editable=setting_obj.editable,
1159 )
1160 self.db_session.add(db_setting)
1162 if commit:
1163 self.db_session.commit()
1164 # Emit WebSocket event for settings change
1165 self._emit_settings_changed([setting_obj.key])
1167 return db_setting
1169 except SQLAlchemyError:
1170 logger.exception(
1171 f"Error creating/updating setting {setting_obj.key}"
1172 )
1173 self.db_session.rollback()
1174 return None
1176 def delete_setting(self, key: str, commit: bool = True) -> bool:
1177 """
1178 Delete a setting
1180 Args:
1181 key: Setting key
1182 commit: Whether to commit the change
1184 Returns:
1185 True if successful, False otherwise
1186 """
1187 if not self.db_session:
1188 logger.warning(
1189 "No database session available, cannot delete setting"
1190 )
1191 return False
1193 try:
1194 # Remove from database
1195 result = (
1196 self.db_session.query(Setting)
1197 .filter(Setting.key == key)
1198 .delete()
1199 )
1201 if commit:
1202 self.db_session.commit()
1204 return result > 0
1205 except SQLAlchemyError:
1206 logger.exception("Error deleting setting")
1207 self.db_session.rollback()
1208 return False
1210 def load_from_defaults_file(
1211 self, commit: bool = True, **kwargs: Any
1212 ) -> None:
1213 """
1214 Import settings from the defaults settings file.
1216 Args:
1217 commit: Whether to commit changes to database. The post-login
1218 atomic block in `web/auth/routes.py` passes ``commit=False``
1219 and combines this call with ``update_db_version(commit=False)``
1220 under a single terminal ``db_session.commit()`` — preserving
1221 the all-or-nothing invariant is what prevents the sticky-loop
1222 bug where `app.version` is missing after a partial write.
1223 **kwargs: Will be passed to `import_settings`.
1225 """
1226 start = time.perf_counter()
1227 row_count = len(self.default_settings)
1228 self.import_settings(self.default_settings, commit=commit, **kwargs)
1229 elapsed_ms = (time.perf_counter() - start) * 1000
1230 if elapsed_ms > 100:
1231 logger.info(
1232 f"load_from_defaults_file imported {row_count} settings "
1233 f"in {elapsed_ms:.0f}ms (commit={commit})"
1234 )
1235 else:
1236 logger.debug(
1237 f"load_from_defaults_file imported {row_count} settings "
1238 f"in {elapsed_ms:.0f}ms (commit={commit})"
1239 )
1241 def db_version_matches_package(self) -> bool:
1242 """
1243 Returns:
1244 True if the version saved in the DB matches the package version.
1246 """
1247 db_version = self.get_setting("app.version")
1248 logger.debug(
1249 f"App version saved in DB is {db_version}, have package "
1250 f"settings from version {package_version}."
1251 )
1253 return bool(db_version == package_version)
1255 def update_db_version(self, commit: bool = True) -> None:
1256 """
1257 Updates the version saved in the DB based on the package version.
1259 Args:
1260 commit: Whether to commit the version write to the database.
1261 Callers that want to combine this with other writes into
1262 a single atomic transaction should pass commit=False and
1263 commit the session themselves. The post-login block in
1264 `web/auth/routes.py` relies on this to bundle the defaults
1265 import and the `app.version` write into one SQLite
1266 transaction; splitting them risks the sticky-loop state
1267 where `app.version` never gets written.
1268 """
1269 logger.debug(f"Updating saved DB version to {package_version}.")
1271 self.delete_setting("app.version", commit=False)
1272 version = Setting(
1273 key="app.version",
1274 value=package_version,
1275 description="Version of the app this database is associated with.",
1276 editable=False,
1277 name="App Version",
1278 type=SettingType.APP,
1279 ui_element="text",
1280 visible=False,
1281 )
1283 if self.db_session is None: 1283 ↛ 1284line 1283 didn't jump to line 1284 because the condition on line 1283 was never true
1284 raise RuntimeError("Database session is not initialized")
1285 self.db_session.add(version)
1286 if commit:
1287 self.db_session.commit()
1289 def import_settings(
1290 self,
1291 settings_data: Dict[str, Any],
1292 commit: bool = True,
1293 overwrite: bool = True,
1294 delete_extra: bool = False,
1295 ) -> None:
1296 """
1297 Import settings directly from the export format. This can be used to
1298 re-import settings that have been exported with `get_all_settings()`.
1300 Args:
1301 settings_data: The raw settings data to import.
1302 commit: Whether to commit the DB after loading the settings.
1303 overwrite: If true, it will overwrite the value of settings that
1304 are already in the database.
1305 delete_extra: If true, it will delete any settings that are in
1306 the database but don't have a corresponding entry in
1307 `settings_data`.
1309 """
1310 if self.db_session is None: 1310 ↛ 1311line 1310 didn't jump to line 1311 because the condition on line 1310 was never true
1311 raise RuntimeError("Database session is not initialized")
1312 logger.debug(f"Importing {len(settings_data)} settings")
1314 # `overwrite=False` is the version-bump reconciliation point: this is
1315 # the ONE place where JSON-defined schema (options, description,
1316 # constraints) refreshes into the DB rows while the user's chosen value
1317 # is preserved. `load_from_defaults_file(overwrite=False)` runs here on
1318 # version mismatch. Because each row is fully recreated from
1319 # `settings_data` below, all schema fields come from the JSON defaults;
1320 # only the existing value is carried over. `get_all_settings()` then
1321 # serves that snapshot verbatim — it does not re-read JSON per call.
1322 # This is why metadata edits surface at version bump, by design; see
1323 # the note at `get_all_settings()` and closed PR #2474.
1324 for key, setting_values in settings_data.items():
1325 # Don't let a corrupted export reintroduce malformed keys
1326 # (e.g. trailing-dot keys); skip them instead of writing verbatim.
1327 if not is_valid_setting_key(key):
1328 logger.warning(
1329 "Skipping import of setting with malformed key {!r}", key
1330 )
1331 continue
1332 setting_values = dict(setting_values)
1333 if not overwrite:
1334 existing_value = self.get_setting(key)
1335 if existing_value is not None:
1336 # Preserve the user's value; everything else (schema) is
1337 # taken fresh from `setting_values` (the JSON defaults).
1338 setting_values["value"] = existing_value
1340 # Delete any existing setting so we can completely overwrite it.
1341 self.delete_setting(key, commit=False)
1343 # Convert type string to SettingType enum if needed
1344 if "type" in setting_values and isinstance( 1344 ↛ 1349line 1344 didn't jump to line 1349 because the condition on line 1344 was always true
1345 setting_values["type"], str
1346 ):
1347 setting_values["type"] = SettingType[setting_values["type"]]
1349 setting = Setting(
1350 key=key, **_filter_setting_columns(setting_values)
1351 )
1352 self.db_session.add(setting)
1354 if commit or delete_extra:
1355 self.db_session.commit()
1356 logger.info(f"Successfully imported {len(settings_data)} settings")
1357 # Emit WebSocket event for all imported settings
1358 self._emit_settings_changed(list(settings_data.keys()))
1360 if delete_extra:
1361 all_settings = self.get_all_settings()
1362 for key in all_settings:
1363 if key not in settings_data:
1364 logger.debug(f"Deleting extraneous setting: {key}")
1365 self.delete_setting(key, commit=False)
1367 def _emit_settings_changed(self, changed_keys: Optional[List[Any]] = None):
1368 """
1369 Emit WebSocket event when settings change
1371 Args:
1372 changed_keys: List of setting keys that changed
1373 """
1374 try:
1375 # Import here to avoid circular imports
1376 from ..web.services.socket_service import SocketIOService
1378 try:
1379 socket_service = SocketIOService()
1380 except ValueError:
1381 logger.debug(
1382 "Not emitting socket event because server is not initialized."
1383 )
1384 return
1386 # settings_changed carries raw setting values (including plaintext
1387 # API keys), so it must reach only the owning user's own browser
1388 # tabs — never every connected client on this shared, single-user-
1389 # agnostic Socket.IO server. Resolve the user from the request that
1390 # triggered the change and scope the emit to that user's room.
1391 # A change made outside a request context (app start-up defaults,
1392 # background workers, migrations) has no user tab to notify, so we
1393 # skip the emit entirely rather than fall back to a broadcast.
1394 from flask import has_request_context, session
1396 if not has_request_context():
1397 return
1398 username = session.get("username")
1399 if not username:
1400 return
1402 # Get the changed settings
1403 settings_data = {}
1404 if changed_keys: 1404 ↛ 1411line 1404 didn't jump to line 1411 because the condition on line 1404 was always true
1405 for key in changed_keys:
1406 setting_value = self.get_setting(key)
1407 if setting_value is not None:
1408 settings_data[key] = {"value": setting_value}
1410 # Emit the settings change event
1411 from datetime import UTC, datetime
1413 socket_service.emit_socket_event(
1414 "settings_changed",
1415 {
1416 "changed_keys": changed_keys or [],
1417 "settings": settings_data,
1418 "timestamp": datetime.now(UTC).isoformat(),
1419 },
1420 room=socket_service.user_room(username),
1421 )
1423 logger.debug(
1424 f"Emitted settings_changed event for keys: {changed_keys}"
1425 )
1427 except Exception:
1428 logger.exception("Failed to emit settings change event")
1429 # Don't let WebSocket emission failures break settings saving
1431 @staticmethod
1432 def get_bootstrap_env_vars() -> Dict[str, str]:
1433 """
1434 Get environment variables that must be available before database access.
1435 These are critical for system initialization.
1437 Returns:
1438 Dict mapping env var names to their descriptions
1439 """
1440 # Get bootstrap vars from env registry
1441 return env_registry.get_bootstrap_vars()
1443 @staticmethod
1444 def is_bootstrap_env_var(env_var: str) -> bool:
1445 """
1446 Check if an environment variable is a bootstrap variable (needed before DB access).
1448 Args:
1449 env_var: Environment variable name
1451 Returns:
1452 True if this is a bootstrap variable
1453 """
1454 bootstrap_vars = SettingsManager.get_bootstrap_env_vars()
1455 return env_var in bootstrap_vars
1457 @staticmethod
1458 def is_env_only_setting(key: str) -> bool:
1459 """
1460 Check if a setting key is environment-only.
1462 Args:
1463 key: Setting key to check
1465 Returns:
1466 True if it's an env-only setting, False otherwise
1467 """
1468 return env_registry.is_env_only(key)
1470 @staticmethod
1471 def get_env_var_for_setting(setting_key: str) -> str:
1472 """
1473 Get the environment variable name for a given setting key.
1475 Args:
1476 setting_key: Setting key (e.g., "app.host")
1478 Returns:
1479 Environment variable name (e.g., "LDR_APP_HOST")
1480 """
1481 # Use the same logic as check_env_setting for consistency
1482 return f"LDR_{'_'.join(setting_key.split('.')).upper()}"
1484 @staticmethod
1485 def get_setting_key_for_env_var(env_var: str) -> Optional[str]:
1486 """
1487 Get the setting key for a given environment variable.
1489 Args:
1490 env_var: Environment variable name (e.g., "LDR_APP_HOST")
1492 Returns:
1493 Setting key (e.g., "app.host") or None if not a valid LDR env var
1494 """
1495 if not env_var.startswith("LDR_"):
1496 return None
1498 # Remove LDR_ prefix and convert to lowercase
1499 without_prefix = env_var[4:]
1500 parts = without_prefix.split("_")
1502 return ".".join(part.lower() for part in parts)
1505class SnapshotSettingsContext:
1506 """Read-only settings context backed by a snapshot dict.
1508 Unwraps {"value": x} setting objects into plain values and provides
1509 get_setting(key, default) for thread-safe snapshot access.
1510 """
1512 def __init__(
1513 self, snapshot=None, username=None, missing_key_log_level="DEBUG"
1514 ):
1515 self.snapshot = snapshot or {}
1516 self.username = username
1517 self._missing_key_log_level = missing_key_log_level
1518 self.values = {}
1519 for key, setting in self.snapshot.items():
1520 if isinstance(setting, dict) and "value" in setting:
1521 self.values[key] = setting["value"]
1522 else:
1523 self.values[key] = setting
1525 def get_setting(self, key, default=None):
1526 """Return the setting value for *key*, or *default* if absent."""
1527 if key in self.values:
1528 return self.values[key]
1529 logger.log(
1530 self._missing_key_log_level,
1531 "Setting '{}' not found in snapshot, using default",
1532 key,
1533 )
1534 return default