Coverage for src / local_deep_research / settings / manager.py: 96%

1import functools 

2import json 

3import os 

4import threading 

5from pathlib import Path 

6from typing import Any, Dict, List, Optional, Type, Union 

7 

8from loguru import logger 

9from sqlalchemy import func, or_ 

10from sqlalchemy.exc import SQLAlchemyError 

11from sqlalchemy.orm import Session 

12 

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 

26 

27 

28def parse_boolean(value: Any) -> bool: 

29 """ 

30 Convert various representations to boolean using HTML checkbox semantics. 

31 

32 This function handles form values, JSON booleans, and environment variables, 

33 ensuring consistent behavior across client and server. 

34 

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 

39 

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 

47 

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 

53 

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. 

58 

59 Args: 

60 value: Value to convert to boolean. Accepts strings, booleans, or None. 

61 

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) 

65 

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") 

72 

73 # Handle already-boolean values 

74 if isinstance(value, bool): 

75 return value 

76 

77 # Handle None (missing values) 

78 if value is None: 

79 return False 

80 

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 

89 

90 # For other types (numbers, lists, etc.), use Python's bool conversion 

91 return bool(value) 

92 

93 

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 

100 

101 

102def _parse_json_value(x): 

103 """Parse JSON ui_element values. 

104 

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 return x 

117 return x 

118 

119 

120def _parse_multiselect(x): 

121 """Parse multiselect value, handling both lists and strings. 

122 

123 DB values (via SQLAlchemy JSON column) arrive as Python lists already. 

124 Env var overrides arrive as strings and need parsing — either as JSON 

125 arrays (e.g. '["markdown","latex"]') or comma-separated values 

126 (e.g. 'markdown,latex'). 

127 """ 

128 if isinstance(x, list): 

129 return x 

130 if isinstance(x, str): 

131 stripped = x.strip() 

132 if stripped.startswith("["): 

133 try: 

134 parsed = json.loads(stripped) 

135 if isinstance(parsed, list): 135 ↛ 140line 135 didn't jump to line 140 because the condition on line 135 was always true

136 return parsed 

137 except (json.JSONDecodeError, ValueError): 

138 pass 

139 # Comma-separated fallback 

140 return [item.strip() for item in stripped.split(",") if item.strip()] 

141 return x 

142 

143 

144def _filter_setting_columns(data: dict) -> dict: 

145 """Filter a dict to only keys that are valid Setting model columns. 

146 

147 Prevents crashes when default_settings.json contains keys not present 

148 as columns on the Setting model (e.g. future flags). 

149 """ 

150 valid_columns = {c.name for c in Setting.__table__.columns} 

151 return {k: v for k, v in data.items() if k in valid_columns} 

152 

153 

154UI_ELEMENT_TO_SETTING_TYPE = { 

155 "text": str, 

156 "json": _parse_json_value, 

157 "password": str, 

158 "select": str, 

159 "number": _parse_number, 

160 "range": _parse_number, # Same behavior as number for consistency 

161 "checkbox": parse_boolean, 

162 "textarea": str, 

163 "multiselect": _parse_multiselect, 

164} 

165 

166 

167def get_typed_setting_value( 

168 key: str, 

169 value: Any, 

170 ui_element: str, 

171 default: Any = None, 

172 check_env: bool = True, 

173) -> Any: 

174 """ 

175 Extracts the value for a particular setting, ensuring that it has the 

176 correct type. 

177 

178 Args: 

179 key: The setting key. 

180 value: The setting value from the database. 

181 ui_element: The setting UI element ID. 

182 default: Default value to return if the value of the setting is 

183 invalid. 

184 check_env: If true, it will check the environment variable for 

185 this setting before reading from the DB. 

186 

187 Returns: 

188 The value of the setting. 

189 

190 """ 

191 setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None) 

192 if setting_type is None: 

