Coverage for src/local_deep_research/web/services/settings_service.py: 92%

1from typing import Any, Dict, Optional, Union 

2 

3from loguru import logger 

4from sqlalchemy.orm import Session 

5 

6from ...database.models import Setting 

7from ...settings.manager import get_typed_setting_value 

8from ...utilities.db_utils import get_settings_manager 

9 

10# Settings with dynamically populated options (excluded from validation) 

11DYNAMIC_SETTINGS = ["llm.provider", "llm.model", "search.tool"] 

12 

13 

14def set_setting( 

15 key: str, 

16 value: Any, 

17 commit: bool = True, 

18 db_session: Optional[Session] = None, 

19) -> bool: 

20 """ 

21 Set a setting value 

22 

23 Args: 

24 key: Setting key 

25 value: Setting value 

26 commit: Whether to commit the change 

27 db_session: Optional database session 

28 

29 Returns: 

30 bool: True if successful 

31 """ 

32 manager = get_settings_manager(db_session) 

33 return bool(manager.set_setting(key, value, commit)) 

34 

35 

36def get_all_settings() -> Dict[str, Any]: 

37 """ 

38 Get all settings, optionally filtered by type 

39 

40 Returns: 

41 Dict[str, Any]: Dictionary of settings 

42 

43 """ 

44 manager = get_settings_manager() 

45 return manager.get_all_settings() # type: ignore[no-any-return] 

46 

47 

48def create_or_update_setting( 

49 setting: Union[Dict[str, Any], Setting], 

50 commit: bool = True, 

51 db_session: Optional[Session] = None, 

52) -> Optional[Setting]: 

53 """ 

54 Create or update a setting 

55 

56 Args: 

57 setting: Setting dictionary or object 

58 commit: Whether to commit the change 

59 db_session: Optional database session 

60 

61 Returns: 

62 Optional[Setting]: The setting object if successful 

63 """ 

64 manager = get_settings_manager(db_session) 

65 return manager.create_or_update_setting(setting, commit) # type: ignore[no-any-return] 

66 

67 

68def invalidate_settings_caches(username=None): 

69 """Invalidate all settings-related caches after a settings mutation. 

70 

71 Call this after any route that creates, updates, or deletes settings 

72 so background services pick up changes immediately. 

73 

74 Safe to call from anywhere — silently skips caches that aren't initialized. 

75 

76 Args: 

77 username: If provided, invalidates only that user's scheduler cache. 

78 If None, invalidates all users' scheduler caches. 

79 """ 

80 # News scheduler per-user settings cache (TTLCache, 5-min TTL). 

81 # NOTE: the LLM provider dropdown is served by the auto-discovery path 

82 # (llm/providers/auto_discovery.get_discovered_provider_options), which 

83 # rebuilds per call and has no cache to invalidate here. 

84 try: 

85 from ...scheduler.background import get_background_job_scheduler 

86 

87 scheduler = get_background_job_scheduler() 

88 if username is not None: 88 ↛ 91line 88 didn't jump to line 91 because the condition on line 88 was always true

89 scheduler.invalidate_user_settings_cache(username) 

90 else: 

91 scheduler.invalidate_all_settings_cache() 

92 except Exception: 

93 logger.debug("Could not invalidate scheduler cache", exc_info=True) 

94 

95 

96def reschedule_document_jobs_if_needed(username, changed_keys): 

97 """Reschedule a user's document-scheduler jobs after a settings change. 

98 

99 Cache invalidation alone (``invalidate_settings_caches``) only clears the 

100 scheduler's cached settings — it does not (re)create or tear down the 

101 per-user interval jobs. So toggling ``document_scheduler.*`` settings (e.g. 

102 ``sweep_library_collections`` or the legacy ``generate_rag``) would not take 

103 effect until the user logged out and back in. Call this AFTER 

104 ``invalidate_settings_caches`` so the scheduler re-reads fresh settings and 

105 the change applies on the next tick. 

106 

107 Only reschedules when a ``document_scheduler.*`` key actually changed, so an 

108 unrelated settings save never churns (and resets the timer of) the document 

109 jobs. Best-effort and silent when the scheduler isn't running (e.g. tests, 

110 CLI), mirroring ``invalidate_settings_caches``. 

111 

112 Args: 

113 username: The user whose jobs should be re-evaluated. 

114 changed_keys: Iterable of setting keys written by the request. 

115 """ 

