Coverage for src/local_deep_research/web/services/settings_service.py: 94%
84 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import math
2from typing import Any, Dict, Optional, Union
4from loguru import logger
5from sqlalchemy.orm import Session
7from ...database.models import Setting
8from ...settings.exceptions import SettingsValueError
9from ...settings.manager import DYNAMIC_SETTINGS, get_typed_setting_value
10from ...utilities.db_utils import get_settings_manager
11from ...utilities.type_utils import to_positive_int
13# Settings with dynamically populated options (excluded from validation).
14# Single source of truth lives in `settings/manager.py` (it is also used by
15# `import_settings` there); re-exported here for the web save path. The
16# re-export must stay identity-equal to the manager's list — pinned by
17# tests/settings/test_import_settings_validation.py.
20def set_setting(
21 key: str,
22 value: Any,
23 commit: bool = True,
24 db_session: Optional[Session] = None,
25) -> bool:
26 """
27 Set a setting value
29 Args:
30 key: Setting key
31 value: Setting value
32 commit: Whether to commit the change
33 db_session: Optional database session
35 Returns:
36 bool: True if successful
37 """
38 manager = get_settings_manager(db_session)
39 return bool(manager.set_setting(key, value, commit))
42def get_all_settings() -> Dict[str, Any]:
43 """
44 Get all settings, optionally filtered by type
46 Returns:
47 Dict[str, Any]: Dictionary of settings
49 """
50 manager = get_settings_manager()
51 return manager.get_all_settings() # type: ignore[no-any-return]
54def create_or_update_setting(
55 setting: Union[Dict[str, Any], Setting],
56 commit: bool = True,
57 db_session: Optional[Session] = None,
58) -> Optional[Setting]:
59 """
60 Create or update a setting
62 Args:
63 setting: Setting dictionary or object
64 commit: Whether to commit the change
65 db_session: Optional database session
67 Returns:
68 Optional[Setting]: The setting object if successful
69 """
70 manager = get_settings_manager(db_session)
71 return manager.create_or_update_setting(setting, commit) # type: ignore[no-any-return]
74def invalidate_settings_caches(username=None):
75 """Invalidate all settings-related caches after a settings mutation.
77 Call this after any route that creates, updates, or deletes settings
78 so background services pick up changes immediately.
80 Safe to call from anywhere — silently skips caches that aren't initialized.
82 Args:
83 username: If provided, invalidates only that user's scheduler cache.
84 If None, invalidates all users' scheduler caches.
85 """
86 # News scheduler per-user settings cache (TTLCache, 5-min TTL).
87 # NOTE: the LLM provider dropdown is served by the auto-discovery path
88 # (llm/providers/auto_discovery.get_discovered_provider_options), which
89 # rebuilds per call and has no cache to invalidate here.
90 try:
91 from ...scheduler.background import get_background_job_scheduler
93 scheduler = get_background_job_scheduler()
94 if username is not None:
95 scheduler.invalidate_user_settings_cache(username)
96 else:
97 scheduler.invalidate_all_settings_cache()
98 except Exception:
99 logger.debug("Could not invalidate scheduler cache", exc_info=True)
102def reschedule_document_jobs_if_needed(username, changed_keys):
103 """Reschedule a user's document-scheduler jobs after a settings change.
105 Cache invalidation alone (``invalidate_settings_caches``) only clears the
106 scheduler's cached settings — it does not (re)create or tear down the
107 per-user interval jobs. So toggling ``document_scheduler.*`` settings (e.g.
108 ``sweep_library_collections`` or the legacy ``generate_rag``) would not take
109 effect until the user logged out and back in. Call this AFTER
110 ``invalidate_settings_caches`` so the scheduler re-reads fresh settings and
111 the change applies on the next tick.
113 Only reschedules when a ``document_scheduler.*`` key actually changed, so an
114 unrelated settings save never churns (and resets the timer of) the document
115 jobs. Best-effort and silent when the scheduler isn't running (e.g. tests,
116 CLI), mirroring ``invalidate_settings_caches``.
118 Args:
119 username: The user whose jobs should be re-evaluated.
120 changed_keys: Iterable of setting keys written by the request.
121 """
122 if not username or not changed_keys:
123 return
124 if not any(
125 str(key).startswith("document_scheduler.") for key in changed_keys
126 ):
127 return
128 try:
129 from ...scheduler.background import get_background_job_scheduler
131 scheduler = get_background_job_scheduler()
132 scheduler.reschedule_document_jobs(username)
133 except Exception:
134 logger.debug(
135 "Could not reschedule document jobs after settings change",
136 exc_info=True,
137 )
140def reschedule_zotero_jobs_if_needed(username, changed_keys):
141 """Reschedule a user's Zotero auto-sync job after a settings change.
143 Mirrors :func:`reschedule_document_jobs_if_needed` for ``zotero.*`` keys.
144 Without it, toggling ``zotero.auto_sync_enabled`` (or changing
145 ``zotero.sync_interval_minutes``) would not create/refresh the interval job
146 until the user logged out and back in — the schedule is otherwise only built
147 on the login path. Call AFTER ``invalidate_settings_caches`` so the
148 scheduler re-reads fresh settings. Only reschedules when a ``zotero.`` key
149 actually changed. Best-effort and silent when the scheduler isn't running
150 (e.g. tests, CLI).
152 Args:
153 username: The user whose Zotero job should be re-evaluated.
154 changed_keys: Iterable of setting keys written by the request.
155 """
156 if not username or not changed_keys:
157 return
158 if not any(str(key).startswith("zotero.") for key in changed_keys):
159 return
160 try:
161 from ...scheduler.background import get_background_job_scheduler
163 scheduler = get_background_job_scheduler()
164 scheduler.reschedule_zotero_jobs(username)
165 except Exception:
166 logger.debug(
167 "Could not reschedule Zotero jobs after settings change",
168 exc_info=True,
169 )
172def validate_setting(
173 setting: Setting, value: Any
174) -> tuple[bool, Optional[str]]:
175 """
176 Validate a setting value based on its type and constraints.
178 Args:
179 setting: The Setting object to validate against
180 value: The value to validate
182 Returns:
183 tuple[bool, Optional[str]]: (is_valid, error_message)
184 """
185 if setting.key == "search.fetch.summary_input_max_length":
186 try:
187 to_positive_int(value)
188 except ValueError:
189 return False, "Value must be a positive integer"
191 # Validate the exact summary integer before generic number conversion
192 # or the blank-input handling for other optional numeric settings.
193 raw_value = value
194 numeric_ui = setting.ui_element in ("number", "slider", "range")
195 # Blank form inputs explicitly clear optional numeric settings; they are
196 # normalized to None on write, not used as a fallback for bad numbers.
197 if numeric_ui and (
198 raw_value is None
199 or (isinstance(raw_value, str) and not raw_value.strip())
200 ):
201 return True, None
202 try:
203 value = get_typed_setting_value(
204 key=str(setting.key),
205 value=value,
206 ui_element=str(setting.ui_element),
207 default=None,
208 check_env=False,
209 )
210 except SettingsValueError as error:
211 if numeric_ui: 211 ↛ 213line 211 didn't jump to line 213 because the condition on line 211 was always true
212 return False, "Value must be a number"
213 if setting.ui_element == "checkbox":
214 return False, "Value must be a boolean"
215 return False, error.reason
217 # Validate based on UI element type
218 if setting.ui_element == "checkbox":
219 # After conversion, should be boolean
220 if not isinstance(value, bool):
221 return False, "Value must be a boolean"
223 elif setting.ui_element in ("number", "slider", "range"):
224 # After conversion, should be numeric
225 if (
226 isinstance(raw_value, bool)
227 or isinstance(value, bool)
228 or not isinstance(value, (int, float))
229 or (isinstance(value, float) and not math.isfinite(value))
230 ):
231 return False, "Value must be a number"
233 # Check min/max constraints if defined
234 if setting.min_value is not None and value < setting.min_value:
235 return False, f"Value must be at least {setting.min_value}"
236 if setting.max_value is not None and value > setting.max_value:
237 return False, f"Value must be at most {setting.max_value}"
239 elif setting.ui_element == "select":
240 # Check if value is in the allowed options
241 if setting.options: 241 ↛ 255line 241 didn't jump to line 255 because the condition on line 241 was always true
242 # Skip options validation for dynamically populated dropdowns
243 if setting.key not in DYNAMIC_SETTINGS:
244 allowed_values = [
245 opt.get("value") if isinstance(opt, dict) else opt
246 for opt in list(setting.options)
247 ]
248 if value not in allowed_values:
249 return (
250 False,
251 f"Value must be one of: {', '.join(str(v) for v in allowed_values)}",
252 )
254 # All checks passed
255 return True, None