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

1import importlib.resources as pkg_resources 

2import json 

3import os 

4import threading 

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

6 

7from loguru import logger 

8from sqlalchemy import func, or_ 

9from sqlalchemy.exc import SQLAlchemyError 

10from sqlalchemy.orm import Session 

11 

12from .. import defaults 

13from ..__version__ import __version__ as package_version 

14from ..database.models import Setting, SettingType 

15from ..web.models.settings import ( 

16 AppSetting, 

17 BaseSetting, 

18 LLMSetting, 

19 ReportSetting, 

20 SearchSetting, 

21) 

22from .base import ISettingsManager 

23from .env_registry import registry as env_registry 

24 

25 

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

27 """ 

28 Convert various representations to boolean using HTML checkbox semantics. 

29 

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

31 ensuring consistent behavior across client and server. 

32 

33 **HTML Checkbox Semantics** (INTENTIONAL DESIGN): 

34 - **Any value present (except explicit false) = checked = True** 

35 - This matches standard HTML form behavior where checkbox presence indicates checked state 

36 - In HTML forms, checkboxes send a value when checked, nothing when unchecked 

37 

38 **Examples**: 

39 parse_boolean("on") # True - standard HTML checkbox value 

40 parse_boolean("true") # True - explicit true 

41 parse_boolean("1") # True - numeric true 

42 parse_boolean("enabled") # True - any non-empty string 

43 parse_boolean("disabled") # True - INTENTIONAL: any string = checkbox was checked! 

44 parse_boolean("custom") # True - custom checkbox value 

45 

46 parse_boolean("false") # False - explicit false 

47 parse_boolean("off") # False - explicit false 

48 parse_boolean("0") # False - explicit false 

49 parse_boolean("") # False - empty string = unchecked 

50 parse_boolean(None) # False - missing = unchecked 

51 

52 **Why "disabled" returns True**: 

53 This is NOT a bug! If a checkbox sends the value "disabled", it means the checkbox 

54 was checked (present in form data). The actual string content doesn't matter for 

55 HTML checkboxes - only presence vs absence matters. 

56 

57 Args: 

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

59 

60 Returns: 

61 bool: True for truthy values (any non-empty string except explicit false); 

62 False for falsy values ('off', 'false', '0', '', 'no', False, None) 

63 

64 Note: 

65 This function implements HTML form semantics, NOT generic boolean parsing. 

66 See tests/settings/test_boolean_parsing.py for comprehensive test coverage. 

67 """ 

68 # Constants for boolean value parsing 

69 FALSY_VALUES = ("off", "false", "0", "", "no") 

70 

71 # Handle already-boolean values 

72 if isinstance(value, bool): 

73 return value 

74 

75 # Handle None (missing values) 

76 if value is None: 

77 return False 

78 

79 # Handle string values 

80 if isinstance(value, str): 

81 value_lower = value.lower().strip() 

82 # Explicitly falsy values (empty string, false-like values) 

83 if value_lower in FALSY_VALUES: 

84 return False 

85 # Any other non-empty string = True (HTML checkbox semantics) 

86 return True 

87 

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

89 return bool(value) 

90 

91 

92def _parse_number(x): 

93 """Parse number, returning int if it's a whole number, otherwise float.""" 

94 f = float(x) 

95 if f.is_integer(): 

96 return int(f) 

97 return f 

98 

99 

100_UI_ELEMENT_TO_SETTING_TYPE = { 

101 "text": str, 

102 # SQLAlchemy should already handle JSON parsing. 

103 "json": lambda x: x, 

104 "password": str, 

105 "select": str, 

106 "number": _parse_number, 

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

108 "checkbox": parse_boolean, 

109} 

110 

111 

112def get_typed_setting_value( 

113 key: str, 

114 value: Any, 

115 ui_element: str, 

116 default: Any = None, 

117 check_env: bool = True, 

118) -> Any: 

119 """ 

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

121 correct type. 

122 

123 Args: 

124 key: The setting key. 

125 value: The setting value from the database. 

126 ui_element: The setting UI element ID. 

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

128 invalid. 

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

130 this setting before reading from the DB. 

131 

132 Returns: 

133 The value of the setting. 

134 

135 """ 