193 logger.warning( 

194 "Got unknown type {} for setting {}, returning default value.", 

195 ui_element, 

196 key, 

197 ) 

198 return default 

199 

200 # Check environment variable first (highest priority). 

201 if check_env: 

202 env_value = check_env_setting(key) 

203 if env_value is not None: 

204 try: 

205 return setting_type(env_value) 

206 except ValueError: 

207 logger.warning( 

208 "Setting {} has invalid value {}. Falling back to DB.", 

209 key, 

210 env_value, 

211 ) 

212 

213 # If value is None (not in database), return default. 

214 if value is None: 

215 return default 

216 

217 # Read from the database. 

218 try: 

219 return setting_type(value) 

220 except (ValueError, TypeError): 

221 logger.warning( 

222 "Setting {} has invalid value {}. Returning default.", 

223 key, 

224 value, 

225 ) 

226 return default 

227 

228 

229def check_env_setting(key: str) -> str | None: 

230 """ 

231 Checks environment variables for a particular setting. 

232 

233 Args: 

234 key: The database key for the setting. 

235 

236 Returns: 

237 The setting from the environment variables, or None if the variable 

238 is not set. 

239 

240 """ 

241 env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}" 

242 env_value = os.getenv(env_variable_name) 

243 if env_value is not None: 

244 logger.debug(f"Overriding {key} setting from environment variable.") 

245 return env_value 

246 

247 

248class SettingsManager(ISettingsManager): 

249 """ 

250 Manager for handling application settings with database storage and file fallback. 

251 Provides methods to get and set settings, with the ability to override settings in memory. 

252 """ 

253 

254 def __init__(self, db_session: Optional[Session] = None): 

255 """ 

256 Initialize the settings manager 

257 

258 Args: 

259 db_session: SQLAlchemy session for database operations 

260 """ 

261 self.db_session = db_session 

262 self.db_first = True # Always prioritize DB settings 

263 

264 # Store the thread ID this instance was created in 

265 self._creation_thread_id = threading.get_ident() 

266 

267 # Initialize settings lock as None - will be checked lazily 

268 self.__settings_locked = None 

269 

270 # Auto-initialize settings if database is empty 

271 if self.db_session: 

272 self._ensure_settings_initialized() 

273 

274 def _ensure_settings_initialized(self): 

275 """Ensure settings are initialized in the database.""" 

276 # Check if we have any settings at all 

277 from ..database.models import Setting 

278 

279 settings_count = self.db_session.query(Setting).count() 

280 

281 if settings_count == 0: 

282 logger.info("No settings found in database, loading defaults") 

283 self.load_from_defaults_file(commit=True) 

284 logger.info("Default settings loaded successfully") 

285 

286 def _check_thread_safety(self): 

287 """Check if this instance is being used in the same thread it was created in.""" 

288 current_thread_id = threading.get_ident() 

289 if self.db_session and current_thread_id != self._creation_thread_id: 

290 raise RuntimeError( 

291 f"SettingsManager instance created in thread {self._creation_thread_id} " 

292 f"is being used in thread {current_thread_id}. This is not thread-safe! " 

293 f"Create a new SettingsManager instance within the current thread context." 

294 ) 

295 

296 @property 

297 def settings_locked(self) -> bool: 

298 """Check if settings are locked (lazy evaluation).""" 

299 if self.__settings_locked is None: 

300 try: 

301 self.__settings_locked = self.get_setting( 

302 "app.lock_settings", False 

303 ) 

304 if self.settings_locked: 

305 logger.info( 

306 "Settings are locked. Disabling all settings changes." 

307 ) 

308 except Exception: 

309 # If we can't check, assume not locked 

310 self.__settings_locked = False 

311 return self.__settings_locked 

312 

313 @functools.cached_property 

314 def default_settings(self) -> Dict[str, Any]: 

315 """ 

316 Returns: 

317 The default settings, loaded from JSON files and merged. 

318 Automatically discovers and loads all .json files in the defaults 

319 directory and its subdirectories. 

320 Theme options are dynamically injected from the theme registry. 

321 

322 """ 

