Coverage for src / local_deep_research / settings / manager.py: 92%
491 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1import functools
2import json
3import os
4import threading
5from pathlib import Path
6from typing import Any, Callable, Dict, List, Optional, Union
8from loguru import logger
9from sqlalchemy import func, or_
10from sqlalchemy.exc import SQLAlchemyError
11from sqlalchemy.orm import Session
13from .. import defaults
14from ..__version__ import __version__ as package_version
15from ..database.models import Setting, SettingType
16from ..web.models.settings import (
17 AppSetting,
18 BaseSetting,
19 LLMSetting,
20 ReportSetting,
21 SearchSetting,
22)
23from ..utilities.type_utils import to_bool
24from .base import ISettingsManager
25from .env_registry import registry as env_registry
28def parse_boolean(value: Any) -> bool:
29 """
30 Convert various representations to boolean using HTML checkbox semantics.
32 This function handles form values, JSON booleans, and environment variables,
33 ensuring consistent behavior across client and server.
35 **HTML Checkbox Semantics** (INTENTIONAL DESIGN):
36 - **Any value present (except explicit false) = checked = True**
37 - This matches standard HTML form behavior where checkbox presence indicates checked state
38 - In HTML forms, checkboxes send a value when checked, nothing when unchecked
40 **Examples**:
41 parse_boolean("on") # True - standard HTML checkbox value
42 parse_boolean("true") # True - explicit true
43 parse_boolean("1") # True - numeric true
44 parse_boolean("enabled") # True - any non-empty string
45 parse_boolean("disabled") # True - INTENTIONAL: any string = checkbox was checked!
46 parse_boolean("custom") # True - custom checkbox value
48 parse_boolean("false") # False - explicit false
49 parse_boolean("off") # False - explicit false
50 parse_boolean("0") # False - explicit false
51 parse_boolean("") # False - empty string = unchecked
52 parse_boolean(None) # False - missing = unchecked
54 **Why "disabled" returns True**:
55 This is NOT a bug! If a checkbox sends the value "disabled", it means the checkbox
56 was checked (present in form data). The actual string content doesn't matter for
57 HTML checkboxes - only presence vs absence matters.
59 Args:
60 value: Value to convert to boolean. Accepts strings, booleans, or None.
62 Returns:
63 bool: True for truthy values (any non-empty string except explicit false);
64 False for falsy values ('off', 'false', '0', '', 'no', False, None)
66 Note:
67 This function implements HTML form semantics, NOT generic boolean parsing.
68 See tests/settings/test_boolean_parsing.py for comprehensive test coverage.
69 """
70 # Constants for boolean value parsing
71 FALSY_VALUES = ("off", "false", "0", "", "no")
73 # Handle already-boolean values
74 if isinstance(value, bool):
75 return value
77 # Handle None (missing values)
78 if value is None:
79 return False
81 # Handle string values
82 if isinstance(value, str):
83 value_lower = value.lower().strip()
84 # Explicitly falsy values (empty string, false-like values)
85 if value_lower in FALSY_VALUES:
86 return False
87 # Any other non-empty string = True (HTML checkbox semantics)
88 return True
90 # For other types (numbers, lists, etc.), use Python's bool conversion
91 return bool(value)
94def _parse_number(x):
95 """Parse number, returning int if it's a whole number, otherwise float."""
96 f = float(x)
97 if f.is_integer():
98 return int(f)
99 return f
102def _parse_json_value(x):
103 """Parse JSON ui_element values.
105 DB values (via SQLAlchemy JSON column) arrive as Python objects already.
106 Form POST and env var overrides arrive as raw strings and need parsing.
107 For example, a textarea containing ``["general"]`` arrives as the string
108 ``'[\\r\\n "general"\\r\\n]'`` which must be decoded into a list.
109 """
110 if isinstance(x, str):
111 stripped = x.strip()
112 if stripped:
113 try:
114 return json.loads(stripped)
115 except (json.JSONDecodeError, ValueError, RecursionError):
116 logger.warning("Failed to parse JSON value, returning raw")
117 return x
118 return x
121def _parse_multiselect(x):
122 """Parse multiselect value, handling both lists and strings.
124 DB values (via SQLAlchemy JSON column) arrive as Python lists already.
125 Env var overrides arrive as strings and need parsing — either as JSON
126 arrays (e.g. '["markdown","latex"]') or comma-separated values
127 (e.g. 'markdown,latex').
128 """
129 if isinstance(x, list):
130 return x
131 if isinstance(x, str):
132 stripped = x.strip()
133 if stripped.startswith("["):
134 try:
135 parsed = json.loads(stripped)
136 if isinstance(parsed, list): 136 ↛ 141line 136 didn't jump to line 141 because the condition on line 136 was always true
137 return parsed
138 except (json.JSONDecodeError, ValueError):
139 pass
140 # Comma-separated fallback
141 return [item.strip() for item in stripped.split(",") if item.strip()]
142 return x
145def _filter_setting_columns(data: dict) -> dict:
146 """Filter a dict to only keys that are valid Setting model columns.
148 Prevents crashes when default_settings.json contains keys not present
149 as columns on the Setting model (e.g. future flags).
150 """
151 valid_columns = {c.name for c in Setting.__table__.columns}
152 return {k: v for k, v in data.items() if k in valid_columns}
155def _infer_ui_element(value: Any, current: str = "text") -> str:
156 """Infer the appropriate ui_element string from a Python value's type.
158 Args:
159 value: The value to infer the ui_element from.
160 current: The existing ui_element. If it is already something more
161 specific than ``"text"``, it is kept as-is.
162 """
163 if current != "text":
164 return current
165 if isinstance(value, bool):
166 return "checkbox"
167 if isinstance(value, (int, float)):
168 return "number"
169 if isinstance(value, (list, dict)):
170 return "json"
171 return "text"
174UI_ELEMENT_TO_SETTING_TYPE: Dict[str, Callable[..., Any]] = {
175 "text": str,
176 "json": _parse_json_value,
177 "password": str,
178 "select": str,
179 "number": _parse_number,
180 "range": _parse_number, # Same behavior as number for consistency
181 "checkbox": parse_boolean,
182 "textarea": str,
183 "multiselect": _parse_multiselect,
184}
187def get_typed_setting_value(
188 key: str,
189 value: Any,
190 ui_element: str,
191 default: Any = None,
192 check_env: bool = True,
193) -> Any:
194 """
195 Extracts the value for a particular setting, ensuring that it has the
196 correct type.
198 Args:
199 key: The setting key.
200 value: The setting value from the database.
201 ui_element: The setting UI element ID.
202 default: Default value to return if the value of the setting is
203 invalid.
204 check_env: If true, it will check the environment variable for
205 this setting before reading from the DB.
207 Returns:
208 The value of the setting.
210 """
211 setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None)
212 if setting_type is None:
213 logger.warning(
214 "Got unknown type {} for setting {}, returning default value.",
215 ui_element,
216 key,
217 )
218 return default
220 # Check environment variable first (highest priority).
221 if check_env:
222 env_value = check_env_setting(key)
223 if env_value is not None:
224 try:
225 return setting_type(env_value)
226 except ValueError:
227 logger.warning(
228 "Setting {} has invalid value {}. Falling back to DB.",
229 key,
230 env_value,
231 )
233 # If value is None (not in database), return default.
234 if value is None:
235 return default
237 # Read from the database.
238 try:
239 return setting_type(value)
240 except (ValueError, TypeError):
241 logger.warning(
242 "Setting {} has invalid value {}. Returning default.",
243 key,
244 value,
245 )
246 return default
249def check_env_setting(key: str) -> str | None:
250 """
251 Checks environment variables for a particular setting.
253 Args:
254 key: The database key for the setting.
256 Returns:
257 The setting from the environment variables, or None if the variable
258 is not set or is empty.
260 Note:
261 Empty environment variables ("") are treated as unset. This is standard
262 practice across the ecosystem — see CPython's official docs (PYTHON*
263 env vars require "a non-empty string"), botocore PR #1687, Pallets/Click
264 PR #2223, and Vercel Turborepo PR #6929. Orchestration tools like Unraid,
265 Terraform, and Kubernetes manifests often cannot conditionally omit env
266 var declarations, so they pass "" for unconfigured values. Treating ""
267 as unset prevents these empty strings from overriding database defaults.
268 See: https://github.com/LearningCircuit/local-deep-research/pull/3362
270 """
271 env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}"
272 env_value = os.getenv(env_variable_name)
273 # Treat empty string as unset — orchestration tools (Unraid, Terraform, K8s)
274 # often cannot omit env var declarations and pass "" for unconfigured values.
275 if env_value is not None and env_value != "":
276 logger.debug(f"Overriding {key} setting from environment variable.")
277 return env_value
278 if env_value == "":
279 logger.warning(
280 "Environment variable {} is set but empty — "
281 "ignoring it and falling back to DB/default for setting '{}'. "
282 "This is expected on Unraid or Docker templates that create "
283 "all variables even when left blank. To suppress this warning, "
284 "remove the variable from your environment or set a value.",
285 env_variable_name,
286 key,
287 )
288 return None
291class SettingsManager(ISettingsManager):
292 """
293 Manager for handling application settings with database storage and file fallback.
294 Provides methods to get and set settings, with the ability to override settings in memory.
295 """
297 def __init__(
298 self,
299 db_session: Optional[Session] = None,
300 owns_session: bool = False,
301 ):
302 """
303 Initialize the settings manager
305 Args:
306 db_session: SQLAlchemy session for database operations
307 owns_session: If True, close() will close the session.
308 Defaults to False (safe for borrowed sessions). Set to True
309 only when this manager created/owns the session — currently
310 only get_settings_manager() in db_utils.py does this.
311 """
312 self.db_session = db_session
313 self._owns_session = owns_session
314 self._closed = False
315 self.db_first = True # Always prioritize DB settings
317 # Store the thread ID this instance was created in
318 self._creation_thread_id = threading.get_ident()
320 # Initialize settings lock as None - will be checked lazily
321 self.__settings_locked: Optional[bool] = None
323 # Auto-initialize settings if database is empty
324 if self.db_session:
325 self._ensure_settings_initialized()
327 def close(self):
328 """Close the DB session if this manager owns it.
330 Borrowed sessions (owns_session=False) are left open for their
331 owner to close (e.g. Flask teardown closes g.db_session).
332 Safe to call multiple times — subsequent calls are no-ops.
333 """
334 if self._owns_session and self.db_session is not None:
335 try:
336 logger.debug("Closing owned DB session in SettingsManager")
337 self.db_session.close()
338 except Exception:
339 logger.warning(
340 "Failed to close SettingsManager DB session — "
341 "connection may leak",
342 )
343 self._closed = True
344 self.db_session = None
346 def _ensure_settings_initialized(self):
347 """Ensure settings are initialized in the database."""
348 # Check if we have any settings at all
349 from ..database.models import Setting
351 if self.db_session is None: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 raise RuntimeError("Database session is not initialized")
353 settings_count = self.db_session.query(Setting).count()
355 if settings_count == 0:
356 logger.info("No settings found in database, loading defaults")
357 self.load_from_defaults_file(commit=True)
358 logger.info("Default settings loaded successfully")
360 def _check_thread_safety(self):
361 """Check if this instance is being used in the same thread it was created in."""
362 current_thread_id = threading.get_ident()
363 if self.db_session and current_thread_id != self._creation_thread_id:
364 raise RuntimeError(
365 f"SettingsManager instance created in thread {self._creation_thread_id} "
366 f"is being used in thread {current_thread_id}. This is not thread-safe! "
367 f"Create a new SettingsManager instance within the current thread context."
368 )
370 @property
371 def settings_locked(self) -> bool:
372 """Check if settings are locked (lazy evaluation)."""
373 if self.__settings_locked is None:
374 try:
375 self.__settings_locked = self.get_setting(
376 "app.lock_settings", False
377 )
378 if self.settings_locked:
379 logger.info(
380 "Settings are locked. Disabling all settings changes."
381 )
382 except Exception:
383 logger.warning(
384 "Failed to check settings lock status, assuming not locked"
385 )
386 self.__settings_locked = False
387 return bool(self.__settings_locked)
389 @functools.cached_property
390 def default_settings(self) -> Dict[str, Any]:
391 """
392 Returns:
393 The default settings, loaded from JSON files and merged.
394 Automatically discovers and loads all .json files in the defaults
395 directory and its subdirectories.
396 Theme options are dynamically injected from the theme registry.
398 """
399 settings: Dict[str, Any] = {}
401 try:
402 # Get the defaults package path
403 defaults_path = Path(defaults.__file__).parent
405 # Find all JSON files recursively in the defaults directory
406 json_files = sorted(defaults_path.rglob("*.json"))
408 logger.debug(f"Found {len(json_files)} JSON settings files")
410 # Load and merge all JSON files
411 for json_file in json_files:
412 try:
413 with open(json_file, "r") as f:
414 file_settings = json.load(f)
416 # Get relative path for logging
417 relative_path = json_file.relative_to(defaults_path)
419 # Warn about key conflicts
420 conflicts = set(settings.keys()) & set(file_settings.keys())
421 if conflicts:
422 logger.warning(
423 f"Keys {conflicts} from {relative_path} "
424 f"override existing values"
425 )
427 settings.update(file_settings)
428 logger.debug(f"Loaded {relative_path}")
430 except json.JSONDecodeError:
431 logger.exception(f"Invalid JSON in {json_file}")
432 except Exception:
433 logger.warning(f"Could not load {json_file}")
435 except Exception:
436 logger.warning("Error loading settings files")
438 # Inject dynamic theme options from theme registry
439 if "app.theme" in settings:
440 try:
441 from local_deep_research.web.themes import theme_registry
443 settings["app.theme"]["options"] = (
444 theme_registry.get_settings_options()
445 )
446 except ImportError:
447 # Theme registry not available, use static options from JSON
448 pass
450 # Inject search strategy options from code (single source of truth)
451 if "search.search_strategy" in settings:
452 from local_deep_research.constants import get_available_strategies
454 # Check the show_all_strategies setting to decide which list
455 show_all = False
456 if "search.show_all_strategies" in settings: 456 ↛ 460line 456 didn't jump to line 460 because the condition on line 456 was always true
457 val = settings["search.show_all_strategies"].get("value", False)
458 show_all = val is True or val == "true"
460 strategies = get_available_strategies(show_all=show_all)
461 settings["search.search_strategy"]["options"] = [
462 {"label": s["label"], "value": s["name"]} for s in strategies
463 ]
465 logger.debug(f"Loaded {len(settings)} total settings")
466 return settings
468 def __get_typed_setting_value(
469 self,
470 setting: Setting,
471 default: Any = None,
472 check_env: bool = True,
473 ) -> Any:
474 """
475 Extracts the value for a particular setting, ensuring that it has the
476 correct type.
478 Args:
479 setting: The setting to get the value for.
480 default: Default value to return if the value of the setting is
481 invalid.
482 check_env: If true, it will check the environment variable for
483 this setting before reading from the DB.
485 Returns:
486 The value of the setting.
488 """
489 return get_typed_setting_value(
490 str(setting.key),
491 setting.value,
492 str(setting.ui_element),
493 default=default,
494 check_env=check_env,
495 )
497 def __query_settings(self, key: str | None = None) -> List[Setting]:
498 """
499 Abstraction for querying settings that also transparently handles
500 reading the default settings file if the DB is not enabled.
502 Args:
503 key: The key to read. If None, it will read everything.
505 Returns:
506 The settings it queried.
508 """
509 if self.db_session:
510 self._check_thread_safety()
511 query = self.db_session.query(Setting)
512 if key is not None:
513 # This will find exact matches and any subkeys.
514 query = query.filter(
515 or_(
516 Setting.key == key,
517 Setting.key.startswith(f"{key}."),
518 )
519 )
520 return query.all()
522 logger.debug(
523 "DB is disabled, reading setting '{}' from defaults file.", key
524 )
526 settings = []
527 for candidate_key, setting in self.default_settings.items():
528 if key is None or (
529 candidate_key == key or candidate_key.startswith(f"{key}.")
530 ):
531 settings.append(
532 Setting(
533 key=candidate_key, # gitleaks:allow
534 **_filter_setting_columns(setting),
535 )
536 )
538 return settings
540 def get_setting(
541 self, key: str, default: Any = None, check_env: bool = True
542 ) -> Any:
543 """
544 Get a setting value
546 Args:
547 key: Setting key
548 default: Default value if setting is not found
549 check_env: If true, it will check the environment variable for
550 this setting before reading from the DB.
552 Returns:
553 Setting value or default if not found
554 """
555 if self._closed: 555 ↛ 556line 555 didn't jump to line 556 because the condition on line 555 was never true
556 logger.error(
557 "SettingsManager.get_setting('{}') called after close() — "
558 "this is a bug; the caller should not reuse a closed manager",
559 key,
560 )
561 raise RuntimeError(
562 "SettingsManager has been closed. "
563 "Create a new instance or call close() only at end of lifecycle."
564 )
566 # First check if this is an env-only setting
567 if env_registry.is_env_only(key):
568 return env_registry.get(key, default)
570 # If using database first approach and session available, check database
571 try:
572 settings = self.__query_settings(key)
573 if len(settings) == 1:
574 # This is a bottom-level key.
575 return self.__get_typed_setting_value(
576 settings[0], default, check_env
577 )
578 # Cache the result
579 if len(settings) > 1:
580 # This is a higher-level key.
581 settings_map = {}
582 for setting in settings:
583 output_key = str(setting.key).removeprefix(f"{key}.")
584 settings_map[output_key] = self.__get_typed_setting_value(
585 setting, default, check_env
586 )
587 return settings_map
588 except SQLAlchemyError:
589 logger.exception(f"Error retrieving setting {key} from database")
591 # Check env var before returning default (setting not in DB)
592 if check_env:
593 env_value = check_env_setting(key)
594 if env_value is not None:
595 default_meta = self.default_settings.get(key)
596 if default_meta and isinstance(default_meta, dict):
597 ui_element = default_meta.get("ui_element", "text")
598 return get_typed_setting_value(
599 key,
600 None,
601 ui_element,
602 default=default,
603 check_env=True,
604 )
605 logger.warning(
606 "Setting '{}' has env var override but is not in "
607 "defaults — returning raw string without type "
608 "conversion. Add this setting to a defaults JSON "
609 "file with a ui_element type to enable proper "
610 "type conversion.",
611 key,
612 )
613 return env_value
615 # Return default if not found
616 return default
618 def get_bool_setting(
619 self, key: str, default: bool = False, check_env: bool = True
620 ) -> bool:
621 """
622 Get a setting value as a boolean, handling string conversion.
624 Args:
625 key: Setting key
626 default: Default boolean value if setting is not found
627 check_env: If true, it will check the environment variable for
628 this setting before reading from the DB.
630 Returns:
631 Boolean value of the setting
632 """
633 value = self.get_setting(key, default, check_env)
634 return to_bool(value, default)
636 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
637 """
638 Set a setting value
640 Args:
641 key: Setting key
642 value: Setting value
643 commit: Whether to commit the change
645 Returns:
646 True if successful, False otherwise
647 """
648 if self._closed: 648 ↛ 649line 648 didn't jump to line 649 because the condition on line 648 was never true
649 logger.error(
650 "SettingsManager.set_setting('{}') called after close() — "
651 "this is a bug; the caller should not reuse a closed manager",
652 key,
653 )
654 raise RuntimeError(
655 "SettingsManager has been closed. "
656 "Create a new instance or call close() only at end of lifecycle."
657 )
658 if not self.db_session:
659 logger.error(
660 "Cannot edit setting {} because no DB was provided.", key
661 )
662 return False
663 if self.settings_locked:
664 logger.error("Cannot edit setting {} because they are locked.", key)
665 return False
667 # Always update database if available
668 try:
669 self._check_thread_safety()
670 setting = (
671 self.db_session.query(Setting)
672 .filter(Setting.key == key)
673 .first()
674 )
675 if setting:
676 if not setting.editable:
677 logger.error(
678 "Cannot change setting '{}' because it "
679 "is marked as non-editable.",
680 key,
681 )
682 return False
684 setting.value = value # type: ignore[assignment]
685 setting.updated_at = ( # type: ignore[assignment]
686 func.now()
687 ) # Explicitly set the current timestamp
689 # Self-heal stale ui_element from before inference was added
690 setting.ui_element = _infer_ui_element(
691 value, setting.ui_element
692 )
693 else:
694 # Determine setting type from key
695 setting_type = SettingType.APP
696 if key.startswith("llm."):
697 setting_type = SettingType.LLM
698 elif key.startswith("search."):
699 setting_type = SettingType.SEARCH
700 elif key.startswith("report."):
701 setting_type = SettingType.REPORT
702 elif key.startswith("database."): 702 ↛ 703line 702 didn't jump to line 703 because the condition on line 702 was never true
703 setting_type = SettingType.DATABASE
705 # Infer ui_element from the value type
706 ui_element = _infer_ui_element(value)
708 # Create a new setting
709 new_setting = Setting(
710 key=key,
711 value=value,
712 type=setting_type,
713 name=key.split(".")[-1].replace("_", " ").title(),
714 ui_element=ui_element,
715 description=f"Setting for {key}",
716 )
717 self.db_session.add(new_setting)
719 if commit:
720 self.db_session.commit()
721 # Emit WebSocket event for settings change
722 self._emit_settings_changed([key])
724 return True
725 except SQLAlchemyError:
726 logger.exception(f"Error setting value for key: {key}")
727 self.db_session.rollback()
728 return False
730 def clear_cache(self):
731 """Clear the settings cache."""
732 self.__dict__.pop("default_settings", None)
733 logger.debug("Settings cache cleared")
735 def get_all_settings(self, bypass_cache: bool = False) -> Dict[str, Any]:
736 """
737 Get all settings, merging defaults with database values.
739 This ensures that new settings added to defaults.json automatically
740 appear in the UI without requiring a database reset.
742 Args:
743 bypass_cache: If True, bypass the cache and read directly from database
745 Returns:
746 Dictionary of all settings
747 """
748 if self._closed: 748 ↛ 749line 748 didn't jump to line 749 because the condition on line 748 was never true
749 logger.error(
750 "SettingsManager.get_all_settings() called after close() — "
751 "this is a bug; the caller should not reuse a closed manager",
752 )
753 raise RuntimeError(
754 "SettingsManager has been closed. "
755 "Create a new instance or call close() only at end of lifecycle."
756 )
758 result = {}
760 # Start with defaults so new settings are always included
761 for key, default_setting in self.default_settings.items():
762 result[key] = dict(default_setting)
764 # Check env var override for defaults not yet in DB
765 env_value = check_env_setting(key)
766 if env_value is not None: 766 ↛ 767line 766 didn't jump to line 767 because the condition on line 766 was never true
767 ui_element = default_setting.get("ui_element", "text")
768 typed_value = get_typed_setting_value(
769 key,
770 None,
771 ui_element,
772 default=env_value,
773 check_env=True,
774 )
775 result[key]["value"] = typed_value
776 result[key]["editable"] = False
778 # Override with database settings
779 try:
780 db_settings = self.__query_settings()
781 except SQLAlchemyError:
782 logger.exception(
783 "Error querying settings from database in get_all_settings"
784 )
785 db_settings = []
787 for setting in db_settings:
788 # Handle type field - it might be a string or an enum
789 setting_type = setting.type
790 if hasattr(setting_type, "name"):
791 setting_type = setting_type.name
793 # Log if this is a custom setting not in defaults
794 if str(setting.key) not in result:
795 logger.debug(
796 f"Database contains custom setting not in "
797 f"defaults: {setting.key} (type={setting_type}, "
798 f"category={setting.category})"
799 )
801 # Override default with database value
802 result[str(setting.key)] = {
803 "value": setting.value,
804 "type": setting_type,
805 "name": setting.name,
806 "description": setting.description,
807 "category": setting.category,
808 "ui_element": setting.ui_element,
809 "options": setting.options,
810 "min_value": setting.min_value,
811 "max_value": setting.max_value,
812 "step": setting.step,
813 "visible": setting.visible,
814 "editable": False if self.settings_locked else setting.editable,
815 }
817 # Override from the environment variables if needed.
818 env_value = check_env_setting(str(setting.key))
819 if env_value is not None:
820 ui_element = result[str(setting.key)].get(
821 "ui_element", setting.ui_element
822 )
823 typed_value = get_typed_setting_value(
824 str(setting.key),
825 None,
826 ui_element,
827 default=env_value,
828 check_env=True,
829 )
830 result[str(setting.key)]["value"] = typed_value
831 # Mark it as non-editable, because changes to the DB
832 # value have no effect as long as the environment
833 # variable is set.
834 result[str(setting.key)]["editable"] = False
836 # Re-inject search strategy options from code after DB merge,
837 # since the DB stores options=null for this setting.
838 if "search.search_strategy" in result:
839 from local_deep_research.constants import get_available_strategies
841 show_all_val = result.get("search.show_all_strategies", {}).get(
842 "value", False
843 )
844 show_all = show_all_val is True or show_all_val == "true"
845 strategies = get_available_strategies(show_all=show_all)
846 result["search.search_strategy"]["options"] = [
847 {"label": s["label"], "value": s["name"]} for s in strategies
848 ]
850 return result
852 def get_settings_snapshot(self) -> Dict[str, Any]:
853 """
854 Get a simplified settings snapshot with just key-value pairs.
855 This is useful for passing settings to background threads or storing in metadata.
857 Returns:
858 Dictionary with setting keys mapped to their values
859 """
860 if self._closed: 860 ↛ 861line 860 didn't jump to line 861 because the condition on line 860 was never true
861 logger.error(
862 "SettingsManager.get_settings_snapshot() called after close() — "
863 "this is a bug; the caller should not reuse a closed manager",
864 )
865 raise RuntimeError(
866 "SettingsManager has been closed. "
867 "Create a new instance or call close() only at end of lifecycle."
868 )
870 all_settings = self.get_all_settings()
871 settings_snapshot = {}
873 for key, setting in all_settings.items():
874 if isinstance(setting, dict) and "value" in setting: 874 ↛ 877line 874 didn't jump to line 877 because the condition on line 874 was always true
875 settings_snapshot[key] = setting["value"]
876 else:
877 settings_snapshot[key] = setting
879 return settings_snapshot
881 def create_or_update_setting(
882 self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True
883 ) -> Optional[Setting]:
884 """
885 Create or update a setting
887 Args:
888 setting: Setting object or dictionary
889 commit: Whether to commit the change
891 Returns:
892 The created or updated Setting model, or None if failed
893 """
894 if not self.db_session:
895 logger.warning(
896 "No database session available, cannot create/update setting"
897 )
898 return None
899 if self.settings_locked:
900 logger.error("Cannot edit settings because they are locked.")
901 return None
903 # Convert dict to BaseSetting if needed
904 if isinstance(setting, dict): 904 ↛ 921line 904 didn't jump to line 921 because the condition on line 904 was always true
905 # Determine type from key if not specified
906 if "type" not in setting and "key" in setting:
907 setting_obj: BaseSetting
908 key = setting["key"]
909 if key.startswith("llm."):
910 setting_obj = LLMSetting(**setting)
911 elif key.startswith("search."):
912 setting_obj = SearchSetting(**setting)
913 elif key.startswith("report."):
914 setting_obj = ReportSetting(**setting)
915 else:
916 setting_obj = AppSetting(**setting)
917 else:
918 # Use generic BaseSetting
919 setting_obj = BaseSetting(**setting)
920 else:
921 setting_obj = setting
923 try:
924 # Check if setting exists
925 db_setting = (
926 self.db_session.query(Setting)
927 .filter(Setting.key == setting_obj.key)
928 .first()
929 )
931 if db_setting:
932 # Update existing setting
933 if not db_setting.editable:
934 logger.error(
935 "Cannot change setting '{}' because it "
936 "is marked as non-editable.",
937 setting_obj.key,
938 )
939 return None
941 db_setting.value = setting_obj.value # type: ignore[assignment]
942 db_setting.name = setting_obj.name # type: ignore[assignment]
943 db_setting.description = setting_obj.description # type: ignore[assignment]
944 db_setting.category = setting_obj.category # type: ignore[assignment]
945 db_setting.ui_element = setting_obj.ui_element # type: ignore[assignment]
946 db_setting.options = setting_obj.options # type: ignore[assignment]
947 db_setting.min_value = setting_obj.min_value # type: ignore[assignment]
948 db_setting.max_value = setting_obj.max_value # type: ignore[assignment]
949 db_setting.step = setting_obj.step # type: ignore[assignment]
950 db_setting.visible = setting_obj.visible # type: ignore[assignment]
951 db_setting.editable = setting_obj.editable # type: ignore[assignment]
952 db_setting.updated_at = ( # type: ignore[assignment]
953 func.now()
954 ) # Explicitly set the current timestamp
955 else:
956 # Create new setting
957 db_setting = Setting(
958 key=setting_obj.key,
959 value=setting_obj.value,
960 type=setting_obj.type,
961 name=setting_obj.name,
962 description=setting_obj.description,
963 category=setting_obj.category,
964 ui_element=setting_obj.ui_element,
965 options=setting_obj.options,
966 min_value=setting_obj.min_value,
967 max_value=setting_obj.max_value,
968 step=setting_obj.step,
969 visible=setting_obj.visible,
970 editable=setting_obj.editable,
971 )
972 self.db_session.add(db_setting)
974 if commit:
975 self.db_session.commit()
976 # Emit WebSocket event for settings change
977 self._emit_settings_changed([setting_obj.key])
979 return db_setting
981 except SQLAlchemyError:
982 logger.exception(
983 f"Error creating/updating setting {setting_obj.key}"
984 )
985 self.db_session.rollback()
986 return None
988 def delete_setting(self, key: str, commit: bool = True) -> bool:
989 """
990 Delete a setting
992 Args:
993 key: Setting key
994 commit: Whether to commit the change
996 Returns:
997 True if successful, False otherwise
998 """
999 if not self.db_session:
1000 logger.warning(
1001 "No database session available, cannot delete setting"
1002 )
1003 return False
1005 try:
1006 # Remove from database
1007 result = (
1008 self.db_session.query(Setting)
1009 .filter(Setting.key == key)
1010 .delete()
1011 )
1013 if commit:
1014 self.db_session.commit()
1016 return result > 0
1017 except SQLAlchemyError:
1018 logger.exception("Error deleting setting")
1019 self.db_session.rollback()
1020 return False
1022 def load_from_defaults_file(
1023 self, commit: bool = True, **kwargs: Any
1024 ) -> None:
1025 """
1026 Import settings from the defaults settings file.
1028 Args:
1029 commit: Whether to commit changes to database
1030 **kwargs: Will be passed to `import_settings`.
1032 """
1033 self.import_settings(self.default_settings, commit=commit, **kwargs)
1035 def db_version_matches_package(self) -> bool:
1036 """
1037 Returns:
1038 True if the version saved in the DB matches the package version.
1040 """
1041 db_version = self.get_setting("app.version")
1042 logger.debug(
1043 f"App version saved in DB is {db_version}, have package "
1044 f"settings from version {package_version}."
1045 )
1047 return bool(db_version == package_version)
1049 def update_db_version(self) -> None:
1050 """
1051 Updates the version saved in the DB based on the package version.
1053 """
1054 logger.debug(f"Updating saved DB version to {package_version}.")
1056 self.delete_setting("app.version", commit=False)
1057 version = Setting(
1058 key="app.version",
1059 value=package_version,
1060 description="Version of the app this database is associated with.",
1061 editable=False,
1062 name="App Version",
1063 type=SettingType.APP,
1064 ui_element="text",
1065 visible=False,
1066 )
1068 if self.db_session is None: 1068 ↛ 1069line 1068 didn't jump to line 1069 because the condition on line 1068 was never true
1069 raise RuntimeError("Database session is not initialized")
1070 self.db_session.add(version)
1071 self.db_session.commit()
1073 def import_settings(
1074 self,
1075 settings_data: Dict[str, Any],
1076 commit: bool = True,
1077 overwrite: bool = True,
1078 delete_extra: bool = False,
1079 ) -> None:
1080 """
1081 Import settings directly from the export format. This can be used to
1082 re-import settings that have been exported with `get_all_settings()`.
1084 Args:
1085 settings_data: The raw settings data to import.
1086 commit: Whether to commit the DB after loading the settings.
1087 overwrite: If true, it will overwrite the value of settings that
1088 are already in the database.
1089 delete_extra: If true, it will delete any settings that are in
1090 the database but don't have a corresponding entry in
1091 `settings_data`.
1093 """
1094 if self.db_session is None: 1094 ↛ 1095line 1094 didn't jump to line 1095 because the condition on line 1094 was never true
1095 raise RuntimeError("Database session is not initialized")
1096 logger.debug(f"Importing {len(settings_data)} settings")
1098 for key, setting_values in settings_data.items():
1099 setting_values = dict(setting_values)
1100 if not overwrite:
1101 existing_value = self.get_setting(key)
1102 if existing_value is not None:
1103 # Preserve the value from this setting.
1104 setting_values["value"] = existing_value
1106 # Delete any existing setting so we can completely overwrite it.
1107 self.delete_setting(key, commit=False)
1109 # Convert type string to SettingType enum if needed
1110 if "type" in setting_values and isinstance( 1110 ↛ 1115line 1110 didn't jump to line 1115 because the condition on line 1110 was always true
1111 setting_values["type"], str
1112 ):
1113 setting_values["type"] = SettingType[setting_values["type"]]
1115 setting = Setting(
1116 key=key, **_filter_setting_columns(setting_values)
1117 )
1118 self.db_session.add(setting)
1120 if commit or delete_extra: 1120 ↛ 1126line 1120 didn't jump to line 1126 because the condition on line 1120 was always true
1121 self.db_session.commit()
1122 logger.info(f"Successfully imported {len(settings_data)} settings")
1123 # Emit WebSocket event for all imported settings
1124 self._emit_settings_changed(list(settings_data.keys()))
1126 if delete_extra:
1127 all_settings = self.get_all_settings()
1128 for key in all_settings:
1129 if key not in settings_data:
1130 logger.debug(f"Deleting extraneous setting: {key}")
1131 self.delete_setting(key, commit=False)
1133 def _create_setting(self, key, value, setting_type):
1134 """Create a setting with appropriate metadata"""
1136 # Determine appropriate category
1137 category = None
1138 ui_element = "text"
1140 # Determine category based on key pattern
1141 if key.startswith("app."):
1142 category = "app_interface"
1143 elif key.startswith("llm."):
1144 if any(
1145 param in key
1146 for param in [
1147 "temperature",
1148 "max_tokens",
1149 "n_batch",
1150 "n_gpu_layers",
1151 ]
1152 ):
1153 category = "llm_parameters"
1154 else:
1155 category = "llm_general"
1156 elif key.startswith("search."):
1157 if any(
1158 param in key
1159 for param in ["iterations", "questions", "results", "region"]
1160 ):
1161 category = "search_parameters"
1162 else:
1163 category = "search_general"
1164 elif key.startswith("report."):
1165 category = "report_parameters"
1167 # Determine UI element type based on value
1168 ui_element = _infer_ui_element(value)
1170 # Build setting object
1171 setting_dict = {
1172 "key": key,
1173 "value": value,
1174 "type": setting_type.value.lower(),
1175 "name": key.split(".")[-1].replace("_", " ").title(),
1176 "description": f"Setting for {key}",
1177 "category": category,
1178 "ui_element": ui_element,
1179 }
1181 # Create the setting in the database
1182 self.create_or_update_setting(setting_dict, commit=False)
1184 def _emit_settings_changed(self, changed_keys: Optional[List[Any]] = None):
1185 """
1186 Emit WebSocket event when settings change
1188 Args:
1189 changed_keys: List of setting keys that changed
1190 """
1191 try:
1192 # Import here to avoid circular imports
1193 from ..web.services.socket_service import SocketIOService
1195 try:
1196 socket_service = SocketIOService()
1197 except ValueError:
1198 logger.debug(
1199 "Not emitting socket event because server is not initialized."
1200 )
1201 return
1203 # Get the changed settings
1204 settings_data = {}
1205 if changed_keys: 1205 ↛ 1212line 1205 didn't jump to line 1212 because the condition on line 1205 was always true
1206 for key in changed_keys:
1207 setting_value = self.get_setting(key)
1208 if setting_value is not None:
1209 settings_data[key] = {"value": setting_value}
1211 # Emit the settings change event
1212 from datetime import datetime, UTC
1214 socket_service.emit_socket_event(
1215 "settings_changed",
1216 {
1217 "changed_keys": changed_keys or [],
1218 "settings": settings_data,
1219 "timestamp": datetime.now(UTC).isoformat(),
1220 },
1221 )
1223 logger.debug(
1224 f"Emitted settings_changed event for keys: {changed_keys}"
1225 )
1227 except Exception:
1228 logger.exception("Failed to emit settings change event")
1229 # Don't let WebSocket emission failures break settings saving
1231 @staticmethod
1232 def get_bootstrap_env_vars() -> Dict[str, str]:
1233 """
1234 Get environment variables that must be available before database access.
1235 These are critical for system initialization.
1237 Returns:
1238 Dict mapping env var names to their descriptions
1239 """
1240 # Get bootstrap vars from env registry
1241 return env_registry.get_bootstrap_vars()
1243 @staticmethod
1244 def is_bootstrap_env_var(env_var: str) -> bool:
1245 """
1246 Check if an environment variable is a bootstrap variable (needed before DB access).
1248 Args:
1249 env_var: Environment variable name
1251 Returns:
1252 True if this is a bootstrap variable
1253 """
1254 bootstrap_vars = SettingsManager.get_bootstrap_env_vars()
1255 return env_var in bootstrap_vars
1257 @staticmethod
1258 def is_env_only_setting(key: str) -> bool:
1259 """
1260 Check if a setting key is environment-only.
1262 Args:
1263 key: Setting key to check
1265 Returns:
1266 True if it's an env-only setting, False otherwise
1267 """
1268 return env_registry.is_env_only(key)
1270 @staticmethod
1271 def get_env_var_for_setting(setting_key: str) -> str:
1272 """
1273 Get the environment variable name for a given setting key.
1275 Args:
1276 setting_key: Setting key (e.g., "app.host")
1278 Returns:
1279 Environment variable name (e.g., "LDR_APP_HOST")
1280 """
1281 # Use the same logic as check_env_setting for consistency
1282 return f"LDR_{'_'.join(setting_key.split('.')).upper()}"
1284 @staticmethod
1285 def get_setting_key_for_env_var(env_var: str) -> Optional[str]:
1286 """
1287 Get the setting key for a given environment variable.
1289 Args:
1290 env_var: Environment variable name (e.g., "LDR_APP_HOST")
1292 Returns:
1293 Setting key (e.g., "app.host") or None if not a valid LDR env var
1294 """
1295 if not env_var.startswith("LDR_"):
1296 return None
1298 # Remove LDR_ prefix and convert to lowercase
1299 without_prefix = env_var[4:]
1300 parts = without_prefix.split("_")
1302 return ".".join(part.lower() for part in parts)
1305class SnapshotSettingsContext:
1306 """Read-only settings context backed by a snapshot dict.
1308 Unwraps {"value": x} setting objects into plain values and provides
1309 get_setting(key, default) for thread-safe snapshot access.
1310 """
1312 def __init__(
1313 self, snapshot=None, username=None, missing_key_log_level="DEBUG"
1314 ):
1315 self.snapshot = snapshot or {}
1316 self.username = username
1317 self._missing_key_log_level = missing_key_log_level
1318 self.values = {}
1319 for key, setting in self.snapshot.items():
1320 if isinstance(setting, dict) and "value" in setting:
1321 self.values[key] = setting["value"]
1322 else:
1323 self.values[key] = setting
1325 def get_setting(self, key, default=None):
1326 """Return the setting value for *key*, or *default* if absent."""
1327 if key in self.values:
1328 return self.values[key]
1329 logger.log(
1330 self._missing_key_log_level,
1331 "Setting '{}' not found in snapshot, using default",
1332 key,
1333 )
1334 return default