Coverage for src / local_deep_research / web / services / settings_manager.py: 58%
289 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1import json
2import os
3import threading
4from pathlib import Path
5from typing import Any, Dict, List, Optional, Type, Union
7from loguru import logger
8from sqlalchemy import func, or_
9from sqlalchemy.exc import SQLAlchemyError
10from sqlalchemy.orm import Session
12from ... import defaults
13from ...__version__ import __version__ as package_version
14from ...database.models import Setting, SettingType
15from ..models.settings import (
16 AppSetting,
17 BaseSetting,
18 LLMSetting,
19 ReportSetting,
20 SearchSetting,
21)
24def check_env_setting(key: str) -> str | None:
25 """
26 Checks environment variables for a particular setting.
28 Args:
29 key: The database key for the setting.
31 Returns:
32 The setting from the environment variables, or None if the variable
33 is not set.
35 """
36 env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}"
37 env_value = os.getenv(env_variable_name)
38 if env_value is not None:
39 logger.debug(f"Overriding {key} setting from environment variable.")
40 return env_value
43class SettingsManager:
44 """
45 Manager for handling application settings with database storage and file fallback.
46 Provides methods to get and set settings, with the ability to override settings in memory.
47 """
49 _UI_ELEMENT_TO_SETTING_TYPE = {
50 "text": str,
51 # SQLAlchemy should already handle JSON parsing.
52 "json": lambda x: x,
53 "password": str,
54 "select": str,
55 "number": float,
56 "range": float,
57 "checkbox": bool,
58 }
60 def __init__(self, db_session: Optional[Session] = None):
61 """
62 Initialize the settings manager
64 Args:
65 db_session: SQLAlchemy session for database operations
66 """
67 self.db_session = db_session
68 self.db_first = True # Always prioritize DB settings
70 # Store the thread ID this instance was created in
71 self._creation_thread_id = threading.get_ident()
73 # Initialize settings lock as None - will be checked lazily
74 self.__settings_locked = None
76 def _check_thread_safety(self):
77 """Check if this instance is being used in the same thread it was created in."""
78 current_thread_id = threading.get_ident()
79 if self.db_session and current_thread_id != self._creation_thread_id: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 raise RuntimeError(
81 f"SettingsManager instance created in thread {self._creation_thread_id} "
82 f"is being used in thread {current_thread_id}. This is not thread-safe! "
83 f"Create a new SettingsManager instance within the current thread context."
84 )
86 @property
87 def settings_locked(self) -> bool:
88 """Check if settings are locked (lazy evaluation)."""
89 if self.__settings_locked is None:
90 try:
91 self.__settings_locked = self.get_setting(
92 "app.lock_settings", False
93 )
94 if self.settings_locked: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true
95 logger.info(
96 "Settings are locked. Disabling all settings changes."
97 )
98 except Exception:
99 # If we can't check, assume not locked
100 self.__settings_locked = False
101 return self.__settings_locked
103 @property
104 def default_settings(self) -> Dict[str, Any]:
105 """
106 Returns:
107 The default settings, loaded from JSON files and merged.
108 Automatically discovers and loads all .json files in the defaults directory
109 and its subdirectories.
111 """
112 settings = {}
114 try:
115 # Get the defaults package path
116 defaults_path = Path(defaults.__file__).parent
118 # Find all JSON files recursively in the defaults directory
119 json_files = list(defaults_path.rglob("*.json"))
121 # Sort for consistent loading order
122 json_files.sort()
124 logger.info(f"Found {len(json_files)} JSON settings files")
126 # Load and merge all JSON files
127 for json_file in json_files:
128 try:
129 with open(json_file, "r") as f:
130 file_settings = json.load(f)
132 # Get relative path for logging
133 relative_path = json_file.relative_to(defaults_path)
135 # Warn about key conflicts
136 conflicts = set(settings.keys()) & set(file_settings.keys())
137 if conflicts:
138 logger.warning(
139 f"Keys {conflicts} from {relative_path} override existing values"
140 )
142 settings.update(file_settings)
143 logger.debug(f"Loaded {relative_path}")
145 except json.JSONDecodeError:
146 logger.exception(f"Invalid JSON in {json_file}")
147 except Exception as e:
148 logger.warning(f"Could not load {json_file}: {e}")
150 except Exception as e:
151 logger.warning(f"Error loading settings files: {e}")
153 logger.info(f"Loaded {len(settings)} total settings")
154 return settings
156 def __get_typed_setting_value(
157 self,
158 setting: Type[Setting],
159 default: Any = None,
160 check_env: bool = True,
161 ) -> Any:
162 """
163 Extracts the value for a particular setting, ensuring that it has the
164 correct type.
166 Args:
167 setting: The setting to get the value for.
168 default: Default value to return if the value of the setting is
169 invalid.
170 check_env: If true, it will check the environment variable for
171 this setting before reading from the DB.
173 Returns:
174 The value of the setting.
176 """
177 setting_type = self._UI_ELEMENT_TO_SETTING_TYPE.get(
178 setting.ui_element, None
179 )
180 if setting_type is None:
181 logger.warning(
182 "Got unknown type {} for setting {}, returning default value.",
183 setting.ui_element,
184 setting.key,
185 )
186 return default
188 # Check environment variable first, then database.
189 if check_env: 189 ↛ 202line 189 didn't jump to line 202 because the condition on line 189 was always true
190 env_value = check_env_setting(setting.key)
191 if env_value is not None:
192 try:
193 return setting_type(env_value)
194 except ValueError:
195 logger.warning(
196 "Setting {} has invalid value {}. Falling back to DB.",
197 setting.key,
198 env_value,
199 )
201 # If environment variable does not exist, read from the database.
202 try:
203 return setting_type(setting.value)
204 except ValueError:
205 logger.warning(
206 "Setting {} has invalid value {}. Returning default.",
207 setting.key,
208 setting.value,
209 )
210 return default
212 def __query_settings(self, key: str | None = None) -> List[Type[Setting]]:
213 """
214 Abstraction for querying settings that also transparently handles
215 reading the default settings file if the DB is not enabled.
217 Args:
218 key: The key to read. If None, it will read everything.
220 Returns:
221 The settings it queried.
223 """
224 if self.db_session: 224 ↛ 238line 224 didn't jump to line 238 because the condition on line 224 was always true
225 self._check_thread_safety()
226 query = self.db_session.query(Setting)
227 if key is not None:
228 # This will find exact matches and any subkeys.
229 query = query.filter(
230 or_(
231 Setting.key == key,
232 Setting.key.startswith(f"{key}."),
233 )
234 )
235 return query.all()
237 else:
238 logger.debug(
239 "DB is disabled, reading setting '{}' from defaults file.", key
240 )
242 settings = []
243 for candidate_key, setting in self.default_settings.items():
244 if key is None or (
245 candidate_key == key or candidate_key.startswith(f"{key}.")
246 ):
247 settings.append(Setting(key=candidate_key, **setting))
249 return settings
251 def get_setting(
252 self, key: str, default: Any = None, check_env: bool = True
253 ) -> Any:
254 """
255 Get a setting value
257 Args:
258 key: Setting key
259 default: Default value if setting is not found
260 check_env: If true, it will check the environment variable for
261 this setting before reading from the DB.
263 Returns:
264 Setting value or default if not found
265 """
266 # If using database first approach and session available, check database
267 try:
268 settings = self.__query_settings(key)
269 if len(settings) == 1:
270 # This is a bottom-level key.
271 return self.__get_typed_setting_value(
272 settings[0], default, check_env
273 )
274 elif len(settings) > 1:
275 # This is a higher-level key.
276 settings_map = {}
277 for setting in settings:
278 output_key = setting.key.removeprefix(f"{key}.")
279 settings_map[output_key] = self.__get_typed_setting_value(
280 setting, default, check_env
281 )
282 return settings_map
283 except SQLAlchemyError as e:
284 logger.exception(
285 f"Error retrieving setting {key} from database: {e}"
286 )
288 # Return default if not found
289 return default
291 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
292 """
293 Set a setting value
295 Args:
296 key: Setting key
297 value: Setting value
298 commit: Whether to commit the change
300 Returns:
301 True if successful, False otherwise
302 """
303 if not self.db_session: 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 logger.error(
305 "Cannot edit setting {} because no DB was provided.", key
306 )
307 return False
308 if self.settings_locked: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 logger.error("Cannot edit setting {} because they are locked.", key)
310 return False
312 # Always update database if available
313 try:
314 self._check_thread_safety()
315 setting = (
316 self.db_session.query(Setting)
317 .filter(Setting.key == key)
318 .first()
319 )
320 if setting:
321 if not setting.editable: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 logger.error(
323 "Cannot change setting '{}' because it "
324 "is marked as non-editable.",
325 key,
326 )
327 return False
329 setting.value = value
330 setting.updated_at = (
331 func.now()
332 ) # Explicitly set the current timestamp
333 else:
334 # Determine setting type from key
335 setting_type = SettingType.APP
336 if key.startswith("llm."): 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 setting_type = SettingType.LLM
338 elif key.startswith("search."): 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 setting_type = SettingType.SEARCH
340 elif key.startswith("report."): 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 setting_type = SettingType.REPORT
343 # Create a new setting
344 new_setting = Setting(
345 key=key,
346 value=value,
347 type=setting_type,
348 name=key.split(".")[-1].replace("_", " ").title(),
349 ui_element="text",
350 description=f"Setting for {key}",
351 )
352 self.db_session.add(new_setting)
354 if commit: 354 ↛ 359line 354 didn't jump to line 359 because the condition on line 354 was always true
355 self.db_session.commit()
356 # Emit WebSocket event for settings change
357 self._emit_settings_changed([key])
359 return True
360 except SQLAlchemyError:
361 logger.exception("Error setting value")
362 self.db_session.rollback()
363 return False
365 def get_all_settings(self) -> Dict[str, Any]:
366 """
367 Get all settings
369 Returns:
370 Dictionary of all settings
371 """
372 result = {}
374 # Add database settings if available
375 try:
376 for setting in self.__query_settings():
377 result[setting.key] = dict(
378 value=setting.value,
379 type=setting.type.name,
380 name=setting.name,
381 description=setting.description,
382 category=setting.category,
383 ui_element=setting.ui_element,
384 options=setting.options,
385 min_value=setting.min_value,
386 max_value=setting.max_value,
387 step=setting.step,
388 visible=setting.visible,
389 editable=False
390 if self.settings_locked
391 else setting.editable,
392 )
394 # Override from the environment variables if needed.
395 env_value = check_env_setting(setting.key)
396 if env_value is not None: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 result[setting.key]["value"] = env_value
398 # Mark it as non-editable, because changes to the DB
399 # value have no effect as long as the environment
400 # variable is set.
401 result[setting.key]["editable"] = False
402 except SQLAlchemyError as e:
403 logger.exception(
404 f"Error retrieving all settings from database: {e}"
405 )
407 return result
409 def create_or_update_setting(
410 self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True
411 ) -> Optional[Setting]:
412 """
413 Create or update a setting
415 Args:
416 setting: Setting object or dictionary
417 commit: Whether to commit the change
419 Returns:
420 The created or updated Setting model, or None if failed
421 """
422 if not self.db_session:
423 logger.warning(
424 "No database session available, cannot create/update setting"
425 )
426 return None
427 if self.settings_locked:
428 logger.error("Cannot edit settings because they are locked.")
429 return None
431 # Convert dict to BaseSetting if needed
432 if isinstance(setting, dict):
433 # Determine type from key if not specified
434 if "type" not in setting and "key" in setting:
435 key = setting["key"]
436 if key.startswith("llm."):
437 setting_obj = LLMSetting(**setting)
438 elif key.startswith("search."):
439 setting_obj = SearchSetting(**setting)
440 elif key.startswith("report."):
441 setting_obj = ReportSetting(**setting)
442 else:
443 setting_obj = AppSetting(**setting)
444 else:
445 # Use generic BaseSetting
446 setting_obj = BaseSetting(**setting)
447 else:
448 setting_obj = setting
450 try:
451 # Check if setting exists
452 db_setting = (
453 self.db_session.query(Setting)
454 .filter(Setting.key == setting_obj.key)
455 .first()
456 )
458 if db_setting:
459 # Update existing setting
460 if not db_setting.editable:
461 logger.error(
462 "Cannot change setting '{}' because it "
463 "is marked as non-editable.",
464 setting_obj.key,
465 )
466 return None
468 db_setting.value = setting_obj.value
469 db_setting.name = setting_obj.name
470 db_setting.description = setting_obj.description
471 db_setting.category = setting_obj.category
472 db_setting.ui_element = setting_obj.ui_element
473 db_setting.options = setting_obj.options
474 db_setting.min_value = setting_obj.min_value
475 db_setting.max_value = setting_obj.max_value
476 db_setting.step = setting_obj.step
477 db_setting.visible = setting_obj.visible
478 db_setting.editable = setting_obj.editable
479 db_setting.updated_at = (
480 func.now()
481 ) # Explicitly set the current timestamp
482 else:
483 # Create new setting
484 db_setting = Setting(
485 key=setting_obj.key,
486 value=setting_obj.value,
487 type=setting_obj.type, # It's already a SettingType enum
488 name=setting_obj.name,
489 description=setting_obj.description,
490 category=setting_obj.category,
491 ui_element=setting_obj.ui_element,
492 options=setting_obj.options,
493 min_value=setting_obj.min_value,
494 max_value=setting_obj.max_value,
495 step=setting_obj.step,
496 visible=setting_obj.visible,
497 editable=setting_obj.editable,
498 )
499 self.db_session.add(db_setting)
501 if commit:
502 self.db_session.commit()
503 # Emit WebSocket event for settings change
504 self._emit_settings_changed([setting_obj.key])
506 return db_setting
508 except SQLAlchemyError as e:
509 logger.exception(
510 f"Error creating/updating setting {setting_obj.key}: {e}"
511 )
512 self.db_session.rollback()
513 return None
515 def delete_setting(self, key: str, commit: bool = True) -> bool:
516 """
517 Delete a setting
519 Args:
520 key: Setting key
521 commit: Whether to commit the change
523 Returns:
524 True if successful, False otherwise
525 """
526 if not self.db_session: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 logger.warning(
528 "No database session available, cannot delete setting"
529 )
530 return False
532 try:
533 # Remove from database
534 result = (
535 self.db_session.query(Setting)
536 .filter(Setting.key == key)
537 .delete()
538 )
540 if commit: 540 ↛ 541line 540 didn't jump to line 541 because the condition on line 540 was never true
541 self.db_session.commit()
543 return result > 0
544 except SQLAlchemyError:
545 logger.exception("Error deleting setting")
546 self.db_session.rollback()
547 return False
549 def load_from_defaults_file(
550 self, commit: bool = True, **kwargs: Any
551 ) -> None:
552 """
553 Import settings from the defaults settings file.
555 Args:
556 commit: Whether to commit changes to database
557 **kwargs: Will be passed to `import_settings`.
559 """
560 self.import_settings(self.default_settings, commit=commit, **kwargs)
562 def db_version_matches_package(self) -> bool:
563 """
564 Returns:
565 True if the version saved in the DB matches the package version.
567 """
568 db_version = self.get_setting("app.version")
569 logger.debug(
570 f"App version saved in DB is {db_version}, have package "
571 f"settings from version {package_version}."
572 )
574 return db_version == package_version
576 def update_db_version(self) -> None:
577 """
578 Updates the version saved in the DB based on the package version.
580 """
581 logger.debug(f"Updating saved DB version to {package_version}.")
583 self.delete_setting("app.version", commit=False)
584 version = Setting(
585 key="app.version",
586 value=package_version,
587 description="Version of the app this database is associated with.",
588 editable=False,
589 name="App Version",
590 type=SettingType.APP,
591 ui_element="text",
592 visible=False,
593 )
595 self.db_session.add(version)
596 self.db_session.commit()
598 def import_settings(
599 self,
600 settings_data: Dict[str, Any],
601 commit: bool = True,
602 overwrite: bool = True,
603 delete_extra: bool = False,
604 ) -> None:
605 """
606 Import settings directly from the export format. This can be used to
607 re-import settings that have been exported with `get_all_settings()`.
609 Args:
610 settings_data: The raw settings data to import.
611 commit: Whether to commit the DB after loading the settings.
612 overwrite: If true, it will overwrite the value of settings that
613 are already in the database.
614 delete_extra: If true, it will delete any settings that are in
615 the database but don't have a corresponding entry in
616 `settings_data`.
618 """
619 logger.debug(f"Importing {len(settings_data)} settings")
621 for key, setting_values in settings_data.items():
622 if not overwrite:
623 existing_value = self.get_setting(key)
624 if existing_value is not None: 624 ↛ 626line 624 didn't jump to line 626 because the condition on line 624 was never true
625 # Preserve the value from this setting.
626 setting_values["value"] = existing_value
628 # Delete any existing setting so we can completely overwrite it.
629 self.delete_setting(key, commit=False)
631 setting = Setting(key=key, **setting_values)
632 self.db_session.add(setting)
634 if commit or delete_extra: 634 ↛ 640line 634 didn't jump to line 640 because the condition on line 634 was always true
635 self.db_session.commit()
636 logger.info(f"Successfully imported {len(settings_data)} settings")
637 # Emit WebSocket event for all imported settings
638 self._emit_settings_changed(list(settings_data.keys()))
640 if delete_extra:
641 all_settings = self.get_all_settings()
642 for key in all_settings:
643 if key not in settings_data: 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true
644 logger.debug(f"Deleting extraneous setting: {key}")
645 self.delete_setting(key, commit=False)
647 def _create_setting(self, key, value, setting_type):
648 """Create a setting with appropriate metadata"""
650 # Determine appropriate category
651 category = None
652 ui_element = "text"
654 # Determine category based on key pattern
655 if key.startswith("app."):
656 category = "app_interface"
657 elif key.startswith("llm."):
658 if any(
659 param in key
660 for param in [
661 "temperature",
662 "max_tokens",
663 "n_batch",
664 "n_gpu_layers",
665 ]
666 ):
667 category = "llm_parameters"
668 else:
669 category = "llm_general"
670 elif key.startswith("search."):
671 if any(
672 param in key
673 for param in ["iterations", "questions", "results", "region"]
674 ):
675 category = "search_parameters"
676 else:
677 category = "search_general"
678 elif key.startswith("report."):
679 category = "report_parameters"
681 # Determine UI element type based on value
682 if isinstance(value, bool):
683 ui_element = "checkbox"
684 elif isinstance(value, (int, float)) and not isinstance(value, bool):
685 ui_element = "number"
686 elif isinstance(value, (dict, list)):
687 ui_element = "textarea"
689 # Build setting object
690 setting_dict = {
691 "key": key,
692 "value": value,
693 "type": setting_type.value.lower(),
694 "name": key.split(".")[-1].replace("_", " ").title(),
695 "description": f"Setting for {key}",
696 "category": category,
697 "ui_element": ui_element,
698 }
700 # Create the setting in the database
701 self.create_or_update_setting(setting_dict, commit=False)
703 def _emit_settings_changed(self, changed_keys: list = None):
704 """
705 Emit WebSocket event when settings change
707 Args:
708 changed_keys: List of setting keys that changed
709 """
710 try:
711 # Import here to avoid circular imports
712 from .socket_service import SocketIOService
714 try:
715 socket_service = SocketIOService()
716 except ValueError:
717 logger.debug(
718 "Not emitting socket event because server is not initialized."
719 )
720 return
722 # Get the changed settings
723 settings_data = {}
724 if changed_keys: 724 ↛ 731line 724 didn't jump to line 731 because the condition on line 724 was always true
725 for key in changed_keys:
726 setting_value = self.get_setting(key)
727 if setting_value is not None:
728 settings_data[key] = {"value": setting_value}
730 # Emit the settings change event
731 from datetime import datetime, UTC
733 socket_service.emit_socket_event(
734 "settings_changed",
735 {
736 "changed_keys": changed_keys or [],
737 "settings": settings_data,
738 "timestamp": datetime.now(UTC).isoformat(),
739 },
740 )
742 logger.debug(
743 f"Emitted settings_changed event for keys: {changed_keys}"
744 )
746 except Exception:
747 logger.exception("Failed to emit settings change event")
748 # Don't let WebSocket emission failures break settings saving