323 settings = {} 

324 

325 try: 

326 # Get the defaults package path 

327 defaults_path = Path(defaults.__file__).parent 

328 

329 # Find all JSON files recursively in the defaults directory 

330 json_files = sorted(defaults_path.rglob("*.json")) 

331 

332 logger.debug(f"Found {len(json_files)} JSON settings files") 

333 

334 # Load and merge all JSON files 

335 for json_file in json_files: 

336 try: 

337 with open(json_file, "r") as f: 

338 file_settings = json.load(f) 

339 

340 # Get relative path for logging 

341 relative_path = json_file.relative_to(defaults_path) 

342 

343 # Warn about key conflicts 

344 conflicts = set(settings.keys()) & set(file_settings.keys()) 

345 if conflicts: 

346 logger.warning( 

347 f"Keys {conflicts} from {relative_path} " 

348 f"override existing values" 

349 ) 

350 

351 settings.update(file_settings) 

352 logger.debug(f"Loaded {relative_path}") 

353 

354 except json.JSONDecodeError: 

355 logger.exception(f"Invalid JSON in {json_file}") 

356 except Exception as e: 

357 logger.warning(f"Could not load {json_file}: {e}") 

358 

359 except Exception as e: 

360 logger.warning(f"Error loading settings files: {e}") 

361 

362 # Inject dynamic theme options from theme registry 

363 if "app.theme" in settings: 

364 try: 

365 from local_deep_research.web.themes import theme_registry 

366 

367 settings["app.theme"]["options"] = ( 

368 theme_registry.get_settings_options() 

369 ) 

370 except ImportError: 

371 # Theme registry not available, use static options from JSON 

372 pass 

373 

374 logger.debug(f"Loaded {len(settings)} total settings") 

375 return settings 

376 

377 def __get_typed_setting_value( 

378 self, 

379 setting: Type[Setting], 

380 default: Any = None, 

381 check_env: bool = True, 

382 ) -> Any: 

383 """ 

384 Extracts the value for a particular setting, ensuring that it has the 

385 correct type. 

386 

387 Args: 

388 setting: The setting to get the value for. 

389 default: Default value to return if the value of the setting is 

390 invalid. 

391 check_env: If true, it will check the environment variable for 

392 this setting before reading from the DB. 

393 

394 Returns: 

395 The value of the setting. 

396 

397 """ 

398 return get_typed_setting_value( 

399 setting.key, 

400 setting.value, 

401 setting.ui_element, 

402 default=default, 

403 check_env=check_env, 

404 ) 

405 

406 def __query_settings(self, key: str | None = None) -> List[Type[Setting]]: 

407 """ 

408 Abstraction for querying settings that also transparently handles 

409 reading the default settings file if the DB is not enabled. 

410 

411 Args: 

412 key: The key to read. If None, it will read everything. 

413 

414 Returns: 

415 The settings it queried. 

416 

417 """ 

418 if self.db_session: 

419 self._check_thread_safety() 

420 query = self.db_session.query(Setting) 

421 if key is not None: 

422 # This will find exact matches and any subkeys. 

423 query = query.filter( 

424 or_( 

425 Setting.key == key, 

426 Setting.key.startswith(f"{key}."), 

427 ) 

428 ) 

429 return query.all() 

430 

431 else: 

432 logger.debug( 

433 "DB is disabled, reading setting '{}' from defaults file.", key 

434 ) 

435 

436 settings = [] 

437 for candidate_key, setting in self.default_settings.items(): 

438 if key is None or ( 

439 candidate_key == key or candidate_key.startswith(f"{key}.") 

440 ): 

441 settings.append( 

442 Setting( 

443 key=candidate_key, # gitleaks:allow 

444 **_filter_setting_columns(setting), 

445 ) 

446 ) 

447 