136 setting_type = _UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None) 

137 if setting_type is None: 

138 logger.warning( 

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

140 ui_element, 

141 key, 

142 ) 

143 return default 

144 

145 # Check environment variable first (highest priority). 

146 if check_env: 

147 env_value = check_env_setting(key) 

148 if env_value is not None: 

149 try: 

150 # Special handling for boolean values 

151 if setting_type is bool: 151 ↛ 153line 151 didn't jump to line 153 because the condition on line 151 was never true

152 # Convert string to boolean properly 

153 return env_value.lower() in ("true", "1", "yes", "on") 

154 return setting_type(env_value) 

155 except ValueError: 

156 logger.warning( 

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

158 key, 

159 env_value, 

160 ) 

161 

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

163 if value is None: 

164 return default 

165 

166 # Read from the database. 

167 try: 

168 return setting_type(value) 

169 except (ValueError, TypeError): 

170 logger.warning( 

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

172 key, 

173 value, 

174 ) 

175 return default 

176 

177 

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

179 """ 

180 Checks environment variables for a particular setting. 

181 

182 Args: 

183 key: The database key for the setting. 

184 

185 Returns: 

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

187 is not set. 

188 

189 """ 

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

191 env_value = os.getenv(env_variable_name) 

192 if env_value is not None: 

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

194 return env_value 

195 

196 

197class SettingsManager(ISettingsManager): 

198 """ 

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

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

201 """ 

202 

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

204 """ 

205 Initialize the settings manager 

206 

207 Args: 

208 db_session: SQLAlchemy session for database operations 

209 """ 

210 self.db_session = db_session 

211 self.db_first = True # Always prioritize DB settings 

212 

213 # Store the thread ID this instance was created in 

214 self._creation_thread_id = threading.get_ident() 

215 

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

217 self.__settings_locked = None 

218 

219 # Auto-initialize settings if database is empty 

220 if self.db_session: 

221 self._ensure_settings_initialized() 

222 

223 def _ensure_settings_initialized(self): 

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

225 # Check if we have any settings at all 

226 from ..database.models import Setting 

227 

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

229 

230 if settings_count == 0: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true

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

232 self.load_from_defaults_file(commit=True) 

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

234 

235 def _check_thread_safety(self): 

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

237 current_thread_id = threading.get_ident() 

238 if self.db_session and current_thread_id != self._creation_thread_id: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 raise RuntimeError( 

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

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

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

243 ) 

244 

245 @property 

246 def settings_locked(self) -> bool: 

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

248 if self.__settings_locked is None: 

249 try: 

250 self.__settings_locked = self.get_setting( 

251 "app.lock_settings", False 

252 ) 

253 if self.settings_locked: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true

254 logger.info( 

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

256 ) 

257 except Exception: 

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

259 self.__settings_locked = False 

260 return self.__settings_locked 

261 

262 @property 

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

264 """ 

265 Returns: 

266 The default settings, loaded from JSON. 

267 

268 """ 

269 default_settings = pkg_resources.read_text( 

270 defaults, "default_settings.json" 

271 ) 

272 return json.loads(default_settings) 

273 

274 def __get_typed_setting_value( 

275 self, 

276 setting: Type[Setting], 

277 default: Any = None, 

278 check_env: bool = True, 

279 ) -> Any: 

280 """ 

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

282 correct type. 

283 

284 Args: 

285 setting: The setting to get the value for. 

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

287 invalid. 

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

289 this setting before reading from the DB. 

290 

291 Returns: 

292 The value of the setting. 

293 

294 """ 

295 return get_typed_setting_value( 

296 setting.key, 

297 setting.value, 

298 setting.ui_element, 

299 default=default, 

300 check_env=check_env, 

301 ) 

302 

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

304 """ 

305 Abstraction for querying settings that also transparently handles 

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

307 

308 Args: 

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

310 

311 Returns: 

312 The settings it queried. 

313 

314 """ 

315 if self.db_session: 

316 self._check_thread_safety() 

317 query = self.db_session.query(Setting) 

318 if key is not None: 

319 # This will find exact matches and any subkeys. 

