Coverage for src/local_deep_research/settings/manager.py: 91%
527 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 23:15 +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 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 ..web.models.settings import (
18 AppSetting,
19 BaseSetting,
20 ChatSetting,
21 LLMSetting,
22 ReportSetting,
23 SearchSetting,
24)
25from ..utilities.type_utils import to_bool
26from .base import ISettingsManager
27from .env_registry import registry as env_registry
30def parse_boolean(value: Any) -> bool:
31 """
32 Convert various representations to boolean using HTML checkbox semantics.
34 This function handles form values, JSON booleans, and environment variables,
35 ensuring consistent behavior across client and server.
37 **HTML Checkbox Semantics** (INTENTIONAL DESIGN):
38 - **Any value present (except explicit false) = checked = True**
39 - This matches standard HTML form behavior where checkbox presence indicates checked state
40 - In HTML forms, checkboxes send a value when checked, nothing when unchecked
42 **Examples**:
43 parse_boolean("on") # True - standard HTML checkbox value
44 parse_boolean("true") # True - explicit true
45 parse_boolean("1") # True - numeric true
46 parse_boolean("enabled") # True - any non-empty string
47 parse_boolean("disabled") # True - INTENTIONAL: any string = checkbox was checked!
48 parse_boolean("custom") # True - custom checkbox value
50 parse_boolean("false") # False - explicit false
51 parse_boolean("off") # False - explicit false
52 parse_boolean("0") # False - explicit false
53 parse_boolean("") # False - empty string = unchecked
54 parse_boolean(None) # False - missing = unchecked
56 **Why "disabled" returns True**:
57 This is NOT a bug! If a checkbox sends the value "disabled", it means the checkbox
58 was checked (present in form data). The actual string content doesn't matter for
59 HTML checkboxes - only presence vs absence matters.
61 Args:
62 value: Value to convert to boolean. Accepts strings, booleans, or None.
64 Returns:
65 bool: True for truthy values (any non-empty string except explicit false);
66 False for falsy values ('off', 'false', '0', '', 'no', False, None)
68 Note:
69 This function implements HTML form semantics, NOT generic boolean parsing.
70 See tests/settings/test_boolean_parsing.py for comprehensive test coverage.
71 """
72 # Constants for boolean value parsing
73 FALSY_VALUES = ("off", "false", "0", "", "no")
75 # Handle already-boolean values
76 if isinstance(value, bool):
77 return value
79 # Handle None (missing values)
80 if value is None:
81 return False
83 # Handle string values
84 if isinstance(value, str):
85 value_lower = value.lower().strip()
86 # Explicitly falsy values (empty string, false-like values)
87 if value_lower in FALSY_VALUES:
88 return False
89 # Any other non-empty string = True (HTML checkbox semantics)
90 return True
92 # For other types (numbers, lists, etc.), use Python's bool conversion
93 return bool(value)
96def _parse_number(x):
97 """Parse number, returning int if it's a whole number, otherwise float."""
98 f = float(x)
99 if f.is_integer():
100 return int(f)
101 return f
104def _parse_json_value(x):
105 """Parse JSON ui_element values.
107 DB values (via SQLAlchemy JSON column) arrive as Python objects already.
108 Form POST and env var overrides arrive as raw strings and need parsing.
109 For example, a textarea containing ``["general"]`` arrives as the string
110 ``'[\\r\\n "general"\\r\\n]'`` which must be decoded into a list.
111 """
112 if isinstance(x, str):
113 stripped = x.strip()
114 if stripped:
115 try:
116 return json.loads(stripped)
117 except (json.JSONDecodeError, ValueError, RecursionError):
118 logger.warning("Failed to parse JSON value, returning raw")
119 return x
120 return x
123def _parse_multiselect(x):
124 """Parse multiselect value, handling both lists and strings.
126 DB values (via SQLAlchemy JSON column) arrive as Python lists already.
127 Env var overrides arrive as strings and need parsing — either as JSON
128 arrays (e.g. '["markdown","latex"]') or comma-separated values
129 (e.g. 'markdown,latex').
130 """
131 if isinstance(x, list):
132 return x
133 if isinstance(x, str):
134 stripped = x.strip()
135 if stripped.startswith("["):
136 try:
137 parsed = json.loads(stripped)
138 if isinstance(parsed, list): 138 ↛ 143line 138 didn't jump to line 143 because the condition on line 138 was always true
139 return parsed
140 except (json.JSONDecodeError, ValueError):
141 pass
142 # Comma-separated fallback
143 return [item.strip() for item in stripped.split(",") if item.strip()]
144 return x
147def _filter_setting_columns(data: dict) -> dict:
148 """Filter a dict to only keys that are valid Setting model columns.
150 Prevents crashes when default_settings.json contains keys not present
151 as columns on the Setting model (e.g. future flags).
152 """
153 valid_columns = {c.name for c in Setting.__table__.columns}
154 return {k: v for k, v in data.items() if k in valid_columns}
157def _infer_ui_element(value: Any, current: str = "text") -> str:
158 """Infer the appropriate ui_element string from a Python value's type.
160 Args:
161 value: The value to infer the ui_element from.
162 current: The existing ui_element. If it is already something more
163 specific than ``"text"``, it is kept as-is.
164 """
165 if current != "text":
166 return current
167 if isinstance(value, bool):
168 return "checkbox"
169 if isinstance(value, (int, float)):
170 return "number"
171 if isinstance(value, (list, dict)):
172 return "json"
173 return "text"
176# Default categories for each typed setting prefix, used by the self-heal
177# block in set_setting() when a row's type column doesn't match its key
178# prefix. The values mirror the canonical strings in
179# web/routes/settings_routes.py: a legacy chat.* row with type=APP and a
180# stale category gets repointed to type=CHAT + category="chat" on next
181# save. We use the most-general category per prefix here — sub-classifying
182# llm_general vs llm_parameters depends on the specific key, but the
183# self-heal only fires when the row was already mis-typed, so over-
184# generalizing the category is preferable to leaving it stale.
185_INFERRED_CATEGORY: Dict[str, str] = {
186 "llm.": "llm_general",
187 "search.": "search_general",
188 "report.": "report_parameters",
189 "database.": "database_parameters",
190 "chat.": "chat",
191}
194UI_ELEMENT_TO_SETTING_TYPE: Dict[str, Callable[..., Any]] = {
195 "text": str,
196 "json": _parse_json_value,
197 "password": str,
198 "select": str,
199 "number": _parse_number,
200 "range": _parse_number, # Same behavior as number for consistency
201 "checkbox": parse_boolean,
202 "textarea": str,
203 "multiselect": _parse_multiselect,
204}
207def get_typed_setting_value(
208 key: str,
209 value: Any,
210 ui_element: str,
211 default: Any = None,
212 check_env: bool = True,
213) -> Any:
214 """
215 Extracts the value for a particular setting, ensuring that it has the
216 correct type.
218 Args:
219 key: The setting key.
220 value: The setting value from the database.
221 ui_element: The setting UI element ID.
222 default: Default value to return if the value of the setting is
223 invalid.
224 check_env: If true, it will check the environment variable for
225 this setting before reading from the DB.
227 Returns:
228 The value of the setting.
230 """
231 setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None)
232 if setting_type is None:
233 logger.warning(
234 "Got unknown type {} for setting {}, returning default value.",
235 ui_element,
236 key,
237 )
238 return default
240 # Check environment variable first (highest priority).
241 if check_env:
242 env_value = check_env_setting(key)
243 if env_value is not None:
244 try:
245 return setting_type(env_value)
246 except ValueError:
247 logger.warning(
248 "Setting {} has invalid value {}. Falling back to DB.",
249 key,
250 env_value,
251 )
253 # If value is None (not in database), return default.
254 if value is None:
255 return default
257 # Read from the database.
258 try:
259 return setting_type(value)
260 except (ValueError, TypeError):
261 logger.warning(
262 "Setting {} has invalid value {}. Returning default.",
263 key,
264 value,
265 )
266 return default
269def check_env_setting(key: str) -> str | None:
270 """
271 Checks environment variables for a particular setting.
273 Args:
274 key: The database key for the setting.
276 Returns:
277 The setting from the environment variables, or None if the variable
278 is not set or is empty.
280 Note:
281 Empty environment variables ("") are treated as unset. This is standard
282 practice across the ecosystem — see CPython's official docs (PYTHON*
283 env vars require "a non-empty string"), botocore PR #1687, Pallets/Click
284 PR #2223, and Vercel Turborepo PR #6929. Orchestration tools like Unraid,
285 Terraform, and Kubernetes manifests often cannot conditionally omit env
286 var declarations, so they pass "" for unconfigured values. Treating ""
287 as unset prevents these empty strings from overriding database defaults.
288 See: https://github.com/LearningCircuit/local-deep-research/pull/3362
290 """
291 env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}"
292 env_value = os.getenv(env_variable_name)
293 # Treat empty string as unset — orchestration tools (Unraid, Terraform, K8s)
294 # often cannot omit env var declarations and pass "" for unconfigured values.
295 if env_value is not None and env_value != "":
296 logger.debug(f"Overriding {key} setting from environment variable.")
297 return env_value
298 if env_value == "":
299 logger.warning(
300 "Environment variable {} is set but empty — "
301 "ignoring it and falling back to DB/default for setting '{}'. "
302 "This is expected on Unraid or Docker templates that create "
303 "all variables even when left blank. To suppress this warning, "
304 "remove the variable from your environment or set a value.",
305 env_variable_name,
306 key,
307 )
308 return None
311class SettingsManager(ISettingsManager):
312 """
313 Manager for handling application settings with database storage and file fallback.
314 Provides methods to get and set settings, with the ability to override settings in memory.
315 """
317 def __init__(
318 self,
319 db_session: Optional[Session] = None,
320 owns_session: bool = False,
321 ):
322 """
323 Initialize the settings manager
325 Args:
326 db_session: SQLAlchemy session for database operations
327 owns_session: If True, close() will close the session.
328 Defaults to False (safe for borrowed sessions). Set to True
329 only when this manager created/owns the session — currently
330 only get_settings_manager() in db_utils.py does this.
331 """
332 self.db_session = db_session
333 self._owns_session = owns_session
334 self._closed = False
335 self.db_first = True # Always prioritize DB settings
337 # Store the thread ID this instance was created in
338 self._creation_thread_id = threading.get_ident()
340 # Initialize settings lock as None - will be checked lazily
341 self.__settings_locked: Optional[bool] = None
343 # Auto-initialize settings if database is empty
344 if self.db_session:
345 self._ensure_settings_initialized()
347 def close(self):
348 """Close the DB session if this manager owns it.
350 Borrowed sessions (owns_session=False) are left open for their
351 owner to close (e.g. Flask teardown closes g.db_session).
352 Safe to call multiple times — subsequent calls are no-ops.
353 """
354 if self._owns_session and self.db_session is not None:
355 try:
356 logger.debug("Closing owned DB session in SettingsManager")
357 self.db_session.close()
358 except Exception:
359 logger.warning(
360 "Failed to close SettingsManager DB session — "
361 "connection may leak",
362 )
363 self._closed = True
364 self.db_session = None
366 def _ensure_settings_initialized(self):
367 """Ensure settings are initialized in the database."""
368 # Check if we have any settings at all
369 from ..database.models import Setting
371 if self.db_session is None: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 raise RuntimeError("Database session is not initialized")
373 settings_count = self.db_session.query(Setting).count()
375 if settings_count == 0:
376 logger.info("No settings found in database, loading defaults")
377 self.load_from_defaults_file(commit=True)
378 logger.info("Default settings loaded successfully")
380 def _check_thread_safety(self):
381 """Check if this instance is being used in the same thread it was created in."""
382 current_thread_id = threading.get_ident()
383 if self.db_session and current_thread_id != self._creation_thread_id:
384 raise RuntimeError(
385 f"SettingsManager instance created in thread {self._creation_thread_id} "
386 f"is being used in thread {current_thread_id}. This is not thread-safe! "
387 f"Create a new SettingsManager instance within the current thread context."
388 )
390 @property
391 def settings_locked(self) -> bool:
392 """Check if settings are locked (lazy evaluation)."""
393 if self.__settings_locked is None:
394 try:
395 self.__settings_locked = self.get_setting(
396 "app.lock_settings", False
397 )
398 if self.settings_locked:
399 logger.info(
400 "Settings are locked. Disabling all settings changes."
401 )
402 except Exception:
403 logger.warning(
404 "Failed to check settings lock status, assuming not locked"
405 )
406 self.__settings_locked = False
407 return bool(self.__settings_locked)
409 @functools.cached_property
410 def default_settings(self) -> Dict[str, Any]:
411 """
412 Returns:
413 The default settings, loaded from JSON files and merged.
414 Automatically discovers and loads all .json files in the defaults
415 directory and its subdirectories.
416 Theme options are dynamically injected from the theme registry.
418 """
419 settings: Dict[str, Any] = {}
421 try:
422 # Get the defaults package path
423 defaults_path = Path(defaults.__file__).parent
425 # Find all JSON files recursively in the defaults directory
426 json_files = sorted(defaults_path.rglob("*.json"))
428 logger.debug(f"Found {len(json_files)} JSON settings files")
430 # Load and merge all JSON files
431 for json_file in json_files:
432 try:
433 with open(json_file, "r", encoding="utf-8-sig") as f:
434 file_settings = json.load(f)
436 # Get relative path for logging
437 relative_path = json_file.relative_to(defaults_path)
439 # Warn about key conflicts
440 conflicts = set(settings.keys()) & set(file_settings.keys())
441 if conflicts:
442 logger.warning(
443 f"Keys {conflicts} from {relative_path} "
444 f"override existing values"
445 )
447 settings.update(file_settings)
448 logger.debug(f"Loaded {relative_path}")
450 except json.JSONDecodeError:
451 logger.exception(f"Invalid JSON in {json_file}")
452 except Exception:
453 logger.warning(f"Could not load {json_file}")
455 except Exception:
456 logger.warning("Error loading settings files")
458 # Inject dynamic theme options from theme registry
459 if "app.theme" in settings:
460 try:
461 from local_deep_research.web.themes import theme_registry
463 settings["app.theme"]["options"] = (
464 theme_registry.get_settings_options()
465 )
466 except ImportError:
467 # Theme registry not available, use static options from JSON
468 pass
470 # Inject search strategy options from code (single source of truth)
471 if "search.search_strategy" in settings:
472 from local_deep_research.constants import get_available_strategies
474 # Check the show_all_strategies setting to decide which list
475 show_all = False
476 if "search.show_all_strategies" in settings: 476 ↛ 480line 476 didn't jump to line 480 because the condition on line 476 was always true
477 val = settings["search.show_all_strategies"].get("value", False)
478 show_all = val is True or val == "true"
480 strategies = get_available_strategies(show_all=show_all)
481 settings["search.search_strategy"]["options"] = [
482 {"label": s["label"], "value": s["name"]} for s in strategies
483 ]
485 logger.debug(f"Loaded {len(settings)} total settings")
486 return settings
488 def __get_typed_setting_value(
489 self,
490 setting: Setting,
491 default: Any = None,
492 check_env: bool = True,
493 ) -> Any:
494 """
495 Extracts the value for a particular setting, ensuring that it has the
496 correct type.
498 Args:
499 setting: The setting to get the value for.
500 default: Default value to return if the value of the setting is
501 invalid.
502 check_env: If true, it will check the environment variable for
503 this setting before reading from the DB.
505 Returns:
506 The value of the setting.
508 """
509 return get_typed_setting_value(
510 str(setting.key),
511 setting.value,
512 str(setting.ui_element),
513 default=default,
514 check_env=check_env,
515 )
517 def __query_settings(self, key: str | None = None) -> List[Setting]:
518 """
519 Abstraction for querying settings that also transparently handles
520 reading the default settings file if the DB is not enabled.
522 Args:
523 key: The key to read. If None, it will read everything.
525 Returns:
526 The settings it queried.
528 """
529 if self.db_session:
530 self._check_thread_safety()
531 query = self.db_session.query(Setting)
532 if key is not None:
533 # This will find exact matches and any subkeys.
534 query = query.filter(
535 or_(
536 Setting.key == key,
537 Setting.key.startswith(f"{key}."),
538 )
539 )
540 return query.all()
542 logger.debug(
543 "DB is disabled, reading setting '{}' from defaults file.", key
544 )
546 settings = []
547 for candidate_key, setting in self.default_settings.items():
548 if key is None or (
549 candidate_key == key or candidate_key.startswith(f"{key}.")
550 ):
551 settings.append(
552 Setting(
553 key=candidate_key, # gitleaks:allow
554 **_filter_setting_columns(setting),
555 )
556 )
558 return settings
560 def get_setting(
561 self, key: str, default: Any = None, check_env: bool = True
562 ) -> Any:
563 """
564 Get a setting value
566 Args:
567 key: Setting key
568 default: Default value if setting is not found
569 check_env: If true, it will check the environment variable for
570 this setting before reading from the DB.
572 Returns:
573 Setting value or default if not found
574 """
575 if self._closed: 575 ↛ 576line 575 didn't jump to line 576 because the condition on line 575 was never true
576 logger.error(
577 "SettingsManager.get_setting('{}') called after close() — "
578 "this is a bug; the caller should not reuse a closed manager",
579 key,
580 )
581 raise RuntimeError(
582 "SettingsManager has been closed. "
583 "Create a new instance or call close() only at end of lifecycle."
584 )
586 # First check if this is an env-only setting
587 if env_registry.is_env_only(key):
588 return env_registry.get(key, default)
590 # If using database first approach and session available, check database
591 try:
592 settings = self.__query_settings(key)
593 if len(settings) == 1:
594 # This is a bottom-level key.
595 return self.__get_typed_setting_value(
596 settings[0], default, check_env
597 )
598 # Cache the result
599 if len(settings) > 1:
600 # This is a higher-level key.
601 settings_map = {}
602 for setting in settings:
603 output_key = str(setting.key).removeprefix(f"{key}.")
604 settings_map[output_key] = self.__get_typed_setting_value(
605 setting, default, check_env
606 )
607 return settings_map
608 except SQLAlchemyError:
609 logger.exception(f"Error retrieving setting {key} from database")
611 # Check env var before returning default (setting not in DB)
612 if check_env:
613 env_value = check_env_setting(key)
614 if env_value is not None:
615 default_meta = self.default_settings.get(key)
616 if default_meta and isinstance(default_meta, dict):
617 ui_element = default_meta.get("ui_element", "text")
618 return get_typed_setting_value(
619 key,
620 None,
621 ui_element,
622 default=default,
623 check_env=True,
624 )
625 logger.warning(
626 "Setting '{}' has env var override but is not in "
627 "defaults — returning raw string without type "
628 "conversion. Add this setting to a defaults JSON "
629 "file with a ui_element type to enable proper "
630 "type conversion.",
631 key,
632 )
633 return env_value
635 # Return default if not found
636 return default
638 def get_bool_setting(
639 self, key: str, default: bool = False, check_env: bool = True
640 ) -> bool:
641 """
642 Get a setting value as a boolean, handling string conversion.
644 Args:
645 key: Setting key
646 default: Default boolean value if setting is not found
647 check_env: If true, it will check the environment variable for
648 this setting before reading from the DB.
650 Returns:
651 Boolean value of the setting
652 """
653 value = self.get_setting(key, default, check_env)
654 return to_bool(value, default)
656 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
657 """
658 Set a setting value
660 Args:
661 key: Setting key
662 value: Setting value
663 commit: Whether to commit the change
665 Returns:
666 True if successful, False otherwise
667 """
668 if self._closed: 668 ↛ 669line 668 didn't jump to line 669 because the condition on line 668 was never true
669 logger.error(
670 "SettingsManager.set_setting('{}') called after close() — "
671 "this is a bug; the caller should not reuse a closed manager",
672 key,
673 )
674 raise RuntimeError(
675 "SettingsManager has been closed. "
676 "Create a new instance or call close() only at end of lifecycle."
677 )
678 if not self.db_session:
679 logger.error(
680 "Cannot edit setting {} because no DB was provided.", key
681 )
682 return False
683 if self.settings_locked:
684 logger.error("Cannot edit setting {} because they are locked.", key)
685 return False
687 # Always update database if available
688 try:
689 self._check_thread_safety()
690 setting = (
691 self.db_session.query(Setting)
692 .filter(Setting.key == key)
693 .first()
694 )
695 if setting:
696 if not setting.editable:
697 logger.error(
698 "Cannot change setting '{}' because it "
699 "is marked as non-editable.",
700 key,
701 )
702 return False
704 setting.value = value # type: ignore[assignment]
705 setting.updated_at = ( # type: ignore[assignment]
706 func.now()
707 ) # Explicitly set the current timestamp
709 # Self-heal stale ui_element from before inference was added
710 setting.ui_element = _infer_ui_element(
711 value, setting.ui_element
712 )
714 # Self-heal stale type from before the prefix dispatch was
715 # added (e.g. legacy chat.* rows created with type=APP).
716 # Also re-points category to the canonical per-prefix
717 # value, since a row with the wrong type column was
718 # almost certainly created before category dispatch was
719 # in place either.
720 inferred_type: Optional[SettingType] = None
721 inferred_category: Optional[str] = None
722 for prefix, category in _INFERRED_CATEGORY.items():
723 if key.startswith(prefix):
724 if prefix == "llm.":
725 inferred_type = SettingType.LLM
726 elif prefix == "search.": 726 ↛ 727line 726 didn't jump to line 727 because the condition on line 726 was never true
727 inferred_type = SettingType.SEARCH
728 elif prefix == "report.": 728 ↛ 729line 728 didn't jump to line 729 because the condition on line 728 was never true
729 inferred_type = SettingType.REPORT
730 elif prefix == "database.": 730 ↛ 731line 730 didn't jump to line 731 because the condition on line 730 was never true
731 inferred_type = SettingType.DATABASE
732 elif prefix == "chat.": 732 ↛ 734line 732 didn't jump to line 734 because the condition on line 732 was always true
733 inferred_type = SettingType.CHAT
734 inferred_category = category
735 break
736 # Only self-heal when the key matches a known prefix. Keys
737 # outside the dispatch map (e.g. focused_iteration.*,
738 # langgraph_agent.* which ship as type=SEARCH) must keep their
739 # shipped type — defaulting to APP here would wrongly demote
740 # them on every edit.
741 if inferred_type is not None and setting.type != inferred_type:
742 setting.type = inferred_type # type: ignore[assignment]
743 if inferred_category is not None: 743 ↛ 773line 743 didn't jump to line 773 because the condition on line 743 was always true
744 setting.category = inferred_category # type: ignore[assignment]
745 else:
746 # Determine setting type from key
747 setting_type = SettingType.APP
748 if key.startswith("llm."):
749 setting_type = SettingType.LLM
750 elif key.startswith("search."):
751 setting_type = SettingType.SEARCH
752 elif key.startswith("report."):
753 setting_type = SettingType.REPORT
754 elif key.startswith("database."): 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 setting_type = SettingType.DATABASE
756 elif key.startswith("chat."): 756 ↛ 757line 756 didn't jump to line 757 because the condition on line 756 was never true
757 setting_type = SettingType.CHAT
759 # Infer ui_element from the value type
760 ui_element = _infer_ui_element(value)
762 # Create a new setting
763 new_setting = Setting(
764 key=key,
765 value=value,
766 type=setting_type,
767 name=key.split(".")[-1].replace("_", " ").title(),
768 ui_element=ui_element,
769 description=f"Setting for {key}",
770 )
771 self.db_session.add(new_setting)
773 if commit:
774 self.db_session.commit()
775 # Emit WebSocket event for settings change
776 self._emit_settings_changed([key])
778 return True
779 except SQLAlchemyError:
780 logger.exception(f"Error setting value for key: {key}")
781 self.db_session.rollback()
782 return False
784 def clear_cache(self):
785 """Clear the settings cache."""
786 self.__dict__.pop("default_settings", None)
787 logger.debug("Settings cache cleared")
789 def get_all_settings(self, bypass_cache: bool = False) -> Dict[str, Any]:
790 """
791 Get all settings, merging defaults with database values.
793 This ensures that new settings added to defaults.json automatically
794 appear in the UI without requiring a database reset.
796 Args:
797 bypass_cache: If True, bypass the cache and read directly from database
799 Returns:
800 Dictionary of all settings
801 """
802 if self._closed: 802 ↛ 803line 802 didn't jump to line 803 because the condition on line 802 was never true
803 logger.error(
804 "SettingsManager.get_all_settings() called after close() — "
805 "this is a bug; the caller should not reuse a closed manager",
806 )
807 raise RuntimeError(
808 "SettingsManager has been closed. "
809 "Create a new instance or call close() only at end of lifecycle."
810 )
812 result = {}
814 # Start with defaults so new settings are always included
815 for key, default_setting in self.default_settings.items():
816 result[key] = dict(default_setting)
818 # Check env var override for defaults not yet in DB
819 env_value = check_env_setting(key)
820 if env_value is not None: 820 ↛ 821line 820 didn't jump to line 821 because the condition on line 820 was never true
821 ui_element = default_setting.get("ui_element", "text")
822 typed_value = get_typed_setting_value(
823 key,
824 None,
825 ui_element,
826 default=env_value,
827 check_env=True,
828 )
829 result[key]["value"] = typed_value
830 result[key]["editable"] = False
832 # Override with database settings
833 try:
834 db_settings = self.__query_settings()
835 except SQLAlchemyError:
836 logger.exception(
837 "Error querying settings from database in get_all_settings"
838 )
839 db_settings = []
841 for setting in db_settings:
842 # Handle type field - it might be a string or an enum
843 setting_type = setting.type
844 if hasattr(setting_type, "name"):
845 setting_type = setting_type.name
847 # Log if this is a custom setting not in defaults
848 if str(setting.key) not in result:
849 logger.debug(
850 f"Database contains custom setting not in "
851 f"defaults: {setting.key} (type={setting_type}, "
852 f"category={setting.category})"
853 )
855 # Override default with database value
856 result[str(setting.key)] = {
857 "value": setting.value,
858 "type": setting_type,
859 "name": setting.name,
860 "description": setting.description,
861 "category": setting.category,
862 "ui_element": setting.ui_element,
863 "options": setting.options,
864 "min_value": setting.min_value,
865 "max_value": setting.max_value,
866 "step": setting.step,
867 "visible": setting.visible,
868 "editable": False if self.settings_locked else setting.editable,
869 }
871 # Override from the environment variables if needed.
872 env_value = check_env_setting(str(setting.key))
873 if env_value is not None:
874 ui_element = result[str(setting.key)].get(
875 "ui_element", setting.ui_element
876 )
877 typed_value = get_typed_setting_value(
878 str(setting.key),
879 None,
880 ui_element,
881 default=env_value,
882 check_env=True,
883 )
884 result[str(setting.key)]["value"] = typed_value
885 # Mark it as non-editable, because changes to the DB
886 # value have no effect as long as the environment
887 # variable is set.
888 result[str(setting.key)]["editable"] = False
890 # Re-inject search strategy options from code after DB merge,
891 # since the DB stores options=null for this setting.
892 if "search.search_strategy" in result:
893 from local_deep_research.constants import get_available_strategies
895 show_all_val = result.get("search.show_all_strategies", {}).get(
896 "value", False
897 )
898 show_all = show_all_val is True or show_all_val == "true"
899 strategies = get_available_strategies(show_all=show_all)
900 result["search.search_strategy"]["options"] = [
901 {"label": s["label"], "value": s["name"]} for s in strategies
902 ]
904 return result
906 def get_settings_snapshot(self) -> Dict[str, Any]:
907 """
908 Get a simplified settings snapshot with just key-value pairs.
909 This is useful for passing settings to background threads or storing in metadata.
911 Returns:
912 Dictionary with setting keys mapped to their values
913 """
914 if self._closed: 914 ↛ 915line 914 didn't jump to line 915 because the condition on line 914 was never true
915 logger.error(
916 "SettingsManager.get_settings_snapshot() called after close() — "
917 "this is a bug; the caller should not reuse a closed manager",
918 )
919 raise RuntimeError(
920 "SettingsManager has been closed. "
921 "Create a new instance or call close() only at end of lifecycle."
922 )
924 all_settings = self.get_all_settings()
925 settings_snapshot = {}
927 for key, setting in all_settings.items():
928 if isinstance(setting, dict) and "value" in setting: 928 ↛ 931line 928 didn't jump to line 931 because the condition on line 928 was always true
929 settings_snapshot[key] = setting["value"]
930 else:
931 settings_snapshot[key] = setting
933 return settings_snapshot
935 def create_or_update_setting(
936 self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True
937 ) -> Optional[Setting]:
938 """
939 Create or update a setting
941 Args:
942 setting: Setting object or dictionary
943 commit: Whether to commit the change
945 Returns:
946 The created or updated Setting model, or None if failed
947 """
948 if not self.db_session:
949 logger.warning(
950 "No database session available, cannot create/update setting"
951 )
952 return None
953 if self.settings_locked:
954 logger.error("Cannot edit settings because they are locked.")
955 return None
957 # Convert dict to BaseSetting if needed
958 if isinstance(setting, dict): 958 ↛ 985line 958 didn't jump to line 985 because the condition on line 958 was always true
959 # Determine type from key if not specified
960 if "type" not in setting and "key" in setting:
961 setting_obj: BaseSetting
962 key = setting["key"]
963 if key.startswith("llm."):
964 setting_obj = LLMSetting(**setting)
965 elif key.startswith("search."):
966 setting_obj = SearchSetting(**setting)
967 elif key.startswith("report."):
968 setting_obj = ReportSetting(**setting)
969 elif key.startswith("chat."): 969 ↛ 970line 969 didn't jump to line 970 because the condition on line 969 was never true
970 setting_obj = ChatSetting(**setting)
971 elif key.startswith("app."):
972 setting_obj = AppSetting(**setting)
973 else:
974 # Keys outside the four buckets (e.g. local_search_*,
975 # embeddings.*, rag.*) live in their own namespaces.
976 # Use BaseSetting so the key is written verbatim —
977 # AppSetting's validator would otherwise prepend
978 # `app.` and silently relocate the row away from
979 # where every reader looks it up. See #4208.
980 setting_obj = BaseSetting(type=SettingType.APP, **setting)
981 else:
982 # Use generic BaseSetting
983 setting_obj = BaseSetting(**setting)
984 else:
985 setting_obj = setting
987 try:
988 # Check if setting exists
989 db_setting = (
990 self.db_session.query(Setting)
991 .filter(Setting.key == setting_obj.key)
992 .first()
993 )
995 if db_setting:
996 # Update existing setting
997 if not db_setting.editable:
998 logger.error(
999 "Cannot change setting '{}' because it "
1000 "is marked as non-editable.",
1001 setting_obj.key,
1002 )
1003 return None
1005 db_setting.value = setting_obj.value # type: ignore[assignment]
1006 db_setting.name = setting_obj.name # type: ignore[assignment]
1007 db_setting.description = setting_obj.description # type: ignore[assignment]
1008 db_setting.category = setting_obj.category # type: ignore[assignment]
1009 db_setting.type = setting_obj.type # type: ignore[assignment]
1010 db_setting.ui_element = setting_obj.ui_element # type: ignore[assignment]
1011 db_setting.options = setting_obj.options # type: ignore[assignment]
1012 db_setting.min_value = setting_obj.min_value # type: ignore[assignment]
1013 db_setting.max_value = setting_obj.max_value # type: ignore[assignment]
1014 db_setting.step = setting_obj.step # type: ignore[assignment]
1015 db_setting.visible = setting_obj.visible # type: ignore[assignment]
1016 db_setting.editable = setting_obj.editable # type: ignore[assignment]
1017 db_setting.updated_at = ( # type: ignore[assignment]
1018 func.now()
1019 ) # Explicitly set the current timestamp
1020 else:
1021 # Create new setting
1022 db_setting = Setting(
1023 key=setting_obj.key,
1024 value=setting_obj.value,
1025 type=setting_obj.type,
1026 name=setting_obj.name,
1027 description=setting_obj.description,
1028 category=setting_obj.category,
1029 ui_element=setting_obj.ui_element,
1030 options=setting_obj.options,
1031 min_value=setting_obj.min_value,
1032 max_value=setting_obj.max_value,
1033 step=setting_obj.step,
1034 visible=setting_obj.visible,
1035 editable=setting_obj.editable,
1036 )
1037 self.db_session.add(db_setting)
1039 if commit:
1040 self.db_session.commit()
1041 # Emit WebSocket event for settings change
1042 self._emit_settings_changed([setting_obj.key])
1044 return db_setting
1046 except SQLAlchemyError:
1047 logger.exception(
1048 f"Error creating/updating setting {setting_obj.key}"
1049 )
1050 self.db_session.rollback()
1051 return None
1053 def delete_setting(self, key: str, commit: bool = True) -> bool:
1054 """
1055 Delete a setting
1057 Args:
1058 key: Setting key
1059 commit: Whether to commit the change
1061 Returns:
1062 True if successful, False otherwise
1063 """
1064 if not self.db_session:
1065 logger.warning(
1066 "No database session available, cannot delete setting"
1067 )
1068 return False
1070 try:
1071 # Remove from database
1072 result = (
1073 self.db_session.query(Setting)
1074 .filter(Setting.key == key)
1075 .delete()
1076 )
1078 if commit:
1079 self.db_session.commit()
1081 return result > 0
1082 except SQLAlchemyError:
1083 logger.exception("Error deleting setting")
1084 self.db_session.rollback()
1085 return False
1087 def load_from_defaults_file(
1088 self, commit: bool = True, **kwargs: Any
1089 ) -> None:
1090 """
1091 Import settings from the defaults settings file.
1093 Args:
1094 commit: Whether to commit changes to database. The post-login
1095 atomic block in `web/auth/routes.py` passes ``commit=False``
1096 and combines this call with ``update_db_version(commit=False)``
1097 under a single terminal ``db_session.commit()`` — preserving
1098 the all-or-nothing invariant is what prevents the sticky-loop
1099 bug where `app.version` is missing after a partial write.
1100 **kwargs: Will be passed to `import_settings`.
1102 """
1103 start = time.perf_counter()
1104 row_count = len(self.default_settings)
1105 self.import_settings(self.default_settings, commit=commit, **kwargs)
1106 elapsed_ms = (time.perf_counter() - start) * 1000
1107 if elapsed_ms > 100:
1108 logger.info(
1109 f"load_from_defaults_file imported {row_count} settings "
1110 f"in {elapsed_ms:.0f}ms (commit={commit})"
1111 )
1112 else:
1113 logger.debug(
1114 f"load_from_defaults_file imported {row_count} settings "
1115 f"in {elapsed_ms:.0f}ms (commit={commit})"
1116 )
1118 def db_version_matches_package(self) -> bool:
1119 """
1120 Returns:
1121 True if the version saved in the DB matches the package version.
1123 """
1124 db_version = self.get_setting("app.version")
1125 logger.debug(
1126 f"App version saved in DB is {db_version}, have package "
1127 f"settings from version {package_version}."
1128 )
1130 return bool(db_version == package_version)
1132 def update_db_version(self, commit: bool = True) -> None:
1133 """
1134 Updates the version saved in the DB based on the package version.
1136 Args:
1137 commit: Whether to commit the version write to the database.
1138 Callers that want to combine this with other writes into
1139 a single atomic transaction should pass commit=False and
1140 commit the session themselves. The post-login block in
1141 `web/auth/routes.py` relies on this to bundle the defaults
1142 import and the `app.version` write into one SQLite
1143 transaction; splitting them risks the sticky-loop state
1144 where `app.version` never gets written.
1145 """
1146 logger.debug(f"Updating saved DB version to {package_version}.")
1148 self.delete_setting("app.version", commit=False)
1149 version = Setting(
1150 key="app.version",
1151 value=package_version,
1152 description="Version of the app this database is associated with.",
1153 editable=False,
1154 name="App Version",
1155 type=SettingType.APP,
1156 ui_element="text",
1157 visible=False,
1158 )
1160 if self.db_session is None: 1160 ↛ 1161line 1160 didn't jump to line 1161 because the condition on line 1160 was never true
1161 raise RuntimeError("Database session is not initialized")
1162 self.db_session.add(version)
1163 if commit:
1164 self.db_session.commit()
1166 def import_settings(
1167 self,
1168 settings_data: Dict[str, Any],
1169 commit: bool = True,
1170 overwrite: bool = True,
1171 delete_extra: bool = False,
1172 ) -> None:
1173 """
1174 Import settings directly from the export format. This can be used to
1175 re-import settings that have been exported with `get_all_settings()`.
1177 Args:
1178 settings_data: The raw settings data to import.
1179 commit: Whether to commit the DB after loading the settings.
1180 overwrite: If true, it will overwrite the value of settings that
1181 are already in the database.
1182 delete_extra: If true, it will delete any settings that are in
1183 the database but don't have a corresponding entry in
1184 `settings_data`.
1186 """
1187 if self.db_session is None: 1187 ↛ 1188line 1187 didn't jump to line 1188 because the condition on line 1187 was never true
1188 raise RuntimeError("Database session is not initialized")
1189 logger.debug(f"Importing {len(settings_data)} settings")
1191 for key, setting_values in settings_data.items():
1192 setting_values = dict(setting_values)
1193 if not overwrite:
1194 existing_value = self.get_setting(key)
1195 if existing_value is not None:
1196 # Preserve the value from this setting.
1197 setting_values["value"] = existing_value
1199 # Delete any existing setting so we can completely overwrite it.
1200 self.delete_setting(key, commit=False)
1202 # Convert type string to SettingType enum if needed
1203 if "type" in setting_values and isinstance( 1203 ↛ 1208line 1203 didn't jump to line 1208 because the condition on line 1203 was always true
1204 setting_values["type"], str
1205 ):
1206 setting_values["type"] = SettingType[setting_values["type"]]
1208 setting = Setting(
1209 key=key, **_filter_setting_columns(setting_values)
1210 )
1211 self.db_session.add(setting)
1213 if commit or delete_extra:
1214 self.db_session.commit()
1215 logger.info(f"Successfully imported {len(settings_data)} settings")
1216 # Emit WebSocket event for all imported settings
1217 self._emit_settings_changed(list(settings_data.keys()))
1219 if delete_extra:
1220 all_settings = self.get_all_settings()
1221 for key in all_settings:
1222 if key not in settings_data:
1223 logger.debug(f"Deleting extraneous setting: {key}")
1224 self.delete_setting(key, commit=False)
1226 def _create_setting(self, key, value, setting_type):
1227 """Create a setting with appropriate metadata"""
1229 # Determine appropriate category
1230 category = None
1231 ui_element = "text"
1233 # Determine category based on key pattern
1234 if key.startswith("app."):
1235 category = "app_interface"
1236 elif key.startswith("llm."):
1237 if any(
1238 param in key
1239 for param in [
1240 "temperature",
1241 "max_tokens",
1242 "n_batch",
1243 "n_gpu_layers",
1244 ]
1245 ):
1246 category = "llm_parameters"
1247 else:
1248 category = "llm_general"
1249 elif key.startswith("search."):
1250 if any(
1251 param in key
1252 for param in ["iterations", "questions", "results", "region"]
1253 ):
1254 category = "search_parameters"
1255 else:
1256 category = "search_general"
1257 elif key.startswith("report."):
1258 category = "report_parameters"
1260 # Determine UI element type based on value
1261 ui_element = _infer_ui_element(value)
1263 # Build setting object
1264 setting_dict = {
1265 "key": key,
1266 "value": value,
1267 "type": setting_type.value.lower(),
1268 "name": key.split(".")[-1].replace("_", " ").title(),
1269 "description": f"Setting for {key}",
1270 "category": category,
1271 "ui_element": ui_element,
1272 }
1274 # Create the setting in the database
1275 self.create_or_update_setting(setting_dict, commit=False)
1277 def _emit_settings_changed(self, changed_keys: Optional[List[Any]] = None):
1278 """
1279 Emit WebSocket event when settings change
1281 Args:
1282 changed_keys: List of setting keys that changed
1283 """
1284 try:
1285 # Import here to avoid circular imports
1286 from ..web.services.socket_service import SocketIOService
1288 try:
1289 socket_service = SocketIOService()
1290 except ValueError:
1291 logger.debug(
1292 "Not emitting socket event because server is not initialized."
1293 )
1294 return
1296 # Get the changed settings
1297 settings_data = {}
1298 if changed_keys: 1298 ↛ 1305line 1298 didn't jump to line 1305 because the condition on line 1298 was always true
1299 for key in changed_keys:
1300 setting_value = self.get_setting(key)
1301 if setting_value is not None:
1302 settings_data[key] = {"value": setting_value}
1304 # Emit the settings change event
1305 from datetime import datetime, UTC
1307 socket_service.emit_socket_event(
1308 "settings_changed",
1309 {
1310 "changed_keys": changed_keys or [],
1311 "settings": settings_data,
1312 "timestamp": datetime.now(UTC).isoformat(),
1313 },
1314 )
1316 logger.debug(
1317 f"Emitted settings_changed event for keys: {changed_keys}"
1318 )
1320 except Exception:
1321 logger.exception("Failed to emit settings change event")
1322 # Don't let WebSocket emission failures break settings saving
1324 @staticmethod
1325 def get_bootstrap_env_vars() -> Dict[str, str]:
1326 """
1327 Get environment variables that must be available before database access.
1328 These are critical for system initialization.
1330 Returns:
1331 Dict mapping env var names to their descriptions
1332 """
1333 # Get bootstrap vars from env registry
1334 return env_registry.get_bootstrap_vars()
1336 @staticmethod
1337 def is_bootstrap_env_var(env_var: str) -> bool:
1338 """
1339 Check if an environment variable is a bootstrap variable (needed before DB access).
1341 Args:
1342 env_var: Environment variable name
1344 Returns:
1345 True if this is a bootstrap variable
1346 """
1347 bootstrap_vars = SettingsManager.get_bootstrap_env_vars()
1348 return env_var in bootstrap_vars
1350 @staticmethod
1351 def is_env_only_setting(key: str) -> bool:
1352 """
1353 Check if a setting key is environment-only.
1355 Args:
1356 key: Setting key to check
1358 Returns:
1359 True if it's an env-only setting, False otherwise
1360 """
1361 return env_registry.is_env_only(key)
1363 @staticmethod
1364 def get_env_var_for_setting(setting_key: str) -> str:
1365 """
1366 Get the environment variable name for a given setting key.
1368 Args:
1369 setting_key: Setting key (e.g., "app.host")
1371 Returns:
1372 Environment variable name (e.g., "LDR_APP_HOST")
1373 """
1374 # Use the same logic as check_env_setting for consistency
1375 return f"LDR_{'_'.join(setting_key.split('.')).upper()}"
1377 @staticmethod
1378 def get_setting_key_for_env_var(env_var: str) -> Optional[str]:
1379 """
1380 Get the setting key for a given environment variable.
1382 Args:
1383 env_var: Environment variable name (e.g., "LDR_APP_HOST")
1385 Returns:
1386 Setting key (e.g., "app.host") or None if not a valid LDR env var
1387 """
1388 if not env_var.startswith("LDR_"):
1389 return None
1391 # Remove LDR_ prefix and convert to lowercase
1392 without_prefix = env_var[4:]
1393 parts = without_prefix.split("_")
1395 return ".".join(part.lower() for part in parts)
1398class SnapshotSettingsContext:
1399 """Read-only settings context backed by a snapshot dict.
1401 Unwraps {"value": x} setting objects into plain values and provides
1402 get_setting(key, default) for thread-safe snapshot access.
1403 """
1405 def __init__(
1406 self, snapshot=None, username=None, missing_key_log_level="DEBUG"
1407 ):
1408 self.snapshot = snapshot or {}
1409 self.username = username
1410 self._missing_key_log_level = missing_key_log_level
1411 self.values = {}
1412 for key, setting in self.snapshot.items():
1413 if isinstance(setting, dict) and "value" in setting:
1414 self.values[key] = setting["value"]
1415 else:
1416 self.values[key] = setting
1418 def get_setting(self, key, default=None):
1419 """Return the setting value for *key*, or *default* if absent."""
1420 if key in self.values:
1421 return self.values[key]
1422 logger.log(
1423 self._missing_key_log_level,
1424 "Setting '{}' not found in snapshot, using default",
1425 key,
1426 )
1427 return default