448 return settings 

449 

450 def get_setting( 

451 self, key: str, default: Any = None, check_env: bool = True 

452 ) -> Any: 

453 """ 

454 Get a setting value 

455 

456 Args: 

457 key: Setting key 

458 default: Default value if setting is not found 

459 check_env: If true, it will check the environment variable for 

460 this setting before reading from the DB. 

461 

462 Returns: 

463 Setting value or default if not found 

464 """ 

465 

466 # First check if this is an env-only setting 

467 if env_registry.is_env_only(key): 

468 return env_registry.get(key, default) 

469 

470 # If using database first approach and session available, check database 

471 try: 

472 settings = self.__query_settings(key) 

473 if len(settings) == 1: 

474 # This is a bottom-level key. 

475 result = self.__get_typed_setting_value( 

476 settings[0], default, check_env 

477 ) 

478 # Cache the result 

479 return result 

480 elif len(settings) > 1: 

481 # This is a higher-level key. 

482 settings_map = {} 

483 for setting in settings: 

484 output_key = setting.key.removeprefix(f"{key}.") 

485 settings_map[output_key] = self.__get_typed_setting_value( 

486 setting, default, check_env 

487 ) 

488 return settings_map 

489 except SQLAlchemyError: 

490 logger.exception(f"Error retrieving setting {key} from database") 

491 

492 # Check env var before returning default (setting not in DB) 

493 if check_env: 

494 env_value = check_env_setting(key) 

495 if env_value is not None: 

496 default_meta = self.default_settings.get(key) 

497 if default_meta and isinstance(default_meta, dict): 

498 ui_element = default_meta.get("ui_element", "text") 

499 return get_typed_setting_value( 

500 key, 

501 None, 

502 ui_element, 

503 default=default, 

504 check_env=True, 

505 ) 

506 logger.warning( 

507 "Setting '{}' has env var override but is not in " 

508 "defaults — returning raw string without type " 

509 "conversion. Add this setting to a defaults JSON " 

510 "file with a ui_element type to enable proper " 

511 "type conversion.", 

512 key, 

513 ) 

514 return env_value 

515 

516 # Return default if not found 

517 return default 

518 

519 def get_bool_setting( 

520 self, key: str, default: bool = False, check_env: bool = True 

521 ) -> bool: 

522 """ 

523 Get a setting value as a boolean, handling string conversion. 

524 

525 Args: 

526 key: Setting key 

527 default: Default boolean value if setting is not found 

528 check_env: If true, it will check the environment variable for 

529 this setting before reading from the DB. 

530 

531 Returns: 

532 Boolean value of the setting 

533 """ 

534 value = self.get_setting(key, default, check_env) 

535 return to_bool(value, default) 

536 

537 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool: 

538 """ 

539 Set a setting value 

540 

541 Args: 

542 key: Setting key 

543 value: Setting value 

544 commit: Whether to commit the change 

545 

546 Returns: 

547 True if successful, False otherwise 

548 """ 

549 if not self.db_session: 

550 logger.error( 

551 "Cannot edit setting {} because no DB was provided.", key 

552 ) 

553 return False 

554 if self.settings_locked: 

555 logger.error("Cannot edit setting {} because they are locked.", key) 

556 return False 

557 

558 # Always update database if available 

559 try: 

560 self._check_thread_safety() 

561 setting = ( 

562 self.db_session.query(Setting) 

563 .filter(Setting.key == key) 

564 .first() 

565 ) 

566 if setting: 

567 if not setting.editable: 

568 logger.error( 

569 "Cannot change setting '{}' because it " 

570 "is marked as non-editable.", 

571 key, 

572 ) 

573 return False 

574 

575 setting.value = value 

576 setting.updated_at = ( 

577 func.now() 

578 ) # Explicitly set the current timestamp 

579 else: 

580 # Determine setting type from key 

581 setting_type = SettingType.APP 

582 if key.startswith("llm."): 