320 query = query.filter( 

321 or_( 

322 Setting.key == key, 

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

324 ) 

325 ) 

326 return query.all() 

327 

328 else: 

329 logger.debug( 

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

331 ) 

332 

333 settings = [] 

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

335 if key is None or ( 

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

337 ): 

338 settings.append(Setting(key=candidate_key, **setting)) 

339 

340 return settings 

341 

342 def get_setting( 

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

344 ) -> Any: 

345 """ 

346 Get a setting value 

347 

348 Args: 

349 key: Setting key 

350 default: Default value if setting is not found 

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

352 this setting before reading from the DB. 

353 

354 Returns: 

355 Setting value or default if not found 

356 """ 

357 

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

359 if env_registry.is_env_only(key): 

360 return env_registry.get(key, default) 

361 

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

363 try: 

364 settings = self.__query_settings(key) 

365 if len(settings) == 1: 

366 # This is a bottom-level key. 

367 result = self.__get_typed_setting_value( 

368 settings[0], default, check_env 

369 ) 

370 # Cache the result 

371 return result 

372 elif len(settings) > 1: 

373 # This is a higher-level key. 

374 settings_map = {} 

375 for setting in settings: 

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

377 settings_map[output_key] = self.__get_typed_setting_value( 

378 setting, default, check_env 

379 ) 

380 return settings_map 

381 except SQLAlchemyError as e: 

382 logger.exception( 

383 f"Error retrieving setting {key} from database: {e}" 

384 ) 

385 

386 # Return default if not found 

387 return default 

388 

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

390 """ 

391 Set a setting value 

392 

393 Args: 

394 key: Setting key 

395 value: Setting value 

396 commit: Whether to commit the change 

397 

398 Returns: 

399 True if successful, False otherwise 

400 """ 

401 if not self.db_session: 

402 logger.error( 

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

404 ) 

405 return False 

406 if self.settings_locked: 

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

408 return False 

409 

410 # Always update database if available 

411 try: 

412 self._check_thread_safety() 

413 setting = ( 

414 self.db_session.query(Setting) 

415 .filter(Setting.key == key) 

416 .first() 

417 ) 

418 if setting: 

419 if not setting.editable: 

420 logger.error( 

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

422 "is marked as non-editable.", 

423 key, 

424 ) 

425 return False 

426 

427 setting.value = value 

428 setting.updated_at = ( 

429 func.now() 

430 ) # Explicitly set the current timestamp 

431 else: 

432 # Determine setting type from key 

433 setting_type = SettingType.APP 

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

435 setting_type = SettingType.LLM 

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

437 setting_type = SettingType.SEARCH 

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

439 setting_type = SettingType.REPORT 

440 elif key.startswith("database."): 

441 setting_type = SettingType.DATABASE 

442 

443 # Create a new setting 

444 new_setting = Setting( 

445 key=key, 

446 value=value, 

447 type=setting_type, 

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

449 ui_element="text", 

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

451 ) 

452 self.db_session.add(new_setting) 

453 

454 if commit: 

455 self.db_session.commit() 

456 # Emit WebSocket event for settings change 

457 self._emit_settings_changed([key]) 

458 

459 return True 

460 except SQLAlchemyError: 

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

462 self.db_session.rollback() 

463 return False 

464 

465 def clear_cache(self): 

466 """Clear the settings cache.""" 

467 logger.debug("Settings cache cleared") 

468 

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

470 """ 

471 Get all settings, merging defaults with database values. 

472 

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

474 appear in the UI without requiring a database reset. 

475 

476 Args: 

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

478 

479 Returns: 

480 Dictionary of all settings 

481 """ 

482 result = {} 

483 

484 # Start with defaults so new settings are always included 

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

486 result[key] = dict(default_setting) 

487 

488 # Override with database settings if available 

489 try: 

490 for setting in self.__query_settings(): 

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

492 setting_type = setting.type 

493 if hasattr(setting_type, "name"): 493 ↛ 497line 493 didn't jump to line 497 because the condition on line 493 was always true

494 setting_type = setting_type.name 

495 

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

497 if setting.key not in self.default_settings: 

498 logger.debug( 

499 f"Database contains custom setting not in " 

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

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

502 ) 

503 