116 if not username or not changed_keys: 

117 return 

118 if not any( 

119 str(key).startswith("document_scheduler.") for key in changed_keys 

120 ): 

121 return 

122 try: 

123 from ...scheduler.background import get_background_job_scheduler 

124 

125 scheduler = get_background_job_scheduler() 

126 scheduler.reschedule_document_jobs(username) 

127 except Exception: 

128 logger.debug( 

129 "Could not reschedule document jobs after settings change", 

130 exc_info=True, 

131 ) 

132 

133 

134def reschedule_zotero_jobs_if_needed(username, changed_keys): 

135 """Reschedule a user's Zotero auto-sync job after a settings change. 

136 

137 Mirrors :func:`reschedule_document_jobs_if_needed` for ``zotero.*`` keys. 

138 Without it, toggling ``zotero.auto_sync_enabled`` (or changing 

139 ``zotero.sync_interval_minutes``) would not create/refresh the interval job 

140 until the user logged out and back in — the schedule is otherwise only built 

141 on the login path. Call AFTER ``invalidate_settings_caches`` so the 

142 scheduler re-reads fresh settings. Only reschedules when a ``zotero.`` key 

143 actually changed. Best-effort and silent when the scheduler isn't running 

144 (e.g. tests, CLI). 

145 

146 Args: 

147 username: The user whose Zotero job should be re-evaluated. 

148 changed_keys: Iterable of setting keys written by the request. 

149 """ 

150 if not username or not changed_keys: 

151 return 

152 if not any(str(key).startswith("zotero.") for key in changed_keys): 

153 return 

154 try: 

155 from ...scheduler.background import get_background_job_scheduler 

156 

157 scheduler = get_background_job_scheduler() 

158 scheduler.reschedule_zotero_jobs(username) 

159 except Exception: 

160 logger.debug( 

161 "Could not reschedule Zotero jobs after settings change", 

162 exc_info=True, 

163 ) 

164 

165 

166def validate_setting( 

167 setting: Setting, value: Any 

168) -> tuple[bool, Optional[str]]: 

169 """ 

170 Validate a setting value based on its type and constraints. 

171 

172 Args: 

173 setting: The Setting object to validate against 

174 value: The value to validate 

175 

176 Returns: 

177 tuple[bool, Optional[str]]: (is_valid, error_message) 

178 """ 

179 # Convert value to appropriate type first using SettingsManager's logic 

180 value = get_typed_setting_value( 

181 key=str(setting.key), 

182 value=value, 

183 ui_element=str(setting.ui_element), 

184 default=None, 

185 check_env=False, 

186 ) 

187 

188 # Validate based on UI element type 

189 if setting.ui_element == "checkbox": 

190 # After conversion, should be boolean 

191 if not isinstance(value, bool): 

192 return False, "Value must be a boolean" 

193 

194 elif setting.ui_element in ("number", "slider", "range"): 

195 # After conversion, should be numeric 

196 if not isinstance(value, (int, float)): 

197 return False, "Value must be a number" 

198 

199 # Check min/max constraints if defined 

200 if setting.min_value is not None and value < setting.min_value: 

201 return False, f"Value must be at least {setting.min_value}" 

202 if setting.max_value is not None and value > setting.max_value: 

203 return False, f"Value must be at most {setting.max_value}" 

204 

205 elif setting.ui_element == "select": 

206 # Check if value is in the allowed options 

207 if setting.options: 

208 # Skip options validation for dynamically populated dropdowns 

209 if setting.key not in DYNAMIC_SETTINGS: 

210 allowed_values = [ 

211 opt.get("value") if isinstance(opt, dict) else opt 

212 for opt in list(setting.options) 

213 ] 

214 if value not in allowed_values: 

215 return ( 

216 False, 

217 f"Value must be one of: {', '.join(str(v) for v in allowed_values)}", 

218 ) 

219 

220 # All checks passed 

221 return True, None