583 setting_type = SettingType.LLM 

584 elif key.startswith("search."): 

585 setting_type = SettingType.SEARCH 

586 elif key.startswith("report."): 

587 setting_type = SettingType.REPORT 

588 elif key.startswith("database."): 588 ↛ 589line 588 didn't jump to line 589 because the condition on line 588 was never true

589 setting_type = SettingType.DATABASE 

590 

591 # Create a new setting 

592 new_setting = Setting( 

593 key=key, 

594 value=value, 

595 type=setting_type, 

596 name=key.split(".")[-1].replace("_", " ").title(), 

597 ui_element="text", 

598 description=f"Setting for {key}", 

599 ) 

600 self.db_session.add(new_setting) 

601 

602 if commit: 602 ↛ 607line 602 didn't jump to line 607 because the condition on line 602 was always true

603 self.db_session.commit() 

604 # Emit WebSocket event for settings change 

605 self._emit_settings_changed([key]) 

606 

607 return True 

608 except SQLAlchemyError: 

609 logger.exception(f"Error setting value for key: {key}") 

610 self.db_session.rollback() 

611 return False 

612 

613 def clear_cache(self): 

614 """Clear the settings cache.""" 

615 self.__dict__.pop("default_settings", None) 

616 logger.debug("Settings cache cleared") 

617 

618 def get_all_settings(self, bypass_cache: bool = False) -> Dict[str, Any]: 

619 """ 

620 Get all settings, merging defaults with database values. 

621 

622 This ensures that new settings added to defaults.json automatically 

623 appear in the UI without requiring a database reset. 

624 

625 Args: 

626 bypass_cache: If True, bypass the cache and read directly from database 

627 

628 Returns: 

629 Dictionary of all settings 

630 """ 

631 result = {} 

632 

633 # Start with defaults so new settings are always included 

634 for key, default_setting in self.default_settings.items(): 

635 result[key] = dict(default_setting) 

636 

637 # Check env var override for defaults not yet in DB 

638 env_value = check_env_setting(key) 

639 if env_value is not None: 639 ↛ 640line 639 didn't jump to line 640 because the condition on line 639 was never true

640 ui_element = default_setting.get("ui_element", "text") 

641 typed_value = get_typed_setting_value( 

642 key, 

643 None, 

644 ui_element, 

645 default=env_value, 

646 check_env=True, 

647 ) 

648 result[key]["value"] = typed_value 

649 result[key]["editable"] = False 

650 

651 # Override with database settings 

652 try: 

653 db_settings = self.__query_settings() 

654 except SQLAlchemyError: 

655 logger.exception( 

656 "Error querying settings from database in get_all_settings" 

657 ) 

658 db_settings = [] 

659 

660 for setting in db_settings: 

661 # Handle type field - it might be a string or an enum 

662 setting_type = setting.type 

663 if hasattr(setting_type, "name"): 

664 setting_type = setting_type.name 

665 

666 # Log if this is a custom setting not in defaults 

667 if setting.key not in result: 

668 logger.debug( 

669 f"Database contains custom setting not in " 

670 f"defaults: {setting.key} (type={setting_type}, " 

671 f"category={setting.category})" 

672 ) 

673 

674 # Override default with database value 

675 result[setting.key] = dict( 

676 value=setting.value, 

677 type=setting_type, 

678 name=setting.name, 

679 description=setting.description, 

680 category=setting.category, 

681 ui_element=setting.ui_element, 

682 options=setting.options, 

683 min_value=setting.min_value, 

684 max_value=setting.max_value, 

685 step=setting.step, 

686 visible=setting.visible, 

687 editable=False if self.settings_locked else setting.editable, 

688 ) 

689 

690 # Override from the environment variables if needed. 

691 env_value = check_env_setting(setting.key) 

692 if env_value is not None: 

693 ui_element = result[setting.key].get( 

694 "ui_element", setting.ui_element 

695 ) 

