Coverage for src/local_deep_research/api/settings_utils.py: 95%
189 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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.env_registry import registry as env_registry
16from ..settings.exceptions import SettingsValueError
17from ..settings.manager import (
18 UI_ELEMENT_TO_SETTING_TYPE as UI_ELEMENT_TO_SETTING_TYPE,
19 validate_imported_setting_value,
20 check_env_setting,
21 get_typed_setting_value,
22 is_operator_only_setting,
23 is_valid_setting_key,
24)
25from ..utilities.type_utils import to_bool, unwrap_setting
28class InMemorySettingsManager(ISettingsManager):
29 """
30 In-memory settings manager that doesn't require database access.
32 This is used for the programmatic API to provide settings without
33 needing a database connection.
34 """
36 def __init__(self):
37 """Initialize with default settings from JSON file."""
38 # Create a base manager to get default settings
39 self._base_manager = SettingsManager(db_session=None)
40 self._settings = {}
41 self._load_defaults()
43 def _get_typed_value(
44 self,
45 setting_data: dict[str, Any],
46 value: Any,
47 *,
48 key: str | None = None,
49 ) -> Any:
50 """Validate a stored value without applying environment overrides."""
51 if not isinstance(setting_data, dict): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 raise SettingsValueError(
53 key or "<in-memory>",
54 source="value",
55 reason="invalid setting metadata",
56 )
57 return get_typed_setting_value(
58 key if key is not None else setting_data.get("key", "<in-memory>"),
59 value,
60 setting_data.get("ui_element", "text"),
61 check_env=False,
62 )
64 def _load_defaults(self):
65 """Load default settings from the JSON file."""
66 # Get default settings from the base manager
67 defaults = self._base_manager.default_settings
69 # Convert to the format expected by get_all_settings
70 for key, setting_data in defaults.items():
71 self._settings[key] = setting_data.copy()
73 # Load search engine configurations from individual JSON files
74 from importlib import resources
75 import json
77 try:
78 # Load search engines from defaults/settings/search_engines/
79 search_engines_dir = resources.files(
80 "local_deep_research.defaults.settings"
81 ).joinpath("search_engines")
83 if search_engines_dir.exists() and search_engines_dir.is_dir(): 83 ↛ exitline 83 didn't return from function '_load_defaults' because the condition on line 83 was always true
84 for json_file in search_engines_dir.glob("*.json"):
85 try:
86 engine_settings = json.loads(
87 json_file.read_text(encoding="utf-8-sig")
88 )
89 # Merge into main settings
90 for key, setting_data in engine_settings.items():
91 if key not in self._settings:
92 self._settings[key] = setting_data.copy()
93 except Exception:
94 logger.warning(
95 f"Failed to load search engine config from {json_file.name}"
96 )
97 except Exception:
98 logger.warning("Failed to load search engine configs")
100 def _read_setting_value(
101 self,
102 key: str,
103 setting_data: dict[str, Any],
104 default: Any = None,
105 check_env: bool = True,
106 ) -> Any:
107 """Read the selected value or report invalid configuration."""
108 if not isinstance(setting_data, dict):
109 raise SettingsValueError(
110 key, source="value", reason="invalid setting metadata"
111 )
112 ui_element = setting_data.get("ui_element", "text")
113 return get_typed_setting_value(
114 key,
115 setting_data.get("value"),
116 ui_element,
117 default=default,
118 check_env=check_env,
119 )
121 def get_setting(
122 self, key: str, default: Any = None, check_env: bool = True
123 ) -> Any:
124 """Read a typed value, or a namespace when no exact entry exists.
126 An environment override for a key with no entry and no declared
127 type is returned as the raw string, as ``SettingsManager`` does.
128 """
129 if not isinstance(key, str) or not key:
130 return default
132 if env_registry.is_env_only(key):
133 return env_registry.get_strict(key, default, check_env=check_env)
135 if key in self._settings:
136 setting_data = self._settings[key]
137 return self._read_setting_value(
138 key, setting_data, default=default, check_env=check_env
139 )
141 # Preserve exact-entry precedence and the DB reader's scalar-only
142 # treatment of operator-owned paths. Malformed legacy children do
143 # not turn a missing leaf into a namespace.
144 if not is_operator_only_setting(key):
145 children = {
146 child.removeprefix(f"{key}."): self._read_setting_value(
147 child, data, default=default, check_env=check_env
148 )
149 for child, data in self._settings.items()
150 if is_valid_setting_key(child) and child.startswith(f"{key}.")
151 }
152 if children:
153 return children
155 if check_env:
156 env_value = check_env_setting(key)
157 if env_value is not None:
158 metadata = self._base_manager.default_settings.get(key)
159 if isinstance(metadata, dict):
160 return get_typed_setting_value(
161 key,
162 None,
163 metadata.get("ui_element", "text"),
164 default=default,
165 )
166 # Same as SettingsManager.get_setting: with no declared type
167 # there is nothing to validate against, so the override is
168 # returned unconverted. Typed keys still raise above.
169 logger.warning(
170 "Setting '{}' has env var override but is not in "
171 "defaults — returning raw string without type "
172 "conversion. Add this setting to a defaults JSON "
173 "file with a ui_element type to enable proper "
174 "type conversion.",
175 key,
176 )
177 return env_value
178 return default
180 def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
181 """Set a setting value (in memory only)."""
182 if key in self._settings:
183 # Validate and convert the value to the correct type
184 typed_value = self._get_typed_value(
185 self._settings[key], value, key=key
186 )
187 self._settings[key]["value"] = typed_value
188 return True
189 return False
191 def get_all_settings(
192 self,
193 bypass_cache: bool = False,
194 include_environment_overrides: bool = True,
195 strict: bool = False,
196 ) -> dict[str, Any]:
197 """Validate effective settings, or return raw storage for repair.
199 Effective reads validate stored values without changing their saved
200 representation. Only active environment overrides replace values.
201 Raw exports skip validation and preserve the stored metadata.
202 """
203 result = copy.deepcopy(self._settings)
204 if include_environment_overrides:
205 for key, setting_data in result.items():
206 value = self._read_setting_value(key, setting_data)
207 if check_env_setting(key) is not None:
208 setting_data["value"] = value
209 setting_data["editable"] = False
210 return result
212 def load_from_defaults_file(
213 self,
214 commit: bool = True,
215 preserve_environment_locked: bool = False,
216 ) -> None:
217 """Reload defaults while honoring environment-lock preservation.
219 No ``**kwargs`` passthrough (see the ``ISettingsManager`` contract,
220 #5841): the in-memory store has no settings lock, so the
221 ``override_locked`` bypass is meaningless here — and now also
222 structurally rejected.
223 """
224 preserved_values = (
225 {
226 key: copy.deepcopy(value["value"])
227 for key, value in self._settings.items()
228 if isinstance(value, dict)
229 and "value" in value
230 and check_env_setting(key) is not None
231 }
232 if preserve_environment_locked
233 else {}
234 )
235 self._settings.clear()
236 self._load_defaults()
237 for key, value in preserved_values.items():
238 if key in self._settings:
239 self._settings[key]["value"] = value
241 def create_or_update_setting(
242 self, setting: dict[str, Any] | Any, commit: bool = True
243 ) -> Any | None:
244 """Create or update a setting (in memory only)."""
245 if isinstance(setting, dict) and "key" in setting:
246 key = setting["key"]
247 # Validate metadata even when an optional value is absent.
248 typed_value = self._get_typed_value(setting, setting.get("value"))
249 if "value" in setting:
250 setting = setting.copy() # Don't modify the original
251 setting["value"] = typed_value
252 self._settings[key] = setting
253 return setting
254 return None
256 def delete_setting(
257 self, key: str, commit: bool = True, override_locked: bool = False
258 ) -> bool:
259 """Delete a setting (in memory only).
261 ``override_locked`` is accepted to satisfy the ``ISettingsManager``
262 contract (#5738) and otherwise ignored: the in-memory store has no
263 settings lock, so there is nothing for it to override.
264 """
265 if key in self._settings:
266 del self._settings[key]
267 return True
268 return False
270 def get_bool_setting(
271 self, key: str, default: bool = False, check_env: bool = True
272 ) -> bool:
273 """Get a setting value as a boolean."""
274 value = self.get_setting(key, default, check_env)
275 return to_bool(value, default)
277 def get_settings_snapshot(self, strict: bool = False) -> dict[str, Any]:
278 """Get a simplified settings snapshot with just key-value pairs."""
279 all_settings = self.get_all_settings(strict=strict)
280 snapshot = {}
281 for key, setting in all_settings.items():
282 if isinstance(setting, dict) and "value" in setting:
283 snapshot[key] = setting["value"]
284 else:
285 snapshot[key] = setting
286 return snapshot
288 def import_settings(
289 self,
290 settings_data: dict[str, Any],
291 commit: bool = True,
292 overwrite: bool = True,
293 delete_extra: bool = False,
294 preserve_environment_locked: bool = False,
295 override_locked: bool = False,
296 ) -> None:
297 """Import settings from a dictionary.
299 Conversion failures leave the live settings unchanged. Values that
300 fail schema validation retain the existing skip-and-preserve policy.
302 ``override_locked`` is accepted to satisfy the ``ISettingsManager``
303 contract (#5738) and otherwise ignored: the in-memory store has no
304 settings lock, so there is nothing for it to override.
305 """
306 # Schema-aware import (#5589) — validates untrusted file values
307 # against the current schema; see validate_imported_setting_value's
308 # docstring for the trusted-vs-untrusted rationale (values retained
309 # from existing in-memory state are trusted here, not validated).
310 defaults_for_import = self._base_manager.default_settings
311 # Rows are replaced, not mutated, during import. Stage those
312 # replacements so a later conversion error cannot expose a partial
313 # import or leave live settings cleared by delete_extra.
314 pending_settings = self._settings.copy()
315 # Under `delete_extra=True`, entries cleared below must still be
316 # restorable for keys whose imported value is rejected by
317 # validation — mirroring the DB manager, where a rejected key keeps
318 # its existing row (the key is retained before validation runs).
319 prior_settings = (
320 {key: copy.deepcopy(value) for key, value in self._settings.items()}
321 if delete_extra
322 else None
323 )
324 if delete_extra:
325 preserved = (
326 {
327 key: copy.deepcopy(value)
328 for key, value in self._settings.items()
329 if check_env_setting(key) is not None
330 }
331 if preserve_environment_locked
332 else {}
333 )
334 pending_settings = preserved
336 for key, value in settings_data.items():
337 existing_setting = pending_settings.get(key)
338 preserve_value = (
339 preserve_environment_locked
340 and check_env_setting(key) is not None
341 and isinstance(existing_setting, dict)
342 and "value" in existing_setting
343 )
344 if preserve_value and not isinstance(value, dict):
345 # A bare-scalar import carries no metadata to merge; keep
346 # the stored env-locked entry untouched.
347 continue
348 if overwrite or key not in pending_settings or preserve_value:
349 # Ensure proper type handling for imported settings
350 setting_values = (
351 value.copy() if isinstance(value, dict) else value
352 )
353 if (
354 isinstance(value, dict)
355 and "value" in value
356 and not preserve_value
357 ):
358 default_meta = defaults_for_import.get(key)
359 if (
360 default_meta is not None
361 and not preserve_value
362 and validate_imported_setting_value(
363 key, value["value"], default_meta
364 )
365 is not None
366 ):
367 # The file-supplied value is invalid under the
368 # current defaults schema (#5589); skip the entry
369 # and keep any prior entry (mirrors the DB manager,
370 # which leaves the existing row in place).
371 logger.warning(
372 "Skipping import of setting {!r}: value is "
373 "invalid under the current defaults schema",
374 key,
375 )
376 if prior_settings is not None and key in prior_settings:
377 pending_settings[key] = prior_settings[key]
378 continue
379 typed_value = self._get_typed_value(
380 value, value["value"], key=key
381 )
382 setting_values["value"] = typed_value
383 if preserve_value:
384 setting_values["value"] = copy.deepcopy(
385 existing_setting["value"]
386 )
387 pending_settings[key] = setting_values
389 self._settings = pending_settings
392def get_default_settings_snapshot() -> dict[str, Any]:
393 """
394 Get a complete settings snapshot with default values.
396 This uses the same mechanism as the web interface but without
397 requiring database access. Environment variables are checked
398 for overrides.
400 Returns:
401 Dict mapping setting keys to their values and metadata
402 """
403 manager = InMemorySettingsManager()
404 return manager.get_all_settings()
407def create_settings_snapshot(
408 overrides: dict[str, Any] | None = None,
409 base_settings: dict[str, Any] | None = None,
410 **kwargs,
411) -> dict[str, Any]:
412 """
413 Create a settings snapshot for the programmatic API.
415 Args:
416 overrides: Dict of setting overrides (e.g., {"llm.provider": "openai"})
417 This is the most common use case - pass a dict of settings to override.
418 base_settings: Base settings dict (defaults to get_default_settings_snapshot())
419 Rarely needed - only for advanced use cases.
420 **kwargs: Common setting shortcuts:
421 - provider: Maps to "llm.provider"
422 - api_key: Maps to "llm.{provider}.api_key"
423 - temperature: Maps to "llm.temperature"
424 - max_search_results: Maps to "search.max_results"
425 - search_engines: Maps to enabled search engines
427 Returns:
428 Complete settings snapshot for use with the API
430 Examples:
431 # Most common - pass overrides as first argument
432 settings = create_settings_snapshot({"search.tool": "wikipedia"})
434 # Or use named parameter
435 settings = create_settings_snapshot(overrides={"llm.provider": "openai"})
437 # Use kwargs shortcuts
438 settings = create_settings_snapshot(provider="openai", temperature=0.7)
440 # Advanced - provide custom base settings
441 settings = create_settings_snapshot(
442 overrides={"search.tool": "wikipedia"},
443 base_settings=my_custom_defaults
444 )
445 """
446 # Start with base settings or defaults
447 if base_settings is None:
448 settings = get_default_settings_snapshot()
449 else:
450 settings = copy.deepcopy(base_settings)
452 # Apply overrides if provided
453 if overrides:
454 for key, value in overrides.items():
455 if key in settings:
456 if isinstance(settings[key], dict) and "value" in settings[key]: 456 ↛ 459line 456 didn't jump to line 459 because the condition on line 456 was always true
457 settings[key]["value"] = value
458 else:
459 settings[key] = value
460 else:
461 # Create a simple setting entry for unknown keys
462 # Infer ui_element from value type
463 ui_element = "text" # default
464 if isinstance(value, bool):
465 ui_element = "checkbox"
466 elif isinstance(value, (int, float)):
467 ui_element = "number"
468 elif isinstance(value, dict):
469 ui_element = "json"
471 settings[key] = {"value": value, "ui_element": ui_element}
473 # Handle common kwargs shortcuts
474 if "provider" in kwargs:
475 provider = kwargs["provider"]
476 if "llm.provider" in settings: 476 ↛ 479line 476 didn't jump to line 479 because the condition on line 476 was always true
477 settings["llm.provider"]["value"] = provider
478 else:
479 settings["llm.provider"] = {"value": provider}
481 # Handle api_key if provided
482 if "api_key" in kwargs:
483 api_key = kwargs["api_key"]
484 api_key_setting = f"llm.{provider}.api_key"
485 if api_key_setting in settings:
486 settings[api_key_setting]["value"] = api_key
487 else:
488 settings[api_key_setting] = {"value": api_key}
490 if "temperature" in kwargs:
491 if "llm.temperature" in settings: 491 ↛ 494line 491 didn't jump to line 494 because the condition on line 491 was always true
492 settings["llm.temperature"]["value"] = kwargs["temperature"]
493 else:
494 settings["llm.temperature"] = {"value": kwargs["temperature"]}
496 if "max_search_results" in kwargs:
497 if "search.max_results" in settings: 497 ↛ 502line 497 didn't jump to line 502 because the condition on line 497 was always true
498 settings["search.max_results"]["value"] = kwargs[
499 "max_search_results"
500 ]
501 else:
502 settings["search.max_results"] = {
503 "value": kwargs["max_search_results"]
504 }
506 # Add any other common shortcuts here...
508 return settings
511def extract_setting_value(
512 settings_snapshot: dict[str, Any], key: str, default: Any = None
513) -> Any:
514 """
515 Extract a setting value from a settings snapshot.
517 Args:
518 settings_snapshot: Settings snapshot dict
519 key: Setting key (e.g., "llm.provider")
520 default: Default value if not found
522 Returns:
523 The setting value
524 """
525 if settings_snapshot is None:
526 return default
527 if key in settings_snapshot:
528 setting = settings_snapshot[key]
529 return unwrap_setting(setting)
530 return default
533def extract_bool_setting(
534 settings_snapshot: dict[str, Any], key: str, default: bool = False
535) -> bool:
536 """
537 Extract a boolean setting value from a settings snapshot.
539 This is a convenience wrapper around extract_setting_value that
540 handles string-to-boolean conversion.
542 Args:
543 settings_snapshot: Settings snapshot dict
544 key: Setting key (e.g., "local_search_normalize_vectors")
545 default: Default boolean value if not found
547 Returns:
548 Boolean value of the setting
549 """
550 value = extract_setting_value(settings_snapshot, key, default)
551 return to_bool(value, default)