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

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 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() 

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) 

58 

59 

60def bulk_update_settings( 

61 settings_dict: Dict[str, Any], commit: bool = True, db_session=None 

62) -> bool: 

63 """ 

64 Update multiple settings from a dictionary 

65 

66 Args: 

67 settings_dict: Dictionary of setting keys and values 

68 commit: Whether to commit the changes 

69 db_session: Optional database session 

70 

71 Returns: 

72 bool: True if all updates were successful 

73 """ 

74 manager = get_settings_manager(db_session) 

75 success = True 

76 

77 for key, value in settings_dict.items(): 

78 if not manager.set_setting(key, value, commit=False): 

79 success = False 

80 

81 if commit and success and manager.db_session: 

82 try: 

83 manager.db_session.commit() 

84 # Emit WebSocket event for all changed settings 

85 manager._emit_settings_changed(list(settings_dict.keys())) 

86 except Exception: 

87 logger.exception("Error committing bulk settings update") 

88 manager.db_session.rollback() 

89 success = False 

90 

91 return success 

92 

93 

94def validate_setting( 

95 setting: Setting, value: Any 

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

97 """ 

98 Validate a setting value based on its type and constraints 

99 

100 Args: 

101 setting: The Setting object to validate against 

102 value: The value to validate 

103 

104 Returns: 

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

106 """ 

107 from ..routes.settings_routes import ( 

108 validate_setting as routes_validate_setting, 

109 ) 

110 

111 return routes_validate_setting(setting, value)