696 typed_value = get_typed_setting_value( 

697 setting.key, 

698 None, 

699 ui_element, 

700 default=env_value, 

701 check_env=True, 

702 ) 

703 result[setting.key]["value"] = typed_value 

704 # Mark it as non-editable, because changes to the DB 

705 # value have no effect as long as the environment 

706 # variable is set. 

707 result[setting.key]["editable"] = False 

708 

709 return result 

710 

711 def get_settings_snapshot(self) -> Dict[str, Any]: 

712 """ 

713 Get a simplified settings snapshot with just key-value pairs. 

714 This is useful for passing settings to background threads or storing in metadata. 

715 

716 Returns: 

717 Dictionary with setting keys mapped to their values 

718 """ 

719 all_settings = self.get_all_settings() 

720 settings_snapshot = {} 

721 

722 for key, setting in all_settings.items(): 

723 if isinstance(setting, dict) and "value" in setting: 723 ↛ 726line 723 didn't jump to line 726 because the condition on line 723 was always true

724 settings_snapshot[key] = setting["value"] 

725 else: 

726 settings_snapshot[key] = setting 

727 

728 return settings_snapshot 

729 

730 def create_or_update_setting( 

731 self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True 

732 ) -> Optional[Setting]: 

733 """ 

734 Create or update a setting 

735 

736 Args: 

737 setting: Setting object or dictionary 

738 commit: Whether to commit the change 

739 

740 Returns: 

741 The created or updated Setting model, or None if failed 

742 """ 

743 if not self.db_session: 

744 logger.warning( 

745 "No database session available, cannot create/update setting" 

746 ) 

747 return None 

748 if self.settings_locked: 

749 logger.error("Cannot edit settings because they are locked.") 

750 return None 

751 

752 # Convert dict to BaseSetting if needed 

753 if isinstance(setting, dict): 753 ↛ 769line 753 didn't jump to line 769 because the condition on line 753 was always true

754 # Determine type from key if not specified 

755 if "type" not in setting and "key" in setting: 

756 key = setting["key"] 

757 if key.startswith("llm."): 

758 setting_obj = LLMSetting(**setting) 

759 elif key.startswith("search."): 

760 setting_obj = SearchSetting(**setting) 

761 elif key.startswith("report."): 

762 setting_obj = ReportSetting(**setting) 

763 else: 

764 setting_obj = AppSetting(**setting) 

765 else: 

766 # Use generic BaseSetting 

767 setting_obj = BaseSetting(**setting) 

768 else: 

769 setting_obj = setting 

770 

771 try: 

772 # Check if setting exists 

773 db_setting = ( 

774 self.db_session.query(Setting) 

775 .filter(Setting.key == setting_obj.key) 

776 .first() 

777 ) 

778 

779 if db_setting: 

780 # Update existing setting 

781 if not db_setting.editable: 

782 logger.error( 

783 "Cannot change setting '{}' because it " 

784 "is marked as non-editable.", 

785 setting_obj.key, 

786 ) 

787 return None 

788 

789 db_setting.value = setting_obj.value 

790 db_setting.name = setting_obj.name 

791 db_setting.description = setting_obj.description 

792 db_setting.category = setting_obj.category 

793 db_setting.ui_element = setting_obj.ui_element 

794 db_setting.options = setting_obj.options 

795 db_setting.min_value = setting_obj.min_value 

796 db_setting.max_value = setting_obj.max_value 

797 db_setting.step = setting_obj.step 

798 db_setting.visible = setting_obj.visible 

799 db_setting.editable = setting_obj.editable 

800 db_setting.updated_at = ( 

801 func.now() 

802 ) # Explicitly set the current timestamp 

803 else: 

804 # Create new setting 

