Coverage for src/local_deep_research/web/services/settings_service.py: 99%
69 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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.manager import DYNAMIC_SETTINGS, get_typed_setting_value
9from ...utilities.db_utils import get_settings_manager
11# Settings with dynamically populated options (excluded from validation).
12# Single source of truth lives in `settings/manager.py` (it is also used by
13# `import_settings` there); re-exported here for the web save path. The
14# re-export must stay identity-equal to the manager's list — pinned by
15# tests/settings/test_import_settings_validation.py.
18def set_setting(
19 key: str,
20 value: Any,
21 commit: bool = True,
22 db_session: Optional[Session] = None,
23) -> bool:
24 """
25 Set a setting value
27 Args:
28 key: Setting key
29 value: Setting value
30 commit: Whether to commit the change
31 db_session: Optional database session
33 Returns:
34 bool: True if successful
35 """
36 manager = get_settings_manager(db_session)
37 return bool(manager.set_setting(key, value, commit))
40def get_all_settings() -> Dict[str, Any]:
41 """
42 Get all settings, optionally filtered by type
44 Returns:
45 Dict[str, Any]: Dictionary of settings
47 """
48 manager = get_settings_manager()
49 return manager.get_all_settings() # type: ignore[no-any-return]
52def create_or_update_setting(
53 setting: Union[Dict[str, Any], Setting],
54 commit: bool = True,
55 db_session: Optional[Session] = None,
56) -> Optional[Setting]:
57 """
58 Create or update a setting
60 Args:
61 setting: Setting dictionary or object
62 commit: Whether to commit the change
63 db_session: Optional database session
65 Returns:
66 Optional[Setting]: The setting object if successful
67 """
68 manager = get_settings_manager(db_session)
69 return manager.create_or_update_setting(setting, commit) # type: ignore[no-any-return]
72def invalidate_settings_caches(username=None):
73 """Invalidate all settings-related caches after a settings mutation.
75 Call this after any route that creates, updates, or deletes settings
76 so background services pick up changes immediately.
78 Safe to call from anywhere — silently skips caches that aren't initialized.
80 Args:
81 username: If provided, invalidates only that user's scheduler cache.
82 If None, invalidates all users' scheduler caches.
83 """
84 # News scheduler per-user settings cache (TTLCache, 5-min TTL).
85 # NOTE: the LLM provider dropdown is served by the auto-discovery path
86 # (llm/providers/auto_discovery.get_discovered_provider_options), which
87 # rebuilds per call and has no cache to invalidate here.
88 try:
89 from ...scheduler.background import get_background_job_scheduler
91 scheduler = get_background_job_scheduler()
92 if username is not None:
93 scheduler.invalidate_user_settings_cache(username)
94 else:
95 scheduler.invalidate_all_settings_cache()
96 except Exception:
97 logger.debug("Could not invalidate scheduler cache", exc_info=True)
100def reschedule_document_jobs_if_needed(username, changed_keys):
101 """Reschedule a user's document-scheduler jobs after a settings change.
103 Cache invalidation alone (``invalidate_settings_caches``) only clears the
104 scheduler's cached settings — it does not (re)create or tear down the
105 per-user interval jobs. So toggling ``document_scheduler.*`` settings (e.g.
106 ``sweep_library_collections`` or the legacy ``generate_rag``) would not take
107 effect until the user logged out and back in. Call this AFTER
108 ``invalidate_settings_caches`` so the scheduler re-reads fresh settings and
109 the change applies on the next tick.
111 Only reschedules when a ``document_scheduler.*`` key actually changed, so an
112 unrelated settings save never churns (and resets the timer of) the document
113 jobs. Best-effort and silent when the scheduler isn't running (e.g. tests,
114 CLI), mirroring ``invalidate_settings_caches``.
116 Args:
117 username: The user whose jobs should be re-evaluated.
118 changed_keys: Iterable of setting keys written by the request.
119 """
120 if not username or not changed_keys:
121 return
122 if not any(
123 str(key).startswith("document_scheduler.") for key in changed_keys
124 ):
125 return
126 try:
127 from ...scheduler.background import get_background_job_scheduler
129 scheduler = get_background_job_scheduler()
130 scheduler.reschedule_document_jobs(username)
131 except Exception:
132 logger.debug(
133 "Could not reschedule document jobs after settings change",
134 exc_info=True,
135 )
138def reschedule_zotero_jobs_if_needed(username, changed_keys):
139 """Reschedule a user's Zotero auto-sync job after a settings change.
141 Mirrors :func:`reschedule_document_jobs_if_needed` for ``zotero.*`` keys.
142 Without it, toggling ``zotero.auto_sync_enabled`` (or changing
143 ``zotero.sync_interval_minutes``) would not create/refresh the interval job
144 until the user logged out and back in — the schedule is otherwise only built
145 on the login path. Call AFTER ``invalidate_settings_caches`` so the
146 scheduler re-reads fresh settings. Only reschedules when a ``zotero.`` key
147 actually changed. Best-effort and silent when the scheduler isn't running
148 (e.g. tests, CLI).
150 Args:
151 username: The user whose Zotero job should be re-evaluated.
152 changed_keys: Iterable of setting keys written by the request.
153 """
154 if not username or not changed_keys:
155 return
156 if not any(str(key).startswith("zotero.") for key in changed_keys):
157 return
158 try:
159 from ...scheduler.background import get_background_job_scheduler
161 scheduler = get_background_job_scheduler()
162 scheduler.reschedule_zotero_jobs(username)
163 except Exception:
164 logger.debug(
165 "Could not reschedule Zotero jobs after settings change",
166 exc_info=True,
167 )
170def validate_setting(
171 setting: Setting, value: Any
172) -> tuple[bool, Optional[str]]:
173 """
174 Validate a setting value based on its type and constraints.
176 Args:
177 setting: The Setting object to validate against
178 value: The value to validate
180 Returns:
181 tuple[bool, Optional[str]]: (is_valid, error_message)
182 """
183 # Keep the submitted value so a failed conversion cannot be confused with
184 # an intentionally-unset optional numeric. The converter uses ``None`` for
185 # both outcomes when its default is None.
186 raw_value = value
188 # Convert value to appropriate type first using SettingsManager's logic.
189 # Inlined (rather than delegating to routers.settings.validate_setting) so
190 # this service no longer imports the router — breaking the settings
191 # service <-> routes circular dependency (#2898).
192 value = get_typed_setting_value(
193 key=str(setting.key),
194 value=value,
195 ui_element=str(setting.ui_element),
196 default=None,
197 check_env=False,
198 )
200 # Validate based on UI element type
201 if setting.ui_element == "checkbox":
202 # After conversion, should be boolean
203 if not isinstance(value, bool):
204 return False, "Value must be a boolean"
206 elif setting.ui_element in ("number", "slider", "range"):
207 # None and blank HTML inputs represent an intentionally-unset optional
208 # numeric; a nonblank conversion failure must remain invalid.
209 if raw_value is None or (
210 isinstance(raw_value, str) and not raw_value.strip()
211 ):
212 return True, None
214 # After conversion, should be numeric
215 if (
216 isinstance(raw_value, bool)
217 or isinstance(value, bool)
218 or not isinstance(value, (int, float))
219 or (isinstance(value, float) and not math.isfinite(value))
220 ):
221 return False, "Value must be a number"
223 # Check min/max constraints if defined
224 if setting.min_value is not None and value < setting.min_value:
225 return False, f"Value must be at least {setting.min_value}"
226 if setting.max_value is not None and value > setting.max_value:
227 return False, f"Value must be at most {setting.max_value}"
229 elif setting.ui_element == "select":
230 # Check if value is in the allowed options
231 if setting.options: 231 ↛ 245line 231 didn't jump to line 245 because the condition on line 231 was always true
232 # Skip options validation for dynamically populated dropdowns
233 if setting.key not in DYNAMIC_SETTINGS:
234 allowed_values = [
235 opt.get("value") if isinstance(opt, dict) else opt
236 for opt in list(setting.options)
237 ]
238 if value not in allowed_values:
239 return (
240 False,
241 f"Value must be one of: {', '.join(str(v) for v in allowed_values)}",
242 )
244 # All checks passed
245 return True, None