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

1from typing import Any, Dict, Optional, Union 

2 

3from loguru import logger 

4 

5from ...database.models import Setting 

6from ...utilities.db_utils import get_settings_manager 

7 

8 

9def set_setting( 

10 key: str, value: Any, commit: bool = True, db_session=None 

11) -> bool: 

12 """ 

13 Set a setting value 

14 

15 Args: 

16 key: Setting key 

17 value: Setting value 

18 commit: Whether to commit the change 

19 db_session: Optional database session 

20 

21 Returns: 

22 bool: True if successful 

23 """ 

24 manager = get_settings_manager(db_session) 

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

26 

27 

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

29 """ 

30 Get all settings, optionally filtered by type 

31 

32 Returns: 

33 Dict[str, Any]: Dictionary of settings 

34 

35 """ 

36 manager = get_settings_manager() 

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

38 

39 

40def create_or_update_setting( 

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

42 commit: bool = True, 

43 db_session=None, 

44) -> Optional[Setting]: 

45 """ 

46 Create or update a setting 

47 

48 Args: 

49 setting: Setting dictionary or object 

50 commit: Whether to commit the change 

51 db_session: Optional database session 

52 

53 Returns: 

54 Optional[Setting]: The setting object if successful 

55 """ 

56 manager = get_settings_manager(db_session) 

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

58 

59 

60def invalidate_settings_caches(username=None): 

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

62 

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

64 so background services pick up changes immediately. 

65 

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

67 

68 Args: 

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

70 If None, invalidates all users' scheduler caches. 

71 """ 

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

73 try: 

74 from ...news.subscription_manager.scheduler import get_news_scheduler 

75 

76 scheduler = get_news_scheduler() 

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

78 scheduler.invalidate_user_settings_cache(username) 

79 else: 

80 scheduler.invalidate_all_settings_cache() 

81 except Exception: 

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

83 

84 # 2. LLM provider cache (functools.cache, no TTL) 

85 try: 

86 from ...config.llm_config import get_available_providers 

87 

88 get_available_providers.cache_clear() 

89 except Exception: 

90 logger.debug("Could not clear provider cache", exc_info=True) 

91 

92 

93def validate_setting( 

94 setting: Setting, value: Any 

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

96 """ 

97 Validate a setting value based on its type and constraints 

98 

99 Args: 

100 setting: The Setting object to validate against 

101 value: The value to validate 

102 

103 Returns: 

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

105 """ 

106 from ..routes.settings_routes import ( 

107 validate_setting as routes_validate_setting, 

108 ) 

109 

110 return routes_validate_setting(setting, value)