Coverage for src / local_deep_research / config / thread_settings.py: 100%

1"""Shared thread-local storage for settings context 

2 

3This module provides a single thread-local storage instance that can be 

4shared across different modules to maintain settings context in threads. 

5""" 

6 

7import threading 

8from contextlib import contextmanager 

9 

10from ..settings.manager import get_typed_setting_value 

11from ..utilities.type_utils import to_bool 

12 

13 

14class NoSettingsContextError(Exception): 

15 """Raised when settings context is not available in a thread.""" 

16 

17 pass 

18 

19 

20# Shared thread-local storage for settings context 

21_thread_local = threading.local() 

22 

23 

24def set_settings_context(settings_context): 

25 """Set a settings context for the current thread.""" 

26 _thread_local.settings_context = settings_context 

27 

28 

29def clear_settings_context(): 

30 """Clear the settings context for the current thread. 

31 

32 Should be called in a finally block after set_settings_context() to prevent 

33 context from leaking to subsequent tasks when threads are reused in a pool. 

34 """ 

35 if hasattr(_thread_local, "settings_context"): 

36 del _thread_local.settings_context 

37 

38 

39def get_settings_context(): 

40 """Get the settings context for the current thread.""" 

41 if hasattr(_thread_local, "settings_context"): 

42 return _thread_local.settings_context 

43 return None 

44 

45 

46@contextmanager 

47def settings_context(ctx): 

48 """Context manager that sets and clears settings context automatically. 

49 

50 Ensures cleanup even if an exception occurs, preventing context leaks 

51 when threads are reused in a pool. 

52 

53 Example: 

54 with settings_context(my_settings): 

55 run_research() 

56 """ 

57 set_settings_context(ctx) 

58 try: 

59 yield 

60 finally: 

61 clear_settings_context() 

62 

63 

64def get_setting_from_snapshot( 

65 key, 

66 default=None, 

67 username=None, 

68 settings_snapshot=None, 

69): 

70 """Get setting from context only - no database access from threads. 

71 

72 Args: 

73 key: Setting key to retrieve 

74 default: Default value if setting not found 

75 username: Username (unused, kept for backward compatibility) 

76 settings_snapshot: Optional settings snapshot dict 

77 

78 Returns: 

79 Setting value or default 

80 

81 Raises: 

82 RuntimeError: If no settings context is available 

83 """ 

84 # First check if we have settings_snapshot passed directly 

85 value = None 

86 if settings_snapshot and key in settings_snapshot: 

87 value = settings_snapshot[key] 

88 # Handle both full format {"value": x} and simplified format (just x) 

89 if isinstance(value, dict) and "value" in value: 

90 value = get_typed_setting_value( 

91 key, 

92 value["value"], 

93 value.get("ui_element", "text"), 

94 ) 

95 # else: value is already the raw value from simplified snapshot 

96 # Search for child keys. 

97 elif settings_snapshot: 

98 for k, v in settings_snapshot.items(): 

99 if k.startswith(f"{key}."): 

100 k = k.removeprefix(f"{key}.") 

101 # Handle both full format {"value": x} and simplified format (just x) 

102 if isinstance(v, dict) and "value" in v: 

103 v = get_typed_setting_value( 

104 k, v["value"], v.get("ui_element", "text") 

105 ) 

106 # else: v is already the raw value from simplified snapshot 

107 if value is None: 

108 value = {k: v} 

109 else: 

110 value[k] = v 

111 

112 if value is not None: 

113 # Extract value from dict structure if needed 

114 return value 

115 

116 # Check if we have a settings context in this thread 

117 if ( 

118 hasattr(_thread_local, "settings_context") 

119 and _thread_local.settings_context 

120 ): 

121 value = _thread_local.settings_context.get_setting(key, default) 

122 # Extract value from dict structure if needed (same as above) 

123 if isinstance(value, dict) and "value" in value: 

124 return value["value"] 

125 return value 

126 

127 # If a default was provided, return it instead of raising an exception 

128 if default is not None: 

129 from loguru import logger 

130 

131 logger.debug( 

132 f"Setting '{key}' not found in snapshot or context, using default" 

133 ) 

134 return default 

135 

136 # Only raise the exception if no default was provided 

137 raise NoSettingsContextError( 

138 f"No settings context available in thread for key '{key}'. All settings must be passed via settings_snapshot." 

139 ) 

140 

141 

142def get_bool_setting_from_snapshot( 

143 key, 

144 default=False, 

145 username=None, 

146 settings_snapshot=None, 

147): 

148 """Get a boolean setting from snapshot, handling string conversion. 

149 

150 This centralizes the string-to-boolean conversion logic for settings 

151 retrieved from snapshots. Handles various truthy string representations 

152 that may come from API requests, config files, or SQLite. 

153 

154 Args: 

155 key: Setting key to retrieve 

156 default: Default boolean value if setting not found 

157 username: Username (unused, kept for backward compatibility) 

158 settings_snapshot: Optional settings snapshot dict 

159 

160 Returns: 

161 Boolean value of the setting 

162 """ 

163 value = get_setting_from_snapshot( 

164 key, 

165 default, 

166 username=username, 

167 settings_snapshot=settings_snapshot, 

168 ) 

169 

170 return to_bool(value, default)