Coverage for src/local_deep_research/settings/manager.py: 93%
642 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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# Select settings whose options lists are populated dynamically (at code or
159# runtime) rather than from the static defaults JSON. Their static options
160# are only a sample of what is legal, so options validation must be skipped
161# for them. Re-exported by `web/services/settings_service.py` (single
162# source of truth) and imported by tests that assert identity across
163# import orders.
164DYNAMIC_SETTINGS = ["llm.provider", "llm.model", "search.tool"]
167def _validate_imported_setting_value(
168 key: str, value: Any, default_meta: Dict[str, Any]
169) -> Optional[str]:
170 """Validate a file-supplied setting value against the current-defaults
171 schema for its key.
173 Mirrors the runtime checks `web.services.settings_service
174 .validate_setting` applies on the settings save path (type coercion via
175 ``get_typed_setting_value``, options / min / max constraints), but reads
176 constraints from the current defaults metadata instead of a stored
177 ``Setting`` row. Used by ``import_settings`` so a pre-upgrade export
178 cannot resurrect values that are invalid under the current schema
179 (#5589).
181 Returns ``None`` when the value is acceptable, or a human-readable
182 reason string when it is not.
183 """
184 ui_element = default_meta.get("ui_element")
185 if (
186 not isinstance(ui_element, str)
187 or ui_element not in UI_ELEMENT_TO_SETTING_TYPE
188 ):
189 # An unknown ui_element comes from the shipped defaults, not from the
190 # imported file: default_meta is self.default_settings. There is
191 # nothing to enforce for one, and probing get_typed_setting_value
192 # would log a default-substitution warning about a value this path
193 # stores verbatim. That value is then dead on read: the same helper
194 # substitutes the default for an element the type map lacks, so the
195 # imported value never comes back out.
196 return None
198 if value == default_meta.get("value"):
199 # The value equals the currently shipped default, so the defaults
200 # file itself vouches for it. Without this guard, a shipped default
201 # that is technically outside its own constraint metadata (e.g. a
202 # null optional numeric default) would be rejected and the key
203 # dropped on fresh-install seeding.
204 #
205 # ``app.theme`` used to be the example here: it shipped as ``dark``
206 # while the theme registry, which replaces that setting's options at
207 # runtime, had no ``dark`` entry. This guard is what let that invalid
208 # default seed successfully — and therefore what kept it invisible
209 # until the no-JS Save path validated it and reported the user's
210 # untouched theme as "failing". The default is now ``system``, which
211 # the registry does serve.
212 return None
214 typed_value = get_typed_setting_value(
215 key=key,
216 value=value,
217 ui_element=ui_element,
218 default=None,
219 check_env=False,
220 )
221 if typed_value is None:
222 # A shipped optional default may legitimately be null; the equality
223 # guard above accepts that exact value. Otherwise this is either an
224 # unconvertible input or a non-default null. The save path rejects it
225 # for typed controls, so imports must do the same rather than storing
226 # a value that the UI could not save.
227 if ui_element == "checkbox": 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true
228 return "Value must be a boolean"
229 if ui_element in ("number", "slider", "range"): 229 ↛ 231line 229 didn't jump to line 231 because the condition on line 229 was always true
230 return "Value must be a number"
231 return None
233 if ui_element == "checkbox":
234 if not isinstance(typed_value, bool): 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 return "Value must be a boolean"
236 elif ui_element in ("number", "slider", "range"):
237 if not isinstance(typed_value, (int, float)): 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 return "Value must be a number"
239 min_value = default_meta.get("min_value")
240 max_value = default_meta.get("max_value")
241 if min_value is not None and typed_value < min_value: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 return f"Value must be at least {min_value}"
243 if max_value is not None and typed_value > max_value:
244 return f"Value must be at most {max_value}"
245 elif ui_element == "select":
246 options = default_meta.get("options")
247 # Skip options validation for dynamically populated dropdowns —
248 # their static options are only a sample of legal values.
249 if options and key not in DYNAMIC_SETTINGS:
250 allowed_values = [
251 opt.get("value") if isinstance(opt, dict) else opt
252 for opt in list(options)
253 ]
254 if typed_value not in allowed_values:
255 return "Value must be one of: " + ", ".join(
256 str(v) for v in allowed_values
257 )
259 return None
262_POLICY_AUDIT_KEYS = frozenset(
263 {
264 "llm.require_local_endpoint",
265 "llm.allowed_local_hostnames",
266 "embeddings.require_local",
267 }
268)
271def _is_policy_setting(key: str) -> bool:
272 """Return True for security-relevant setting keys that need an
273 audit-log entry on change. Scope is intentionally narrow so this
274 audit hook doesn't widen into general settings-change logging.
275 """
276 if key.startswith("policy."):
277 return True
278 return key in _POLICY_AUDIT_KEYS
281def is_valid_setting_key(key: Any) -> bool:
282 """Return True if *key* is a well-formed setting key.
284 A valid key is a non-empty string of dot-separated segments where each
285 segment is non-empty and contains no whitespace. This rejects the
286 malformed keys that corrupt prefix/namespace lookups (see #4840) — a
287 trailing dot ``"foo."``, a leading dot ``".foo"``, an empty segment
288 ``"foo..bar"``, a blank key — as well as keys carrying whitespace
289 (``" foo"``, ``"foo. bar"``, ``"llm.o k"``), which would also break the
290 ``key.split(".")`` → ``LDR_...`` env-var mapping. A malformed row makes
291 ``get_setting("foo")`` return a ``{"": value}`` wrapper dict, which the
292 UI renders as ``[object Object]``.
294 Used on both sides: writes reject new malformed keys, and both read paths
295 (``SettingsManager.get_setting`` on the DB and ``get_setting_from_snapshot``
296 on a settings snapshot) filter malformed rows out of a prefix read so a
297 stray legacy row can never wrap a leaf value into an ``[object Object]``
298 dict.
299 """
300 if not isinstance(key, str) or not key:
301 return False
302 return all(
303 seg and not any(c.isspace() for c in seg) for seg in key.split(".")
304 )
307def _infer_ui_element(value: Any, current: str = "text") -> str:
308 """Infer the appropriate ui_element string from a Python value's type.
310 Args:
311 value: The value to infer the ui_element from.
312 current: The existing ui_element. If it is already something more
313 specific than ``"text"``, it is kept as-is.
314 """
315 if current != "text":
316 return current
317 if isinstance(value, bool):
318 return "checkbox"
319 if isinstance(value, (int, float)):
320 return "number"
321 if isinstance(value, (list, dict)):
322 return "json"
323 return "text"
326# Default categories for each typed setting prefix, used by the self-heal
327# block in set_setting() when a row's type column doesn't match its key
328# prefix. The values mirror the canonical strings in
329# web/routers/settings.py: a legacy chat.* row with type=APP and a
330# stale category gets repointed to type=CHAT + category="chat" on next
331# save. We use the most-general category per prefix here — sub-classifying
332# llm_general vs llm_parameters depends on the specific key, but the
333# self-heal only fires when the row was already mis-typed, so over-
334# generalizing the category is preferable to leaving it stale.
335_INFERRED_CATEGORY: Dict[str, str] = {
336 "llm.": "llm_general",
337 "search.": "search_general",
338 "report.": "report_parameters",
339 "database.": "database_parameters",
340 "chat.": "chat",
341}
344UI_ELEMENT_TO_SETTING_TYPE: Dict[str, Callable[..., Any]] = {
345 "text": str,
346 "json": _parse_json_value,
347 "password": str,
348 "select": str,
349 "number": _parse_number,
350 "range": _parse_number, # Same behavior as number for consistency
351 "checkbox": parse_boolean,
352 "textarea": str,
353 "multiselect": _parse_multiselect,
354}
357def get_typed_setting_value(
358 key: str,
359 value: Any,
360 ui_element: str,
361 default: Any = None,
362 check_env: bool = True,
363) -> Any:
364 """
365 Extracts the value for a particular setting, ensuring that it has the
366 correct type.
368 Args:
369 key: The setting key.
370 value: The setting value from the database.
371 ui_element: The setting UI element ID.
372 default: Default value to return if the value of the setting is
373 invalid.
374 check_env: If true, it will check the environment variable for
375 this setting before reading from the DB.
377 Returns:
378 The value of the setting.
380 """
381 setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None)
382 if setting_type is None:
383 logger.warning(
384 "Got unknown type {} for setting {}, returning default value.",
385 ui_element,
386 key,
387 )
388 return default
390 # Check environment variable first (highest priority).
391 if check_env:
392 env_value = check_env_setting(key)
393 if env_value is not None:
394 try:
395 return setting_type(env_value)
396 except ValueError:
397 logger.warning(
398 "Setting {} has invalid value {}. Falling back to DB.",
399 key,
400 env_value,
401 )
403 # If value is None (not in database), return default.
404 if value is None:
405 return default
407 # Read from the database.
408 try:
409 return setting_type(value)
410 except (ValueError, TypeError):
411 logger.warning(
412 "Setting {} has invalid value {}. Returning default.",
413 key,
414 value,
415 )
416 return default
419def check_env_setting(key: str) -> str | None:
420 """
421 Checks environment variables for a particular setting.
423 Args:
424 key: The database key for the setting.
426 Returns:
427 The setting from the environment variables, or None if the variable
428 is not set or is empty.
430 Note:
431 Empty environment variables ("") are treated as unset. This is standard
432 practice across the ecosystem — see CPython's official docs (PYTHON*
433 env vars require "a non-empty string"), botocore PR #1687, Pallets/Click
434 PR #2223, and Vercel Turborepo PR #6929. Orchestration tools like Unraid,
435 Terraform, and Kubernetes manifests often cannot conditionally omit env
436 var declarations, so they pass "" for unconfigured values. Treating ""
437 as unset prevents these empty strings from overriding database defaults.
438 See: https://github.com/LearningCircuit/local-deep-research/pull/3362
440 """
441 env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}"
442 env_value = os.getenv(env_variable_name)
443 # Treat empty string as unset — orchestration tools (Unraid, Terraform, K8s)
444 # often cannot omit env var declarations and pass "" for unconfigured values.
445 if env_value is not None and env_value != "":
446 logger.debug(f"Overriding {key} setting from environment variable.")
447 return env_value
448 if env_value == "":
449 logger.warning(
450 "Environment variable {} is set but empty — "
451 "ignoring it and falling back to DB/default for setting '{}'. "
452 "This is expected on Unraid or Docker templates that create "
453 "all variables even when left blank. To suppress this warning, "
454 "remove the variable from your environment or set a value.",
455 env_variable_name,
456 key,
457 )
458 return None
461def apply_environment_overrides_to_snapshot(
462 snapshot: Dict[str, Any],
463) -> Dict[str, Any]:
464 """Return a copied snapshot with current LDR_* overrides reapplied.
466 Queued research snapshots can outlive a process restart. Environment values
467 are operator policy and retain highest precedence at actual dispatch time.
468 """
469 if not isinstance(snapshot, dict): 469 ↛ 470line 469 didn't jump to line 470 because the condition on line 469 was never true
470 return snapshot
471 effective = dict(snapshot)
472 # Old or partial queued snapshots may omit keys added after enqueue. Use
473 # the registered defaults as metadata so every active LDR_* override is
474 # injected at dispatch, not only overrides for keys already serialized.
475 default_metadata = SettingsManager().default_settings
476 candidate_keys = set(snapshot) | set(default_metadata)
477 for key in candidate_keys:
478 if check_env_setting(str(key)) is None:
479 continue
480 stored = snapshot.get(key, default_metadata.get(key))
481 metadata = stored if isinstance(stored, dict) else None
482 current = metadata.get("value") if metadata is not None else stored
483 if metadata is not None:
484 ui_element = str(metadata.get("ui_element", "text"))
485 elif isinstance(current, bool): 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 ui_element = "checkbox"
487 elif isinstance(current, (list, dict)): 487 ↛ 489line 487 didn't jump to line 489 because the condition on line 487 was always true
488 ui_element = "json"
489 elif isinstance(current, (int, float)):
490 ui_element = "number"
491 else:
492 ui_element = "text"
493 typed = get_typed_setting_value(
494 str(key), current, ui_element, default=current, check_env=True
495 )
496 if metadata is not None:
497 updated = dict(metadata)
498 updated["value"] = typed
499 updated["editable"] = False
500 effective[key] = updated
501 else:
502 effective[key] = typed
503 return effective
506class SettingsManager(ISettingsManager):
507 """
508 Manager for handling application settings with database storage and file fallback.
509 Provides methods to get and set settings, with the ability to override settings in memory.
510 """
512 def __init__(
513 self,
514 db_session: Optional[Session] = None,
515 owns_session: bool = False,
516 ):
517 """
518 Initialize the settings manager
520 Args:
521 db_session: SQLAlchemy session for database operations
522 owns_session: If True, close() will close the session.
523 Defaults to False (safe for borrowed sessions). Set to True
524 only when this manager created/owns the session — currently
525 only get_settings_manager() in db_utils.py does this.
526 """
527 self.db_session = db_session
528 self._owns_session = owns_session
529 self._closed = False
530 self.db_first = True # Always prioritize DB settings
532 # Store the thread ID this instance was created in
533 self._creation_thread_id = threading.get_ident()
535 # Initialize settings lock as None - will be checked lazily
536 self.__settings_locked: Optional[bool] = None
538 # Auto-initialize settings if database is empty
539 if self.db_session:
540 self._ensure_settings_initialized()
542 def close(self):
543 """Close the DB session if this manager owns it.
545 Borrowed sessions (owns_session=False) are left open for their
546 owner to close (e.g. a caller that passed in its own db_session).
547 Safe to call multiple times — subsequent calls are no-ops.
548 """
549 if self._owns_session and self.db_session is not None:
550 try:
551 logger.debug("Closing owned DB session in SettingsManager")
552 self.db_session.close()
553 except Exception:
554 logger.warning(
555 "Failed to close SettingsManager DB session — "
556 "connection may leak",
557 )
558 self._closed = True
559 self.db_session = None
561 def _ensure_settings_initialized(self):
562 """Ensure settings are initialized in the database."""
563 # Check if we have any settings at all
564 from ..database.models import Setting
566 if self.db_session is None: 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 raise RuntimeError("Database session is not initialized")
568 settings_count = self.db_session.query(Setting).count()
570 if settings_count == 0:
571 logger.info("No settings found in database, loading defaults")
572 self.load_from_defaults_file(commit=True, override_locked=True)
573 logger.info("Default settings loaded successfully")
575 def _check_thread_safety(self):
576 """Check if this instance is being used in the same thread it was created in."""
577 current_thread_id = threading.get_ident()
578 if self.db_session and current_thread_id != self._creation_thread_id:
579 raise RuntimeError(
580 f"SettingsManager instance created in thread {self._creation_thread_id} "
581 f"is being used in thread {current_thread_id}. This is not thread-safe! "
582 f"Create a new SettingsManager instance within the current thread context."
583 )
585 @property
586 def settings_locked(self) -> bool:
587 """Check if settings are locked (lazy evaluation)."""
588 if self.__settings_locked is None:
589 try:
590 self.__settings_locked = self.get_setting(
591 "app.lock_settings", False
592 )
593 if self.settings_locked:
594 logger.info(
595 "Settings are locked. Disabling all settings changes."
596 )
597 except Exception:
598 logger.warning(
599 "Failed to check settings lock status, assuming not locked"
600 )
601 self.__settings_locked = False
602 return bool(self.__settings_locked)
604 def _is_environment_locked(self, key: str, operation: str) -> bool:
605 if check_env_setting(key) is None:
606 return False
607 logger.bind(policy_audit=True).warning(
608 "environment-locked setting rejected at {}", operation, key=key
609 )
610 return True
612 @functools.cached_property
613 def default_settings(self) -> Dict[str, Any]:
614 """
615 Returns:
616 The default settings, loaded from JSON files and merged.
617 Automatically discovers and loads all .json files in the defaults
618 directory and its subdirectories.
619 Theme options are dynamically injected from the theme registry.
621 """
622 settings: Dict[str, Any] = {}
624 try:
625 # Get the defaults package path
626 defaults_path = Path(defaults.__file__).parent
628 # Find all JSON files recursively in the defaults directory
629 json_files = sorted(defaults_path.rglob("*.json"))
631 logger.debug(f"Found {len(json_files)} JSON settings files")
633 # Load and merge all JSON files
634 for json_file in json_files:
635 try:
636 with open(json_file, "r", encoding="utf-8-sig") as f:
637 file_settings = json.load(f)
639 # Get relative path for logging
640 relative_path = json_file.relative_to(defaults_path)
642 # Warn about key conflicts
643 conflicts = set(settings.keys()) & set(file_settings.keys())
644 if conflicts:
645 logger.warning(
646 f"Keys {conflicts} from {relative_path} "
647 f"override existing values"
648 )
650 settings.update(file_settings)
651 logger.debug(f"Loaded {relative_path}")
653 except json.JSONDecodeError:
654 logger.exception(f"Invalid JSON in {json_file}")
655 except Exception:
656 logger.warning(f"Could not load {json_file}")
658 except Exception:
659 logger.warning("Error loading settings files")
661 # Inject dynamic theme options from theme registry
662 if "app.theme" in settings:
663 try:
664 from local_deep_research.web.themes import theme_registry
666 settings["app.theme"]["options"] = (
667 theme_registry.get_settings_options()
668 )
669 except ImportError:
670 # Theme registry not available, use static options from JSON
671 pass
673 # Inject search strategy options from code (single source of truth)
674 if "search.search_strategy" in settings:
675 from local_deep_research.constants import get_available_strategies
677 strategies = get_available_strategies()
678 settings["search.search_strategy"]["options"] = [
679 {"label": s["label"], "value": s["name"]} for s in strategies
680 ]
682 logger.debug(f"Loaded {len(settings)} total settings")
683 return settings
685 def __get_typed_setting_value(
686 self,
687 setting: Setting,
688 default: Any = None,
689 check_env: bool = True,
690 ) -> Any:
691 """
692 Extracts the value for a particular setting, ensuring that it has the
693 correct type.
695 Args:
696 setting: The setting to get the value for.
697 default: Default value to return if the value of the setting is
698 invalid.
699 check_env: If true, it will check the environment variable for
700 this setting before reading from the DB.
702 Returns:
703 The value of the setting.
705 """
706 return get_typed_setting_value(
707 str(setting.key),
708 setting.value,
709 str(setting.ui_element),
710 default=default,
711 check_env=check_env,
712 )
714 def __query_settings(self, key: str | None = None) -> List[Setting]:
715 """
716 Abstraction for querying settings that also transparently handles
717 reading the default settings file if the DB is not enabled.
719 Args:
720 key: The key to read. If None, it will read everything.
722 Returns:
723 The settings it queried.
725 """
726 if self.db_session:
727 self._check_thread_safety()
728 query = self.db_session.query(Setting)
729 if key is not None:
730 # This will find exact matches and any subkeys.
731 #
732 # The ``startswith`` arm is intentionally restricted to
733 # subkeys that have AT LEAST one character past the dot
734 # (``key LIKE 'foo.%' AND key != 'foo.'``). The bare
735 # ``LIKE 'foo.%'`` form silently also matches ``'foo.'``
736 # (because ``%`` matches zero or more chars), which would
737 # treat a malformed trailing-dot duplicate row as a
738 # subkey of ``foo`` and cause ``get_setting`` to return
739 # a 2-key dict (``{"": value, "foo": value}``) instead
740 # of the primitive. See embedding-settings page bug.
741 # escape_like: the requested key is user-supplied (e.g. via
742 # /settings/api/bulk?keys[]=...), so escape its LIKE wildcards
743 # or ``keys[]=%`` would match every dotted key and dump the
744 # whole table. The ``!= f"{key}."`` guard compares against the
745 # raw stored key, so it stays unescaped.
746 query = query.filter(
747 or_(
748 Setting.key == key,
749 and_(
750 Setting.key.like(
751 f"{escape_like(key)}.%", escape="\\"
752 ),
753 Setting.key != f"{key}.",
754 ),
755 )
756 )
757 return query.all()
759 logger.debug(
760 "DB is disabled, reading setting '{}' from defaults file.", key
761 )
763 settings = []
764 for candidate_key, setting in self.default_settings.items():
765 if key is None or (
766 candidate_key == key or candidate_key.startswith(f"{key}.")
767 ):
768 settings.append(
769 Setting(
770 key=candidate_key, # gitleaks:allow
771 **_filter_setting_columns(setting),
772 )
773 )
775 return settings
777 def get_setting(
778 self, key: str, default: Any = None, check_env: bool = True
779 ) -> Any:
780 """
781 Get a setting value
783 Args:
784 key: Setting key
785 default: Default value if setting is not found
786 check_env: If true, it will check the environment variable for
787 this setting before reading from the DB.
789 Returns:
790 Setting value or default if not found
791 """
792 if self._closed: 792 ↛ 793line 792 didn't jump to line 793 because the condition on line 792 was never true
793 logger.error(
794 "SettingsManager.get_setting('{}') called after close() — "
795 "this is a bug; the caller should not reuse a closed manager",
796 key,
797 )
798 raise RuntimeError(
799 "SettingsManager has been closed. "
800 "Create a new instance or call close() only at end of lifecycle."
801 )
803 # First check if this is an env-only setting
804 if env_registry.is_env_only(key):
805 return env_registry.get(key, default)
807 # If using database first approach and session available, check database
808 try:
809 settings = self.__query_settings(key)
810 # Drop malformed subkey rows (e.g. legacy "foo." / "foo.." keys)
811 # that slip past the SQL prefix filter. #4852 excludes only the
812 # exact single-trailing-dot row; a "foo.." / "foo. " row would
813 # otherwise turn a leaf read into a `{".": v}`-style wrapper dict
814 # rendered as `[object Object]` (#4840). The exact-match row is
815 # always kept so a direct read of a malformed key still works.
816 settings = [
817 s
818 for s in settings
819 if str(s.key) == key or is_valid_setting_key(str(s.key))
820 ]
821 if len(settings) == 1 and str(settings[0].key) == key:
822 # Only an exact row is a bottom-level key. A lone child row
823 # still represents a namespace and must return a mapping,
824 # matching get_setting_from_snapshot().
825 return self.__get_typed_setting_value(
826 settings[0], default, check_env
827 )
828 if settings:
829 # This is a higher-level key.
830 settings_map = {}
831 for setting in settings:
832 output_key = str(setting.key).removeprefix(f"{key}.")
833 settings_map[output_key] = self.__get_typed_setting_value(
834 setting, default, check_env
835 )
836 return settings_map
837 except SQLAlchemyError:
838 logger.exception(f"Error retrieving setting {key} from database")
840 # Check env var before returning default (setting not in DB)
841 if check_env:
842 env_value = check_env_setting(key)
843 if env_value is not None:
844 default_meta = self.default_settings.get(key)
845 if default_meta and isinstance(default_meta, dict):
846 ui_element = default_meta.get("ui_element", "text")
847 return get_typed_setting_value(
848 key,
849 None,
850 ui_element,
851 default=default,
852 check_env=True,
853 )
854 logger.warning(
855 "Setting '{}' has env var override but is not in "
856 "defaults — returning raw string without type "
857 "conversion. Add this setting to a defaults JSON "
858 "file with a ui_element type to enable proper "
859 "type conversion.",
860 key,
861 )
862 return env_value
864 # Return default if not found
865 return default
867 def get_bool_setting(
868 self, key: str, default: bool = False, check_env: bool = True
869 ) -> bool:
870 """
871 Get a setting value as a boolean, handling string conversion.
873 Args:
874 key: Setting key
875 default: Default boolean value if setting is not found
876 check_env: If true, it will check the environment variable for
877 this setting before reading from the DB.
879 Returns:
880 Boolean value of the setting
881 """
882 value = self.get_setting(key, default, check_env)
883 return to_bool(value, default)
885 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
886 """
887 Set a setting value
889 Args:
890 key: Setting key
891 value: Setting value
892 commit: Whether to commit the change
894 Returns:
895 True if successful, False otherwise
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.set_setting('{}') called after close() — "
900 "this is a bug; the caller should not reuse a closed manager",
901 key,
902 )
903 raise RuntimeError(
904 "SettingsManager has been closed. "
905 "Create a new instance or call close() only at end of lifecycle."
906 )
907 if not self.db_session:
908 logger.error(
909 "Cannot edit setting {} because no DB was provided.", key
910 )
911 return False
912 if self.settings_locked:
913 logger.error("Cannot edit setting {} because they are locked.", key)
914 return False
915 if self._is_environment_locked(key, "set_setting"):
916 return False
918 # Always update database if available
919 try:
920 self._check_thread_safety()
921 setting = (
922 self.db_session.query(Setting)
923 .filter(Setting.key == key)
924 .first()
925 )
926 # Capture old value for the policy-change audit log below.
927 old_value = setting.value if setting is not None else None
928 if setting:
929 if not setting.editable:
930 logger.error(
931 "Cannot change setting '{}' because it "
932 "is marked as non-editable.",
933 key,
934 )
935 return False
937 setting.value = value # type: ignore[assignment]
938 setting.updated_at = ( # type: ignore[assignment]
939 func.now()
940 ) # Explicitly set the current timestamp
942 # Self-heal stale ui_element from before inference was added
943 setting.ui_element = _infer_ui_element(
944 value, setting.ui_element
945 )
947 # Self-heal stale type from before the prefix dispatch was
948 # added (e.g. legacy chat.* rows created with type=APP).
949 # Also re-points category to the canonical per-prefix
950 # value, since a row with the wrong type column was
951 # almost certainly created before category dispatch was
952 # in place either.
953 inferred_type: Optional[SettingType] = None
954 inferred_category: Optional[str] = None
955 for prefix, category in _INFERRED_CATEGORY.items():
956 if key.startswith(prefix):
957 if prefix == "llm.":
958 inferred_type = SettingType.LLM
959 elif prefix == "search.":
960 inferred_type = SettingType.SEARCH
961 elif prefix == "report.":
962 inferred_type = SettingType.REPORT
963 elif prefix == "database.": 963 ↛ 964line 963 didn't jump to line 964 because the condition on line 963 was never true
964 inferred_type = SettingType.DATABASE
965 elif prefix == "chat.": 965 ↛ 967line 965 didn't jump to line 967 because the condition on line 965 was always true
966 inferred_type = SettingType.CHAT
967 inferred_category = category
968 break
969 # Only self-heal when the key matches a known prefix. Keys
970 # outside the dispatch map (e.g. focused_iteration.*,
971 # langgraph_agent.* which ship as type=SEARCH) must keep their
972 # shipped type — defaulting to APP here would wrongly demote
973 # them on every edit.
974 if inferred_type is not None and setting.type != inferred_type:
975 setting.type = inferred_type # type: ignore[assignment]
976 if inferred_category is not None: 976 ↛ 1016line 976 didn't jump to line 1016 because the condition on line 976 was always true
977 setting.category = inferred_category # type: ignore[assignment]
978 else:
979 # Refuse to CREATE a new row for a malformed key (e.g. a
980 # trailing-dot key like "foo."). Existing rows are still
981 # updatable above; this only blocks minting new corruption.
982 if not is_valid_setting_key(key):
983 logger.error(
984 "Refusing to create setting with malformed key {!r}",
985 key,
986 )
987 return False
989 # Determine setting type from key
990 setting_type = SettingType.APP
991 if key.startswith("llm."):
992 setting_type = SettingType.LLM
993 elif key.startswith("search."):
994 setting_type = SettingType.SEARCH
995 elif key.startswith("report."):
996 setting_type = SettingType.REPORT
997 elif key.startswith("database."): 997 ↛ 998line 997 didn't jump to line 998 because the condition on line 997 was never true
998 setting_type = SettingType.DATABASE
999 elif key.startswith("chat."): 999 ↛ 1000line 999 didn't jump to line 1000 because the condition on line 999 was never true
1000 setting_type = SettingType.CHAT
1002 # Infer ui_element from the value type
1003 ui_element = _infer_ui_element(value)
1005 # Create a new setting
1006 new_setting = Setting(
1007 key=key,
1008 value=value,
1009 type=setting_type,
1010 name=key.split(".")[-1].replace("_", " ").title(),
1011 ui_element=ui_element,
1012 description=f"Setting for {key}",
1013 )
1014 self.db_session.add(new_setting)
1016 if commit:
1017 self.db_session.commit()
1018 # Emit WebSocket event for settings change
1019 self._emit_settings_changed([key])
1021 # N16: audit log on policy.* / llm.require_local_endpoint /
1022 # embeddings.require_local changes. Targeted scope — only
1023 # security-relevant settings are logged, to avoid widening
1024 # this PR into a general audit-log refactor.
1025 if _is_policy_setting(key):
1026 logger.bind(policy_audit=True).warning(
1027 "policy setting changed | key={} old={} new={}",
1028 key,
1029 old_value,
1030 value,
1031 )
1033 return True
1034 except SQLAlchemyError:
1035 logger.exception(f"Error setting value for key: {key}")
1036 self.db_session.rollback()
1037 return False
1039 def clear_cache(self):
1040 """Clear the settings cache."""
1041 self.__dict__.pop("default_settings", None)
1042 logger.debug("Settings cache cleared")
1044 def get_all_settings(
1045 self,
1046 bypass_cache: bool = False,
1047 include_environment_overrides: bool = True,
1048 strict: bool = False,
1049 ) -> Dict[str, Any]:
1050 """
1051 Get all settings, merging defaults with database values.
1053 This ensures that new settings added to defaults.json automatically
1054 appear in the UI without requiring a database reset.
1056 Args:
1057 bypass_cache: If True, bypass the cache and read directly from database
1058 include_environment_overrides: If True, overlay current LDR_* values.
1059 strict: If True, propagate database and enum query failures.
1062 Returns:
1063 Dictionary of all settings
1064 """
1065 if self._closed: 1065 ↛ 1066line 1065 didn't jump to line 1066 because the condition on line 1065 was never true
1066 logger.error(
1067 "SettingsManager.get_all_settings() called after close() — "
1068 "this is a bug; the caller should not reuse a closed manager",
1069 )
1070 raise RuntimeError(
1071 "SettingsManager has been closed. "
1072 "Create a new instance or call close() only at end of lifecycle."
1073 )
1075 result = {}
1077 # Start with defaults so new settings are always included
1078 for key, default_setting in self.default_settings.items():
1079 result[key] = dict(default_setting)
1081 # Check env var override for defaults not yet in DB
1082 if include_environment_overrides:
1083 env_value = check_env_setting(key)
1084 if env_value is not None:
1085 ui_element = default_setting.get("ui_element", "text")
1086 typed_value = get_typed_setting_value(
1087 key,
1088 None,
1089 ui_element,
1090 default=env_value,
1091 check_env=True,
1092 )
1093 result[key]["value"] = typed_value
1094 result[key]["editable"] = False
1096 # Override with database settings
1097 try:
1098 db_settings = self.__query_settings()
1099 except (SQLAlchemyError, LookupError):
1100 # LookupError fires when a row's `type` column holds an enum
1101 # value that's no longer in `SettingType` (e.g. legacy 'CHAT'-
1102 # typed rows from removed features). The previous handler only
1103 # caught SQLAlchemyError, so a single stale row would crash the
1104 # whole snapshot — and every caller of get_all_settings (the
1105 # /settings/api endpoint, benchmark start, research start, MCP
1106 # entry points, …) downstream of it. Falling back to defaults-
1107 # only is strictly safer than crashing.
1108 logger.exception(
1109 "Error querying settings from database in get_all_settings"
1110 )
1111 if strict:
1112 raise
1113 db_settings = []
1115 for setting in db_settings:
1116 # Handle type field - it might be a string or an enum
1117 setting_type = setting.type
1118 if hasattr(setting_type, "name"):
1119 setting_type = setting_type.name
1121 # Log if this is a custom setting not in defaults
1122 if str(setting.key) not in result:
1123 logger.debug(
1124 f"Database contains custom setting not in "
1125 f"defaults: {setting.key} (type={setting_type}, "
1126 f"category={setting.category})"
1127 )
1129 # Override the default with the full database row — value AND
1130 # schema (options, description, name, constraints).
1131 #
1132 # This is deliberate, not a bug: the DB row is a self-contained
1133 # snapshot. Schema is NOT read live from JSON on every call; it is
1134 # reconciled from the JSON defaults only at version bump, via
1135 # `import_settings(overwrite=False)` (see `load_from_defaults_file`
1136 # and its callers in `database/initialize.py` and post-login in
1137 # `web/auth/routes.py`). Every release bumps the package version, so
1138 # JSON metadata changes reach a user on their next login after
1139 # upgrade. Do NOT change this to overlay defaults schema on every
1140 # read — it bypasses that version gate and breaks the snapshot
1141 # invariant (clean export/import round-trip, no mid-session drift).
1142 # See closed PR #2474 for the rejected schema/value-separation
1143 # approach and the reasoning.
1144 result[str(setting.key)] = {
1145 "value": setting.value,
1146 "type": setting_type,
1147 "name": setting.name,
1148 "description": setting.description,
1149 "category": setting.category,
1150 "ui_element": setting.ui_element,
1151 "options": setting.options,
1152 "min_value": setting.min_value,
1153 "max_value": setting.max_value,
1154 "step": setting.step,
1155 "visible": setting.visible,
1156 "editable": False if self.settings_locked else setting.editable,
1157 }
1159 # Override from the environment variables if needed.
1160 if include_environment_overrides:
1161 env_value = check_env_setting(str(setting.key))
1162 if env_value is not None:
1163 ui_element = result[str(setting.key)].get(
1164 "ui_element", setting.ui_element
1165 )
1166 typed_value = get_typed_setting_value(
1167 str(setting.key),
1168 None,
1169 ui_element,
1170 default=env_value,
1171 check_env=True,
1172 )
1173 result[str(setting.key)]["value"] = typed_value
1174 # Mark it as non-editable, because changes to the DB
1175 # value have no effect as long as the environment
1176 # variable is set.
1177 result[str(setting.key)]["editable"] = False
1179 # Re-inject search strategy options from code after DB merge,
1180 # since the DB stores options=null for this setting.
1181 if "search.search_strategy" in result:
1182 from local_deep_research.constants import get_available_strategies
1184 strategies = get_available_strategies()
1185 result["search.search_strategy"]["options"] = [
1186 {"label": s["label"], "value": s["name"]} for s in strategies
1187 ]
1189 return result
1191 def get_settings_snapshot(self, strict: bool = False) -> Dict[str, Any]:
1192 """
1193 Get a simplified settings snapshot with just key-value pairs.
1194 This is useful for passing settings to background threads or storing in metadata.
1196 Args:
1197 strict: If True, propagate database and enum query failures.
1199 Returns:
1200 Dictionary with setting keys mapped to their values
1201 """
1202 if self._closed: 1202 ↛ 1203line 1202 didn't jump to line 1203 because the condition on line 1202 was never true
1203 logger.error(
1204 "SettingsManager.get_settings_snapshot() called after close() — "
1205 "this is a bug; the caller should not reuse a closed manager",
1206 )
1207 raise RuntimeError(
1208 "SettingsManager has been closed. "
1209 "Create a new instance or call close() only at end of lifecycle."
1210 )
1212 all_settings = self.get_all_settings(strict=strict)
1213 settings_snapshot = {}
1215 for key, setting in all_settings.items():
1216 if isinstance(setting, dict) and "value" in setting: 1216 ↛ 1219line 1216 didn't jump to line 1219 because the condition on line 1216 was always true
1217 settings_snapshot[key] = setting["value"]
1218 else:
1219 settings_snapshot[key] = setting
1221 return settings_snapshot
1223 def create_or_update_setting(
1224 self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True
1225 ) -> Optional[Setting]:
1226 """
1227 Create or update a setting
1229 Args:
1230 setting: Setting object or dictionary
1231 commit: Whether to commit the change
1233 Returns:
1234 The created or updated Setting model, or None if failed
1235 """
1236 if not self.db_session:
1237 logger.warning(
1238 "No database session available, cannot create/update setting"
1239 )
1240 return None
1241 if self.settings_locked:
1242 logger.error("Cannot edit settings because they are locked.")
1243 return None
1245 # Convert dict to BaseSetting if needed
1246 if isinstance(setting, dict): 1246 ↛ 1273line 1246 didn't jump to line 1273 because the condition on line 1246 was always true
1247 # Determine type from key if not specified
1248 if "type" not in setting and "key" in setting:
1249 setting_obj: BaseSetting
1250 key = setting["key"]
1251 if key.startswith("llm."):
1252 setting_obj = LLMSetting(**setting)
1253 elif key.startswith("search."):
1254 setting_obj = SearchSetting(**setting)
1255 elif key.startswith("report."):
1256 setting_obj = ReportSetting(**setting)
1257 elif key.startswith("chat."): 1257 ↛ 1258line 1257 didn't jump to line 1258 because the condition on line 1257 was never true
1258 setting_obj = ChatSetting(**setting)
1259 elif key.startswith("app."):
1260 setting_obj = AppSetting(**setting)
1261 else:
1262 # Keys outside the four buckets (e.g. local_search_*,
1263 # embeddings.*, rag.*) live in their own namespaces.
1264 # Use BaseSetting so the key is written verbatim —
1265 # AppSetting's validator would otherwise prepend
1266 # `app.` and silently relocate the row away from
1267 # where every reader looks it up. See #4208.
1268 setting_obj = BaseSetting(type=SettingType.APP, **setting)
1269 else:
1270 # Use generic BaseSetting
1271 setting_obj = BaseSetting(**setting)
1272 else:
1273 setting_obj = setting
1275 if self._is_environment_locked(
1276 setting_obj.key, "create_or_update_setting"
1277 ):
1278 return None
1280 try:
1281 # Check if setting exists
1282 db_setting = (
1283 self.db_session.query(Setting)
1284 .filter(Setting.key == setting_obj.key)
1285 .first()
1286 )
1288 if db_setting:
1289 # Update existing setting
1290 if not db_setting.editable:
1291 logger.error(
1292 "Cannot change setting '{}' because it "
1293 "is marked as non-editable.",
1294 setting_obj.key,
1295 )
1296 return None
1298 db_setting.value = setting_obj.value # type: ignore[assignment]
1299 db_setting.name = setting_obj.name # type: ignore[assignment]
1300 db_setting.description = setting_obj.description # type: ignore[assignment]
1301 db_setting.category = setting_obj.category # type: ignore[assignment]
1302 db_setting.type = setting_obj.type # type: ignore[assignment]
1303 db_setting.ui_element = setting_obj.ui_element # type: ignore[assignment]
1304 db_setting.options = setting_obj.options # type: ignore[assignment]
1305 db_setting.min_value = setting_obj.min_value # type: ignore[assignment]
1306 db_setting.max_value = setting_obj.max_value # type: ignore[assignment]
1307 db_setting.step = setting_obj.step # type: ignore[assignment]
1308 db_setting.visible = setting_obj.visible # type: ignore[assignment]
1309 db_setting.editable = setting_obj.editable # type: ignore[assignment]
1310 db_setting.updated_at = ( # type: ignore[assignment]
1311 func.now()
1312 ) # Explicitly set the current timestamp
1313 else:
1314 # Refuse to CREATE a new row for a malformed key (see
1315 # is_valid_setting_key / #4840). Updates to existing rows
1316 # above are unaffected.
1317 if not is_valid_setting_key(setting_obj.key):
1318 logger.error(
1319 "Refusing to create setting with malformed key {!r}",
1320 setting_obj.key,
1321 )
1322 return None
1324 # Create new setting
1325 db_setting = Setting(
1326 key=setting_obj.key,
1327 value=setting_obj.value,
1328 type=setting_obj.type,
1329 name=setting_obj.name,
1330 description=setting_obj.description,
1331 category=setting_obj.category,
1332 ui_element=setting_obj.ui_element,
1333 options=setting_obj.options,
1334 min_value=setting_obj.min_value,
1335 max_value=setting_obj.max_value,
1336 step=setting_obj.step,
1337 visible=setting_obj.visible,
1338 editable=setting_obj.editable,
1339 )
1340 self.db_session.add(db_setting)
1342 if commit:
1343 self.db_session.commit()
1344 # Emit WebSocket event for settings change
1345 self._emit_settings_changed([setting_obj.key])
1347 return db_setting
1349 except SQLAlchemyError:
1350 logger.exception(
1351 f"Error creating/updating setting {setting_obj.key}"
1352 )
1353 self.db_session.rollback()
1354 return None
1356 def delete_setting(
1357 self, key: str, commit: bool = True, override_locked: bool = False
1358 ) -> bool:
1359 """
1360 Delete a setting
1362 Args:
1363 key: Setting key
1364 commit: Whether to commit the change
1365 override_locked: Delete even when settings are locked.
1367 Returns:
1368 True if successful, False otherwise
1369 """
1370 if not self.db_session:
1371 logger.warning(
1372 "No database session available, cannot delete setting"
1373 )
1374 return False
1376 if self._is_environment_locked(key, "delete_setting"):
1377 return False
1379 if not override_locked and self.settings_locked:
1380 logger.bind(policy_audit=True).warning(
1381 "locked setting rejected at delete_setting", key=key
1382 )
1383 return False
1385 try:
1386 # Remove from database
1387 result = (
1388 self.db_session.query(Setting)
1389 .filter(Setting.key == key)
1390 .delete()
1391 )
1393 if commit: 1393 ↛ 1396line 1393 didn't jump to line 1396 because the condition on line 1393 was always true
1394 self.db_session.commit()
1396 return result > 0
1397 except SQLAlchemyError:
1398 logger.exception("Error deleting setting")
1399 self.db_session.rollback()
1400 return False
1402 def load_from_defaults_file(
1403 self, commit: bool = True, **kwargs: Any
1404 ) -> None:
1405 """
1406 Import settings from the defaults settings file.
1408 Args:
1409 commit: Whether to commit changes to database. The post-login
1410 atomic block in `web/auth/routes.py` passes ``commit=False``
1411 and combines this call with ``update_db_version(commit=False)``
1412 under a single terminal ``db_session.commit()`` — preserving
1413 the all-or-nothing invariant is what prevents the sticky-loop
1414 bug where `app.version` is missing after a partial write.
1415 **kwargs: Will be passed to `import_settings`, including
1416 ``override_locked`` when the caller has to run while the
1417 settings lock is set.
1419 """
1420 start = time.perf_counter()
1421 row_count = len(self.default_settings)
1422 self.import_settings(self.default_settings, commit=commit, **kwargs)
1423 elapsed_ms = (time.perf_counter() - start) * 1000
1424 if elapsed_ms > 100:
1425 logger.info(
1426 f"load_from_defaults_file imported {row_count} settings "
1427 f"in {elapsed_ms:.0f}ms (commit={commit})"
1428 )
1429 else:
1430 logger.debug(
1431 f"load_from_defaults_file imported {row_count} settings "
1432 f"in {elapsed_ms:.0f}ms (commit={commit})"
1433 )
1435 def db_version_matches_package(self) -> bool:
1436 """
1437 Returns:
1438 True if the version saved in the DB matches the package version.
1440 Reads the stored ``app.version`` row directly (``check_env=False``)
1441 rather than going through the env-aware read path. The schema-version
1442 marker must reflect what is persisted so a stale DB triggers a
1443 migration even when ``LDR_APP_VERSION`` happens to match the running
1444 package; reading the env value would mask the staleness and skip
1445 the migration.
1446 """
1447 db_version = self.get_setting("app.version", check_env=False)
1448 logger.debug(
1449 f"App version saved in DB is {db_version}, have package "
1450 f"settings from version {package_version}."
1451 )
1453 return bool(db_version == package_version)
1455 def update_db_version(self, commit: bool = True) -> None:
1456 """
1457 Updates the version saved in the DB based on the package version.
1459 Args:
1460 commit: Whether to commit the version write to the database.
1461 Callers that want to combine this with other writes into
1462 a single atomic transaction should pass commit=False and
1463 commit the session themselves. The post-login block in
1464 `web/auth/routes.py` relies on this to bundle the defaults
1465 import and the `app.version` write into one SQLite
1466 transaction; splitting them risks the sticky-loop state
1467 where `app.version` never gets written.
1468 """
1469 logger.debug(f"Updating saved DB version to {package_version}.")
1471 # Internal migration plumbing: this row is the schema-version
1472 # marker, not a user-facing setting. The env lock applies to
1473 # ordinary set/delete mutations via the public API; using the
1474 # env-guarded delete_setting() here would let a stale
1475 # LDR_APP_VERSION suppress the row replacement and either leave
1476 # the unique constraint violated (next commit raises) or leave
1477 # the stale marker in place. The narrowest correct path is to
1478 # delete the prior row directly and abort the call on failure
1479 # rather than blindly inserting a duplicate.
1480 if self.db_session is None: 1480 ↛ 1481line 1480 didn't jump to line 1481 because the condition on line 1480 was never true
1481 raise RuntimeError("Database session is not initialized")
1482 try:
1483 self._check_thread_safety()
1484 self.db_session.query(Setting).filter(
1485 Setting.key == "app.version"
1486 ).delete(synchronize_session=False)
1487 except SQLAlchemyError:
1488 logger.exception("Error deleting prior app.version row")
1489 self.db_session.rollback()
1490 return
1492 version = Setting(
1493 key="app.version",
1494 value=package_version,
1495 description="Version of the app this database is associated with.",
1496 editable=False,
1497 name="App Version",
1498 type=SettingType.APP,
1499 ui_element="text",
1500 visible=False,
1501 )
1502 self.db_session.add(version)
1503 if commit:
1504 self.db_session.commit()
1506 def import_settings(
1507 self,
1508 settings_data: Dict[str, Any],
1509 commit: bool = True,
1510 overwrite: bool = True,
1511 delete_extra: bool = False,
1512 preserve_environment_locked: bool = False,
1513 override_locked: bool = False,
1514 ) -> None:
1515 """
1516 Import settings directly from the export format. This can be used to
1517 re-import settings that have been exported with `get_all_settings()`.
1519 Args:
1520 settings_data: The raw settings data to import.
1521 commit: Whether to commit the DB after loading the settings.
1522 overwrite: If true, it will overwrite the value of settings that
1523 are already in the database.
1524 delete_extra: If true, it will delete any settings that are in
1525 the database but don't have a corresponding entry in
1526 `settings_data`.
1527 preserve_environment_locked: If true, preserve stored values for
1528 settings with active environment overrides while imported
1529 metadata is refreshed.
1530 override_locked: Import even when settings are locked. The
1531 bootstrap callers that add newly shipped defaults with
1532 `overwrite=False` pass this, since a locked account still
1533 has to receive settings introduced by an upgrade.
1535 """
1536 if not override_locked and self.settings_locked:
1537 logger.bind(policy_audit=True).warning(
1538 "locked settings rejected at import_settings"
1539 )
1540 return
1542 if self.db_session is None: 1542 ↛ 1543line 1542 didn't jump to line 1543 because the condition on line 1542 was never true
1543 raise RuntimeError("Database session is not initialized")
1544 logger.debug(f"Importing {len(settings_data)} settings")
1545 changed_keys: list[str] = []
1546 retained_keys: set[str] = set()
1547 # Schema-aware import (#5589): validate values that come from the
1548 # imported file against the CURRENT defaults schema, so a
1549 # pre-upgrade export cannot resurrect values that are invalid under
1550 # the current options/constraints. Values retained from the database
1551 # (the `overwrite=False` version-bump reconciliation path, and
1552 # environment-locked values under `preserve_environment_locked`)
1553 # are deliberately NOT validated: they are trusted stored state,
1554 # not untrusted file input, and rejecting them would break the
1555 # reconciliation contract ("refresh schema, keep value").
1556 defaults_for_import = self.default_settings
1557 try:
1558 # `overwrite=False` is the version-bump reconciliation point:
1559 # refresh JSON-defined schema while preserving the stored value.
1560 # Environment overrides must not become persisted values.
1561 for key, raw_setting_values in settings_data.items():
1562 if not is_valid_setting_key(key):
1563 logger.warning(
1564 "Skipping import of setting with malformed key {!r}",
1565 key,
1566 )
1567 continue
1569 retained_keys.add(key)
1570 setting_values = dict(raw_setting_values)
1571 value_replaced_from_db = False
1572 if (
1573 preserve_environment_locked
1574 and check_env_setting(key) is not None
1575 ):
1576 existing_value = (
1577 self.db_session.query(Setting.value)
1578 .filter(Setting.key == key)
1579 .first()
1580 )
1581 if existing_value is not None:
1582 logger.bind(policy_audit=True).warning(
1583 "Preserving stored value for environment-locked "
1584 "setting during import: {}",
1585 key,
1586 )
1587 setting_values["value"] = existing_value[0]
1588 value_replaced_from_db = True
1589 if not overwrite:
1590 existing_value = self.get_setting(key, check_env=False)
1591 if existing_value is not None:
1592 setting_values["value"] = existing_value
1593 value_replaced_from_db = True
1595 if not value_replaced_from_db:
1596 # Schema-aware import (#5589): a value that comes from
1597 # the imported file is untrusted input. Validate it
1598 # against the CURRENT defaults schema so a pre-upgrade
1599 # export cannot resurrect values that are invalid under
1600 # the current options/constraints. Values retained from
1601 # the database (the `overwrite=False` version-bump
1602 # reconciliation path, and environment-locked values
1603 # under `preserve_environment_locked`) are deliberately
1604 # NOT validated: they are trusted stored state, and
1605 # rejecting them would break the reconciliation
1606 # contract ("refresh schema, keep value").
1607 default_meta = defaults_for_import.get(key)
1608 if default_meta is not None:
1609 invalid_reason = _validate_imported_setting_value(
1610 key,
1611 setting_values.get("value"),
1612 default_meta,
1613 )
1614 if invalid_reason is not None:
1615 logger.warning(
1616 "Skipping import of setting {!r}: value "
1617 "{!r} is invalid under the current defaults "
1618 "schema — {}",
1619 key,
1620 setting_values.get("value"),
1621 invalid_reason,
1622 )
1623 continue
1625 # Import needs strict failure semantics. The public
1626 # delete_setting() helper intentionally swallows SQL errors,
1627 # which would allow this transaction to continue after a
1628 # rollback and commit only a suffix of the import.
1629 (
1630 self.db_session.query(Setting)
1631 .filter(Setting.key == key)
1632 .delete(synchronize_session=False)
1633 )
1635 if "type" in setting_values and isinstance( 1635 ↛ 1640line 1635 didn't jump to line 1640 because the condition on line 1635 was always true
1636 setting_values["type"], str
1637 ):
1638 setting_values["type"] = SettingType[setting_values["type"]]
1640 self.db_session.add(
1641 Setting(key=key, **_filter_setting_columns(setting_values))
1642 )
1643 changed_keys.append(key)
1645 if delete_extra:
1646 existing_keys = [
1647 str(row[0])
1648 for row in self.db_session.query(Setting.key).all()
1649 ]
1650 for key in existing_keys:
1651 if key in retained_keys or (
1652 preserve_environment_locked
1653 and check_env_setting(key) is not None
1654 ):
1655 continue
1656 logger.debug(f"Deleting extraneous setting: {key}")
1657 (
1658 self.db_session.query(Setting)
1659 .filter(Setting.key == key)
1660 .delete(synchronize_session=False)
1661 )
1662 changed_keys.append(key)
1664 if commit:
1665 self.db_session.commit()
1666 logger.info(
1667 f"Successfully imported {len(changed_keys)} settings"
1668 )
1669 self._emit_settings_changed(changed_keys)
1670 except Exception:
1671 self.db_session.rollback()
1672 raise
1674 def emit_settings_changed_after_commit(
1675 self, changed_keys: List[str]
1676 ) -> None:
1677 """Notify the current user's clients after the caller commits settings."""
1678 self._emit_settings_changed(changed_keys)
1680 def _emit_settings_changed(self, changed_keys: Optional[List[Any]] = None):
1681 """
1682 Emit WebSocket event when settings change
1684 Args:
1685 changed_keys: List of setting keys that changed
1686 """
1687 try:
1688 # Import here to avoid circular imports
1689 from ..web.services.socketio_asgi import emit_to_user
1691 # settings_changed carries raw setting values (including plaintext
1692 # API keys), so it must reach only the owning user's own browser
1693 # tabs — never every connected client on this shared, single-user-
1694 # agnostic Socket.IO server (#4437). A change made outside a
1695 # request context (app start-up defaults, background workers,
1696 # migrations) has no user tab to notify, so we skip the emit
1697 # entirely rather than fall back to a broadcast.
1698 from datetime import datetime, UTC
1700 # Resolve the user from the contextvar set by DatabaseMiddleware.
1701 # SettingsManager itself does not hold a username attribute.
1702 from ..utilities.request_context import get_current_username
1704 username = get_current_username()
1705 if not username:
1706 logger.debug(
1707 "Skipping settings_changed emit: no request username context"
1708 )
1709 return
1711 # Read the changed values only after the recipient check: each
1712 # read types the stored row, so an unknown ui_element logs a
1713 # default-substitution warning for an emit that never happens.
1714 settings_data = {}
1715 if changed_keys: 1715 ↛ 1721line 1715 didn't jump to line 1721 because the condition on line 1715 was always true
1716 for key in changed_keys:
1717 setting_value = self.get_setting(key)
1718 if setting_value is not None:
1719 settings_data[key] = {"value": setting_value}
1721 emit_to_user(
1722 "settings_changed",
1723 username,
1724 {
1725 "changed_keys": changed_keys or [],
1726 "settings": settings_data,
1727 "timestamp": datetime.now(UTC).isoformat(),
1728 },
1729 )
1731 logger.debug(
1732 f"Emitted settings_changed event for keys: {changed_keys}"
1733 )
1735 except Exception:
1736 logger.exception("Failed to emit settings change event")
1737 # Don't let WebSocket emission failures break settings saving
1739 @staticmethod
1740 def get_bootstrap_env_vars() -> Dict[str, str]:
1741 """
1742 Get environment variables that must be available before database access.
1743 These are critical for system initialization.
1745 Returns:
1746 Dict mapping env var names to their descriptions
1747 """
1748 # Get bootstrap vars from env registry
1749 return env_registry.get_bootstrap_vars()
1751 @staticmethod
1752 def is_bootstrap_env_var(env_var: str) -> bool:
1753 """
1754 Check if an environment variable is a bootstrap variable (needed before DB access).
1756 Args:
1757 env_var: Environment variable name
1759 Returns:
1760 True if this is a bootstrap variable
1761 """
1762 bootstrap_vars = SettingsManager.get_bootstrap_env_vars()
1763 return env_var in bootstrap_vars
1765 @staticmethod
1766 def is_env_only_setting(key: str) -> bool:
1767 """
1768 Check if a setting key is environment-only.
1770 Args:
1771 key: Setting key to check
1773 Returns:
1774 True if it's an env-only setting, False otherwise
1775 """
1776 return env_registry.is_env_only(key)
1778 @staticmethod
1779 def get_env_var_for_setting(setting_key: str) -> str:
1780 """
1781 Get the environment variable name for a given setting key.
1783 Args:
1784 setting_key: Setting key (e.g., "app.host")
1786 Returns:
1787 Environment variable name (e.g., "LDR_APP_HOST")
1788 """
1789 # Use the same logic as check_env_setting for consistency
1790 return f"LDR_{'_'.join(setting_key.split('.')).upper()}"
1792 @staticmethod
1793 def get_setting_key_for_env_var(env_var: str) -> Optional[str]:
1794 """
1795 Get the setting key for a given environment variable.
1797 Args:
1798 env_var: Environment variable name (e.g., "LDR_APP_HOST")
1800 Returns:
1801 Setting key (e.g., "app.host") or None if not a valid LDR env var
1802 """
1803 if not env_var.startswith("LDR_"):
1804 return None
1806 # Remove LDR_ prefix and convert to lowercase
1807 without_prefix = env_var[4:]
1808 parts = without_prefix.split("_")
1810 return ".".join(part.lower() for part in parts)
1813class SnapshotSettingsContext:
1814 """Read-only settings context backed by a snapshot dict.
1816 Unwraps {"value": x} setting objects into plain values and provides
1817 get_setting(key, default) for thread-safe snapshot access.
1818 """
1820 def __init__(
1821 self, snapshot=None, username=None, missing_key_log_level="DEBUG"
1822 ):
1823 self.snapshot = snapshot or {}
1824 self.username = username
1825 self._missing_key_log_level = missing_key_log_level
1826 self.values = {}
1827 for key, setting in self.snapshot.items():
1828 if isinstance(setting, dict) and "value" in setting:
1829 self.values[key] = setting["value"]
1830 else:
1831 self.values[key] = setting
1833 def get_setting(self, key, default=None):
1834 """Return the setting value for *key*, or *default* if absent."""
1835 if key in self.values:
1836 return self.values[key]
1837 logger.log(
1838 self._missing_key_log_level,
1839 "Setting '{}' not found in snapshot, using default",
1840 key,
1841 )
1842 return default