805 db_setting = Setting( 

806 key=setting_obj.key, 

807 value=setting_obj.value, 

808 type=setting_obj.type, 

809 name=setting_obj.name, 

810 description=setting_obj.description, 

811 category=setting_obj.category, 

812 ui_element=setting_obj.ui_element, 

813 options=setting_obj.options, 

814 min_value=setting_obj.min_value, 

815 max_value=setting_obj.max_value, 

816 step=setting_obj.step, 

817 visible=setting_obj.visible, 

818 editable=setting_obj.editable, 

819 ) 

820 self.db_session.add(db_setting) 

821 

822 if commit: 

823 self.db_session.commit() 

824 # Emit WebSocket event for settings change 

825 self._emit_settings_changed([setting_obj.key]) 

826 

827 return db_setting 

828 

829 except SQLAlchemyError: 

830 logger.exception( 

831 f"Error creating/updating setting {setting_obj.key}" 

832 ) 

833 self.db_session.rollback() 

834 return None 

835 

836 def delete_setting(self, key: str, commit: bool = True) -> bool: 

837 """ 

838 Delete a setting 

839 

840 Args: 

841 key: Setting key 

842 commit: Whether to commit the change 

843 

844 Returns: 

845 True if successful, False otherwise 

846 """ 

847 if not self.db_session: 

848 logger.warning( 

849 "No database session available, cannot delete setting" 

850 ) 

851 return False 

852 

853 try: 

854 # Remove from database 

855 result = ( 

856 self.db_session.query(Setting) 

857 .filter(Setting.key == key) 

858 .delete() 

859 ) 

860 

861 if commit: 

862 self.db_session.commit() 

863 

864 return result > 0 

865 except SQLAlchemyError: 

866 logger.exception("Error deleting setting") 

867 self.db_session.rollback() 

868 return False 

869 

870 def load_from_defaults_file( 

871 self, commit: bool = True, **kwargs: Any 

872 ) -> None: 

873 """ 

874 Import settings from the defaults settings file. 

875 

876 Args: 

877 commit: Whether to commit changes to database 

878 **kwargs: Will be passed to `import_settings`. 

879 

880 """ 

881 self.import_settings(self.default_settings, commit=commit, **kwargs) 

882 

883 def db_version_matches_package(self) -> bool: 

884 """ 

885 Returns: 

886 True if the version saved in the DB matches the package version. 

887 

888 """ 

889 db_version = self.get_setting("app.version") 

890 logger.debug( 

891 f"App version saved in DB is {db_version}, have package " 

892 f"settings from version {package_version}." 

893 ) 

894 

895 return db_version == package_version 

896 

897 def update_db_version(self) -> None: 

898 """ 

899 Updates the version saved in the DB based on the package version. 

900 

901 """ 

902 logger.debug(f"Updating saved DB version to {package_version}.") 

903 

904 self.delete_setting("app.version", commit=False) 

905 version = Setting( 

906 key="app.version", 

907 value=package_version, 

908 description="Version of the app this database is associated with.", 

909 editable=False, 

910 name="App Version", 

911 type=SettingType.APP, 

912 ui_element="text", 

913 visible=False, 

914 ) 

915 

916 self.db_session.add(version) 

917 self.db_session.commit() 

918 

919 def import_settings( 

920 self, 

921 settings_data: Dict[str, Any], 

922 commit: bool = True, 

923 overwrite: bool = True, 

924 delete_extra: bool = False, 

925 ) -> None: 

926 """ 

927 Import settings directly from the export format. This can be used to 

928 re-import settings that have been exported with `get_all_settings()`. 

929 

930 Args: 

931 settings_data: The raw settings data to import. 

932 commit: Whether to commit the DB after loading the settings. 

933 overwrite: If true, it will overwrite the value of settings that 

934 are already in the database. 

935 delete_extra: If true, it will delete any settings that are in 

936 the database but don't have a corresponding entry in 

937 `settings_data`. 

938 

939 """ 

940 logger.debug(f"Importing {len(settings_data)} settings") 

941 

942 for key, setting_values in settings_data.items():