504 # Override default with database value 

505 result[setting.key] = dict( 

506 value=setting.value, 

507 type=setting_type, 

508 name=setting.name, 

509 description=setting.description, 

510 category=setting.category, 

511 ui_element=setting.ui_element, 

512 options=setting.options, 

513 min_value=setting.min_value, 

514 max_value=setting.max_value, 

515 step=setting.step, 

516 visible=setting.visible, 

517 editable=False 

518 if self.settings_locked 

519 else setting.editable, 

520 ) 

521 

522 # Override from the environment variables if needed. 

523 env_value = check_env_setting(setting.key) 

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

525 result[setting.key]["value"] = env_value 

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

527 # value have no effect as long as the environment 

528 # variable is set. 

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

530 except SQLAlchemyError as e: 

531 logger.exception( 

532 f"Error retrieving all settings from database: {e}" 

533 ) 

534 

535 return result 

536 

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

538 """ 

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

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

541 

542 Returns: 

543 Dictionary with setting keys mapped to their values 

544 """ 

545 all_settings = self.get_all_settings() 

546 settings_snapshot = {} 

547 

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

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

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

551 else: 

552 settings_snapshot[key] = setting 

553 

554 return settings_snapshot 

555 

556 def create_or_update_setting( 

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

558 ) -> Optional[Setting]: 

559 """ 

560 Create or update a setting 

561 

562 Args: 

563 setting: Setting object or dictionary 

564 commit: Whether to commit the change 

565 

566 Returns: 

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

568 """ 

569 if not self.db_session: 569 ↛ 570line 569 didn't jump to line 570 because the condition on line 569 was never true

570 logger.warning( 

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

572 ) 

573 return None 

574 if self.settings_locked: 574 ↛ 575line 574 didn't jump to line 575 because the condition on line 574 was never true

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

576 return None 

577 

578 # Convert dict to BaseSetting if needed 

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

580 # Determine type from key if not specified 

581 if "type" not in setting and "key" in setting: 581 ↛ 593line 581 didn't jump to line 593 because the condition on line 581 was always true

582 key = setting["key"] 

583 if key.startswith("llm."): 583 ↛ 584line 583 didn't jump to line 584 because the condition on line 583 was never true

584 setting_obj = LLMSetting(**setting) 

585 elif key.startswith("search."): 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true

586 setting_obj = SearchSetting(**setting) 

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

588 setting_obj = ReportSetting(**setting) 

589 else: 

590 setting_obj = AppSetting(**setting) 

591 else: 

592 # Use generic BaseSetting 

593 setting_obj = BaseSetting(**setting) 

594 else: 

595 setting_obj = setting 

596 

597 try: 

598 # Check if setting exists 

599 db_setting = ( 

600 self.db_session.query(Setting) 

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

602 .first() 

603 ) 

604 

605 if db_setting: 605 ↛ 607line 605 didn't jump to line 607 because the condition on line 605 was never true

606 # Update existing setting 

607 if setting.editable: 

608 logger.error( 

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

610 "is marked as non-editable.", 

611 setting["key"], 

612 ) 

613 return None 

614 

615 db_setting.value = setting_obj.value 

616 db_setting.name = setting_obj.name 

617 db_setting.description = setting_obj.description 

618 db_setting.category = setting_obj.category 

619 db_setting.ui_element = setting_obj.ui_element 

620 db_setting.options = setting_obj.options 

621 db_setting.min_value = setting_obj.min_value 

622 db_setting.max_value = setting_obj.max_value 

623 db_setting.step = setting_obj.step 

624 db_setting.visible = setting_obj.visible 

625 db_setting.editable = setting_obj.editable 

626 db_setting.updated_at = ( 

627 func.now() 

628 ) # Explicitly set the current timestamp 

629 else: 

630 # Create new setting 

631 db_setting = Setting( 

632 key=setting_obj.key, 

633 value=setting_obj.value, 

634 type=SettingType[setting_obj.type.upper()], 

635 name=setting_obj.name, 

636 description=setting_obj.description, 

637 category=setting_obj.category, 

638 ui_element=setting_obj.ui_element, 

639 options=setting_obj.options, 

640 min_value=setting_obj.min_value, 

641 max_value=setting_obj.max_value, 

642 step=setting_obj.step, 

643 visible=setting_obj.visible, 

644 editable=setting_obj.editable, 

645 ) 

