Coverage for src/local_deep_research/api/settings_utils.py: 95%
182 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Utilities for managing settings in the programmatic API.
4This module provides functions to create settings snapshots for the API
5without requiring database access, reusing the same mechanisms as the
6web interface.
7"""
9import copy
10from typing import Any
11from loguru import logger
13from ..settings import SettingsManager
14from ..settings.base import ISettingsManager
15from ..settings.manager import (
16 UI_ELEMENT_TO_SETTING_TYPE,
17 _validate_imported_setting_value,
18 check_env_setting,
19)
20from ..utilities.type_utils import to_bool, unwrap_setting
23class InMemorySettingsManager(ISettingsManager):
24 """
25 In-memory settings manager that doesn't require database access.
27 This is used for the programmatic API to provide settings without
28 needing a database connection.
29 """
31 def __init__(self):
32 """Initialize with default settings from JSON file."""
33 # Create a base manager to get default settings
34 self._base_manager = SettingsManager(db_session=None)
35 self._settings = {}
36 self._load_defaults()
38 def _get_typed_value(self, setting_data: dict[str, Any], value: Any) -> Any:
39 """
40 Convert a value to the appropriate type based on the setting's ui_element.
42 Args:
43 setting_data: The setting metadata containing ui_element
44 value: The value to convert
46 Returns:
47 The typed value, or the original value if conversion fails
48 """
49 if value is None:
50 return None
52 ui_element = setting_data.get("ui_element", "text")
53 setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element)
55 if setting_type is None:
56 logger.warning(
57 f"Unknown ui_element type: {ui_element}, returning value as-is"
58 )
59 return value
61 try:
62 return setting_type(value)
63 except (ValueError, TypeError):
64 logger.warning(
65 f"Failed to convert value {value} to type {setting_type}"
66 )
67 return value
69 def _load_defaults(self):
70 """Load default settings from the JSON file."""
71 # Get default settings from the base manager
72 defaults = self._base_manager.default_settings
74 # Convert to the format expected by get_all_settings
75 for key, setting_data in defaults.items():
76 self._settings[key] = setting_data.copy()
78 # Load search engine configurations from individual JSON files
79 from importlib import resources
80 import json
82 try:
83 # Load search engines from defaults/settings/search_engines/
84 search_engines_dir = resources.files(
85 "local_deep_research.defaults.settings"
86 ).joinpath("search_engines")
88 if search_engines_dir.exists() and search_engines_dir.is_dir(): 88 ↛ exitline 88 didn't return from function '_load_defaults' because the condition on line 88 was always true
89 for json_file in search_engines_dir.glob("*.json"):
90 try:
91 engine_settings = json.loads(
92 json_file.read_text(encoding="utf-8-sig")
93 )
94 # Merge into main settings
95 for key, setting_data in engine_settings.items():
96 if key not in self._settings:
97 self._settings[key] = setting_data.copy()
98 except Exception:
99 logger.warning(
100 f"Failed to load search engine config from {json_file.name}"
101 )
102 except Exception:
103 logger.warning("Failed to load search engine configs")
105 def get_setting(
106 self, key: str, default: Any = None, check_env: bool = True
107 ) -> Any:
108 """Get a setting value."""
109 if key in self._settings:
110 setting_data = self._settings[key]
111 if check_env:
112 env_value = check_env_setting(key)
113 if env_value is not None:
114 return self._get_typed_value(setting_data, env_value)
115 value = setting_data.get("value", default)
116 # Ensure the value has the correct type
117 return self._get_typed_value(setting_data, value)
118 return default
120 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
121 """Set a setting value (in memory only)."""
122 if key in self._settings:
123 # Validate and convert the value to the correct type
124 typed_value = self._get_typed_value(self._settings[key], value)
125 self._settings[key]["value"] = typed_value
126 return True
127 return False
129 def get_all_settings(
130 self,
131 bypass_cache: bool = False,
132 include_environment_overrides: bool = True,
133 strict: bool = False,
134 ) -> dict[str, Any]:
135 """Get all settings with metadata."""
136 result = copy.deepcopy(self._settings)
137 if include_environment_overrides:
138 for key, setting_data in result.items():
139 if (
140 not isinstance(setting_data, dict)
141 or "value" not in setting_data
142 ):
143 continue
144 env_value = check_env_setting(key)
145 if env_value is not None:
146 setting_data["value"] = self._get_typed_value(
147 setting_data, env_value
148 )
149 setting_data["editable"] = False
150 return result
152 def load_from_defaults_file(
153 self, commit: bool = True, **kwargs: Any
154 ) -> None:
155 """Reload defaults while honoring environment-lock preservation."""
156 preserve_locked = bool(kwargs.get("preserve_environment_locked", False))
157 preserved_values = (
158 {
159 key: copy.deepcopy(value["value"])
160 for key, value in self._settings.items()
161 if isinstance(value, dict)
162 and "value" in value
163 and check_env_setting(key) is not None
164 }
165 if preserve_locked
166 else {}
167 )
168 self._settings.clear()
169 self._load_defaults()
170 for key, value in preserved_values.items():
171 if key in self._settings:
172 self._settings[key]["value"] = value
174 def create_or_update_setting(
175 self, setting: dict[str, Any] | Any, commit: bool = True
176 ) -> Any | None:
177 """Create or update a setting (in memory only)."""
178 if isinstance(setting, dict) and "key" in setting:
179 key = setting["key"]
180 # If the setting has a value, ensure it has the correct type
181 if "value" in setting:
182 typed_value = self._get_typed_value(setting, setting["value"])
183 setting = setting.copy() # Don't modify the original
184 setting["value"] = typed_value
185 self._settings[key] = setting
186 return setting
187 return None
189 def delete_setting(self, key: str, commit: bool = True) -> bool:
190 """Delete a setting (in memory only)."""
191 if key in self._settings:
192 del self._settings[key]
193 return True
194 return False
196 def get_bool_setting(
197 self, key: str, default: bool = False, check_env: bool = True
198 ) -> bool:
199 """Get a setting value as a boolean."""
200 value = self.get_setting(key, default, check_env)
201 return to_bool(value, default)
203 def get_settings_snapshot(self, strict: bool = False) -> dict[str, Any]:
204 """Get a simplified settings snapshot with just key-value pairs."""
205 all_settings = self.get_all_settings(strict=strict)
206 snapshot = {}
207 for key, setting in all_settings.items():
208 if isinstance(setting, dict) and "value" in setting:
209 snapshot[key] = setting["value"]
210 else:
211 snapshot[key] = setting
212 return snapshot
214 def import_settings(
215 self,
216 settings_data: dict[str, Any],
217 commit: bool = True,
218 overwrite: bool = True,
219 delete_extra: bool = False,
220 preserve_environment_locked: bool = False,
221 ) -> None:
222 """Import settings from a dictionary."""
223 # Schema-aware import (#5589): validate values that come from the
224 # imported file against the CURRENT defaults schema, so a
225 # pre-upgrade export cannot resurrect values that are invalid under
226 # the current options/constraints. Values retained from existing
227 # in-memory state (the `overwrite=False` path and environment-locked
228 # values under `preserve_environment_locked`) are trusted, not
229 # untrusted file input, and are deliberately not validated.
230 defaults_for_import = self._base_manager.default_settings
231 # Under `delete_extra=True`, entries cleared below must still be
232 # restorable for keys whose imported value is rejected by
233 # validation — mirroring the DB manager, where a rejected key keeps
234 # its existing row (the key is retained before validation runs).
235 prior_settings = (
236 {key: copy.deepcopy(value) for key, value in self._settings.items()}
237 if delete_extra
238 else None
239 )
240 if delete_extra:
241 preserved = (
242 {
243 key: copy.deepcopy(value)
244 for key, value in self._settings.items()
245 if check_env_setting(key) is not None
246 }
247 if preserve_environment_locked
248 else {}
249 )
250 self._settings.clear()
251 self._settings.update(preserved)
253 for key, value in settings_data.items():
254 existing_setting = self._settings.get(key)
255 preserve_value = (
256 preserve_environment_locked
257 and check_env_setting(key) is not None
258 and isinstance(existing_setting, dict)
259 and "value" in existing_setting
260 )
261 if preserve_value and not isinstance(value, dict):
262 # A bare-scalar import carries no metadata to merge; keep
263 # the stored env-locked entry untouched.
264 continue
265 if overwrite or key not in self._settings or preserve_value:
266 # Ensure proper type handling for imported settings
267 setting_values = (
268 value.copy() if isinstance(value, dict) else value
269 )
270 if isinstance(value, dict) and "value" in value:
271 default_meta = defaults_for_import.get(key)
272 if (
273 default_meta is not None
274 and not preserve_value
275 and _validate_imported_setting_value(
276 key, value["value"], default_meta
277 )
278 is not None
279 ):
280 # The file-supplied value is invalid under the
281 # current defaults schema (#5589); skip the entry
282 # and keep any prior entry (mirrors the DB manager,
283 # which leaves the existing row in place).
284 logger.warning(
285 "Skipping import of setting {!r}: value is "
286 "invalid under the current defaults schema",
287 key,
288 )
289 if prior_settings is not None and key in prior_settings:
290 self._settings[key] = prior_settings[key]
291 continue
292 typed_value = self._get_typed_value(value, value["value"])
293 setting_values["value"] = typed_value
294 if preserve_value:
295 setting_values["value"] = copy.deepcopy(
296 existing_setting["value"]
297 )
298 self._settings[key] = setting_values
301def get_default_settings_snapshot() -> dict[str, Any]:
302 """
303 Get a complete settings snapshot with default values.
305 This uses the same mechanism as the web interface but without
306 requiring database access. Environment variables are checked
307 for overrides.
309 Returns:
310 Dict mapping setting keys to their values and metadata
311 """
312 manager = InMemorySettingsManager()
313 return manager.get_all_settings()
316def create_settings_snapshot(
317 overrides: dict[str, Any] | None = None,
318 base_settings: dict[str, Any] | None = None,
319 **kwargs,
320) -> dict[str, Any]:
321 """
322 Create a settings snapshot for the programmatic API.
324 Args:
325 overrides: Dict of setting overrides (e.g., {"llm.provider": "openai"})
326 This is the most common use case - pass a dict of settings to override.
327 base_settings: Base settings dict (defaults to get_default_settings_snapshot())
328 Rarely needed - only for advanced use cases.
329 **kwargs: Common setting shortcuts:
330 - provider: Maps to "llm.provider"
331 - api_key: Maps to "llm.{provider}.api_key"
332 - temperature: Maps to "llm.temperature"
333 - max_search_results: Maps to "search.max_results"
334 - search_engines: Maps to enabled search engines
336 Returns:
337 Complete settings snapshot for use with the API
339 Examples:
340 # Most common - pass overrides as first argument
341 settings = create_settings_snapshot({"search.tool": "wikipedia"})
343 # Or use named parameter
344 settings = create_settings_snapshot(overrides={"llm.provider": "openai"})
346 # Use kwargs shortcuts
347 settings = create_settings_snapshot(provider="openai", temperature=0.7)
349 # Advanced - provide custom base settings
350 settings = create_settings_snapshot(
351 overrides={"search.tool": "wikipedia"},
352 base_settings=my_custom_defaults
353 )
354 """
355 # Start with base settings or defaults
356 if base_settings is None:
357 settings = get_default_settings_snapshot()
358 else:
359 settings = copy.deepcopy(base_settings)
361 # Apply overrides if provided
362 if overrides:
363 for key, value in overrides.items():
364 if key in settings:
365 if isinstance(settings[key], dict) and "value" in settings[key]: 365 ↛ 368line 365 didn't jump to line 368 because the condition on line 365 was always true
366 settings[key]["value"] = value
367 else:
368 settings[key] = value
369 else:
370 # Create a simple setting entry for unknown keys
371 # Infer ui_element from value type
372 ui_element = "text" # default
373 if isinstance(value, bool):
374 ui_element = "checkbox"
375 elif isinstance(value, (int, float)):
376 ui_element = "number"
377 elif isinstance(value, dict):
378 ui_element = "json"
380 settings[key] = {"value": value, "ui_element": ui_element}
382 # Handle common kwargs shortcuts
383 if "provider" in kwargs:
384 provider = kwargs["provider"]
385 if "llm.provider" in settings: 385 ↛ 388line 385 didn't jump to line 388 because the condition on line 385 was always true
386 settings["llm.provider"]["value"] = provider
387 else:
388 settings["llm.provider"] = {"value": provider}
390 # Handle api_key if provided
391 if "api_key" in kwargs:
392 api_key = kwargs["api_key"]
393 api_key_setting = f"llm.{provider}.api_key"
394 if api_key_setting in settings:
395 settings[api_key_setting]["value"] = api_key
396 else:
397 settings[api_key_setting] = {"value": api_key}
399 if "temperature" in kwargs:
400 if "llm.temperature" in settings: 400 ↛ 403line 400 didn't jump to line 403 because the condition on line 400 was always true
401 settings["llm.temperature"]["value"] = kwargs["temperature"]
402 else:
403 settings["llm.temperature"] = {"value": kwargs["temperature"]}
405 if "max_search_results" in kwargs:
406 if "search.max_results" in settings: 406 ↛ 411line 406 didn't jump to line 411 because the condition on line 406 was always true
407 settings["search.max_results"]["value"] = kwargs[
408 "max_search_results"
409 ]
410 else:
411 settings["search.max_results"] = {
412 "value": kwargs["max_search_results"]
413 }
415 # Add any other common shortcuts here...
417 return settings
420def extract_setting_value(
421 settings_snapshot: dict[str, Any], key: str, default: Any = None
422) -> Any:
423 """
424 Extract a setting value from a settings snapshot.
426 Args:
427 settings_snapshot: Settings snapshot dict
428 key: Setting key (e.g., "llm.provider")
429 default: Default value if not found
431 Returns:
432 The setting value
433 """
434 if settings_snapshot is None:
435 return default
436 if key in settings_snapshot:
437 setting = settings_snapshot[key]
438 return unwrap_setting(setting)
439 return default
442def extract_bool_setting(
443 settings_snapshot: dict[str, Any], key: str, default: bool = False
444) -> bool:
445 """
446 Extract a boolean setting value from a settings snapshot.
448 This is a convenience wrapper around extract_setting_value that
449 handles string-to-boolean conversion.
451 Args:
452 settings_snapshot: Settings snapshot dict
453 key: Setting key (e.g., "local_search_normalize_vectors")
454 default: Default boolean value if not found
456 Returns:
457 Boolean value of the setting
458 """
459 value = extract_setting_value(settings_snapshot, key, default)
460 return to_bool(value, default)