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# Sentinel distinguishing "key absent from snapshot" from "key present with 

24# value None". Using None for both collapses legitimately-stored null values 

25# (e.g. embeddings.openai.dimensions, which defaults to JSON null) into the 

26# "not found" path, raising NoSettingsContextError in Flask request threads 

27# that have no thread-local context. See #4208. 

28_NOT_FOUND = object() 

29 

30 

31def set_settings_context(settings_context): 

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

33 _thread_local.settings_context = settings_context 

34 

35 

36def clear_settings_context(): 

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

38 

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

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

41 """ 

42 if hasattr(_thread_local, "settings_context"): 

43 del _thread_local.settings_context 

44 

45 

46def get_settings_context(): 

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

48 if hasattr(_thread_local, "settings_context"): 

49 return _thread_local.settings_context 

50 return None 

51 

52 

53@contextmanager 

54def settings_context(ctx): 

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

56 

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

58 when threads are reused in a pool. 

59 

60 Example: 

61 with settings_context(my_settings): 

62 run_research() 

63 """ 

64 set_settings_context(ctx) 

65 try: 

66 yield 

67 finally: 

68 clear_settings_context() 

69 

70 

71def get_setting_from_snapshot( 

72 key, 

73 default=None, 

74 username=None, 

75 settings_snapshot=None, 

76): 

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

78 

79 Args: 

80 key: Setting key to retrieve 

81 default: Default value if setting not found 

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

83 settings_snapshot: Optional settings snapshot dict 

84 

85 Returns: 

86 Setting value or default 

87 

88 Raises: 

89 RuntimeError: If no settings context is available 

90 """ 

91 # First check if we have settings_snapshot passed directly. 

92 # _NOT_FOUND (not None) is the absence sentinel so a key whose stored 

93 # value is None is still treated as found. See #4208. 

94 value = _NOT_FOUND 

95 if settings_snapshot and key in settings_snapshot: 

96 raw = settings_snapshot[key] 

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

98 if isinstance(raw, dict) and "value" in raw: 

99 value = get_typed_setting_value( 

100 key, 

101 raw["value"], 

102 raw.get("ui_element", "text"), 

103 ) 

104 else: 

105 value = raw 

106 # Search for child keys. 

107 elif settings_snapshot: 

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

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

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

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

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

113 v = get_typed_setting_value( 

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

115 ) 

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

117 if value is _NOT_FOUND: 

118 value = {k: v} 

119 else: 

120 value[k] = v 

121 

122 if value is not _NOT_FOUND: 

123 # Extract value from dict structure if needed 

124 return value 

125 

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

127 if ( 

128 hasattr(_thread_local, "settings_context") 

129 and _thread_local.settings_context 

130 ): 

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

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

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

134 return value["value"] 

135 return value 

136 

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

138 if default is not None: 

139 from loguru import logger 

140 

141 logger.debug( 

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

143 ) 

144 return default 

145 

146 # Only raise the exception if no default was provided 

147 raise NoSettingsContextError( 

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

149 ) 

150 

151 

152def get_bool_setting_from_snapshot( 

153 key, 

154 default=False, 

155 username=None, 

156 settings_snapshot=None, 

157): 

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

159 

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

161 retrieved from snapshots. Handles various truthy string representations 

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

163 

164 Args: 

165 key: Setting key to retrieve 

166 default: Default boolean value if setting not found 

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

168 settings_snapshot: Optional settings snapshot dict 

169 

170 Returns: 

171 Boolean value of the setting 

172 """ 

173 value = get_setting_from_snapshot( 

174 key, 

175 default, 

176 username=username, 

177 settings_snapshot=settings_snapshot, 

178 ) 

179 

180 return to_bool(value, default)