646 self.db_session.add(db_setting) 

647 

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

649 self.db_session.commit() 

650 # Emit WebSocket event for settings change 

651 self._emit_settings_changed([setting_obj.key]) 

652 

653 return db_setting 

654 

655 except SQLAlchemyError as e: 

656 logger.exception( 

657 f"Error creating/updating setting {setting_obj.key}: {e}" 

658 ) 

659 self.db_session.rollback() 

660 return None 

661 

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

663 """ 

664 Delete a setting 

665 

666 Args: 

667 key: Setting key 

668 commit: Whether to commit the change 

669 

670 Returns: 

671 True if successful, False otherwise 

672 """ 

673 if not self.db_session: 673 ↛ 674line 673 didn't jump to line 674 because the condition on line 673 was never true

674 logger.warning( 

675 "No database session available, cannot delete setting" 

676 ) 

677 return False 

678 

679 try: 

680 # Remove from database 

681 result = ( 

682 self.db_session.query(Setting) 

683 .filter(Setting.key == key) 

684 .delete() 

685 ) 

686 

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

688 self.db_session.commit() 

689 

690 return result > 0 

691 except SQLAlchemyError: 

692 logger.exception("Error deleting setting") 

693 self.db_session.rollback() 

694 return False 

695 

696 def load_from_defaults_file( 

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

698 ) -> None: 

699 """ 

700 Import settings from the defaults settings file. 

701 

702 Args: 

703 commit: Whether to commit changes to database 

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

705 

706 """ 

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

708 

709 def db_version_matches_package(self) -> bool: 

710 """ 

711 Returns: 

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

713 

714 """ 

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

716 logger.debug( 

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

718 f"settings from version {package_version}." 

719 ) 

720 

721 return db_version == package_version 

722 

723 def update_db_version(self) -> None: 

724 """ 

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

726 

727 """ 

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

729 

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

731 version = Setting( 

732 key="app.version", 

733 value=package_version, 

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

735 editable=False, 

736 name="App Version", 

737 type=SettingType.APP, 

738 ui_element="text", 

739 visible=False, 

740 ) 

741 

742 self.db_session.add(version) 

743 self.db_session.commit() 

744 

745 def import_settings( 

746 self, 

747 settings_data: Dict[str, Any], 

748 commit: bool = True, 

749 overwrite: bool = True, 

750 delete_extra: bool = False, 

751 ) -> None: 

752 """ 

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

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

755 

756 Args: 

757 settings_data: The raw settings data to import. 

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

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

760 are already in the database. 

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

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

763 `settings_data`. 

764 

765 """ 

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

767 

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

769 if not overwrite: 

770 existing_value = self.get_setting(key) 

771 if existing_value is not None: 

772 # Preserve the value from this setting. 

773 setting_values["value"] = existing_value 

774 

775 # Delete any existing setting so we can completely overwrite it. 

776 self.delete_setting(key, commit=False) 

777 

778 setting = Setting(key=key, **setting_values) 

779 self.db_session.add(setting) 

780 

781 if commit or delete_extra: 

782 self.db_session.commit() 

783 logger.info(f"Successfully imported {len(settings_data)} settings") 

784 # Emit WebSocket event for all imported settings 

785 self._emit_settings_changed(list(settings_data.keys())) 

786 

787 if delete_extra: 

788 all_settings = self.get_all_settings() 

789 for key in all_settings: 

790 if key not in settings_data: 

791 logger.debug(f"Deleting extraneous setting: {key}") 

792 self.delete_setting(key, commit=False) 

793 

794 def _create_setting(self, key, value, setting_type): 

795 """Create a setting with appropriate metadata""" 

796 

797 # Determine appropriate category 

798 category = None 

799 ui_element = "text" 

800 

801 # Determine category based on key pattern 

802 if key.startswith("app."): 

803 category = "app_interface" 

804 elif key.startswith("llm."): 

805 if any( 

806 param in key 

807 for param in [ 

808 "temperature", 

809 "max_tokens", 

810 "n_batch", 

811 "n_gpu_layers", 

812 ] 

813 ): 

814 category = "llm_parameters" 

815 else: 

816 category = "llm_general" 

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

818 if any( 

819 param in key 

820 for param in ["iterations", "questions", "results", "region"] 

821 ): 

822 category = "search_parameters" 

823 else: 

824 category = "search_general" 

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

826 category = "report_parameters" 

827 

828 # Determine UI element type based on value 

829 if isinstance(value, bool): 

830 ui_element = "checkbox" 

831 elif isinstance(value, (int, float)) and not isinstance(value, bool): 

832 ui_element = "number" 

833 elif isinstance(value, (dict, list)): 

834 ui_element = "textarea" 

835 

836 # Build setting object 

837 setting_dict = { 

838 "key": key, 

839 "value": value, 

840 "type": setting_type.value.lower(), 

841 "name": key.split(".")[-1].replace("_", " ").title(), 

842 "description": f"Setting for {key}", 

843 "category": category, 

844 "ui_element": ui_element, 

845 } 

846 

847 # Create the setting in the database 

848 self.create_or_update_setting(setting_dict, commit=False) 

849 

850 def _emit_settings_changed(self, changed_keys: list = None): 

851 """ 

852 Emit WebSocket event when settings change 

853 

854 Args: 

855 changed_keys: List of setting keys that changed 

856 """ 

857 try: 

858 # Import here to avoid circular imports 

859 from ..web.services.socket_service import SocketIOService 

860 

861 try: 

862 socket_service = SocketIOService() 

863 except ValueError: 

864 logger.debug( 

865 "Not emitting socket event because server is not initialized." 

866 ) 

867 return 

868 

869 # Get the changed settings 

870 settings_data = {} 

871 if changed_keys: 871 ↛ 878line 871 didn't jump to line 878 because the condition on line 871 was always true

872 for key in changed_keys: 

873 setting_value = self.get_setting(key) 

874 if setting_value is not None: 874 ↛ 872line 874 didn't jump to line 872 because the condition on line 874 was always true

875 settings_data[key] = {"value": setting_value} 

876 

877 # Emit the settings change event 

878 from datetime import datetime, UTC 

879 

880 socket_service.emit_socket_event( 

881 "settings_changed", 

882 { 

883 "changed_keys": changed_keys or [], 

884 "settings": settings_data, 

885 "timestamp": datetime.now(UTC).isoformat(), 

886 }, 

887 ) 

888 

889 logger.debug( 

890 f"Emitted settings_changed event for keys: {changed_keys}" 

891 ) 

892 

893 except Exception: 

894 logger.exception("Failed to emit settings change event") 

895 # Don't let WebSocket emission failures break settings saving 

896 

897 @staticmethod 

898 def get_bootstrap_env_vars() -> Dict[str, str]: 

899 """ 

900 Get environment variables that must be available before database access. 

901 These are critical for system initialization. 

902 

903 Returns: 

904 Dict mapping env var names to their descriptions 

905 """ 

906 # Get bootstrap vars from env registry 

907 return env_registry.get_bootstrap_vars() 

908 

909 @staticmethod 

910 def is_bootstrap_env_var(env_var: str) -> bool: 

911 """ 

912 Check if an environment variable is a bootstrap variable (needed before DB access). 

913 

914 Args: 

915 env_var: Environment variable name 

916 

917 Returns: 

918 True if this is a bootstrap variable 

919 """ 

920 bootstrap_vars = SettingsManager.get_bootstrap_env_vars() 

921 return env_var in bootstrap_vars 

922 

923 @staticmethod 

924 def is_env_only_setting(key: str) -> bool: 

925 """ 

926 Check if a setting key is environment-only. 

927 

928 Args: 

929 key: Setting key to check 

930 

931 Returns: 

932 True if it's an env-only setting, False otherwise 

933 """ 

934 return env_registry.is_env_only(key) 

935 

936 @staticmethod 

937 def get_env_var_for_setting(setting_key: str) -> str: 

938 """ 

939 Get the environment variable name for a given setting key. 

940 

941 Args: 

942 setting_key: Setting key (e.g., "app.host")