Coverage for src / local_deep_research / config / thread_settings.py: 100%
59 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-25 01:07 +0000
1"""Shared thread-local storage for settings context
3This module provides a single thread-local storage instance that can be
4shared across different modules to maintain settings context in threads.
5"""
7import os
8import threading
9from contextlib import contextmanager
11from ..settings.manager import get_typed_setting_value
12from ..utilities.type_utils import to_bool
15class NoSettingsContextError(Exception):
16 """Raised when settings context is not available in a thread."""
18 pass
21# Shared thread-local storage for settings context
22_thread_local = threading.local()
25def set_settings_context(settings_context):
26 """Set a settings context for the current thread."""
27 _thread_local.settings_context = settings_context
30def clear_settings_context():
31 """Clear the settings context for the current thread.
33 Should be called in a finally block after set_settings_context() to prevent
34 context from leaking to subsequent tasks when threads are reused in a pool.
35 """
36 if hasattr(_thread_local, "settings_context"):
37 del _thread_local.settings_context
40def get_settings_context():
41 """Get the settings context for the current thread."""
42 if hasattr(_thread_local, "settings_context"):
43 return _thread_local.settings_context
44 return None
47@contextmanager
48def settings_context(ctx):
49 """Context manager that sets and clears settings context automatically.
51 Ensures cleanup even if an exception occurs, preventing context leaks
52 when threads are reused in a pool.
54 Example:
55 with settings_context(my_settings):
56 run_research()
57 """
58 set_settings_context(ctx)
59 try:
60 yield
61 finally:
62 clear_settings_context()
65def get_setting_from_snapshot(
66 key,
67 default=None,
68 username=None,
69 settings_snapshot=None,
70 check_fallback_llm=False,
71):
72 """Get setting from context only - no database access from threads.
74 Args:
75 key: Setting key to retrieve
76 default: Default value if setting not found
77 username: Username (unused, kept for backward compatibility)
78 settings_snapshot: Optional settings snapshot dict
79 check_fallback_llm: Whether to check LDR_USE_FALLBACK_LLM env var
81 Returns:
82 Setting value or default
84 Raises:
85 RuntimeError: If no settings context is available
86 """
87 # First check if we have settings_snapshot passed directly
88 value = None
89 if settings_snapshot and key in settings_snapshot:
90 value = settings_snapshot[key]
91 # Handle both full format {"value": x} and simplified format (just x)
92 if isinstance(value, dict) and "value" in value:
93 value = get_typed_setting_value(
94 key,
95 value["value"],
96 value.get("ui_element", "text"),
97 )
98 # else: value is already the raw value from simplified snapshot
99 # Search for child keys.
100 elif settings_snapshot:
101 for k, v in settings_snapshot.items():
102 if k.startswith(f"{key}."):
103 k = k.removeprefix(f"{key}.")
104 # Handle both full format {"value": x} and simplified format (just x)
105 if isinstance(v, dict) and "value" in v:
106 v = get_typed_setting_value(
107 key, v["value"], v.get("ui_element", "text")
108 )
109 # else: v is already the raw value from simplified snapshot
110 if value is None:
111 value = {k: v}
112 else:
113 value[k] = v
115 if value is not None:
116 # Extract value from dict structure if needed
117 return value
119 # Check if we have a settings context in this thread
120 if (
121 hasattr(_thread_local, "settings_context")
122 and _thread_local.settings_context
123 ):
124 value = _thread_local.settings_context.get_setting(key, default)
125 # Extract value from dict structure if needed (same as above)
126 if isinstance(value, dict) and "value" in value:
127 return value["value"]
128 return value
130 # In CI/test environment with fallback LLM, return default values
131 # But skip this if we're in test mode with mocks
132 if (
133 check_fallback_llm
134 and os.environ.get("LDR_USE_FALLBACK_LLM", "")
135 and not os.environ.get("LDR_TESTING_WITH_MOCKS", "")
136 ):
137 from loguru import logger
139 logger.debug(
140 f"Using default value for {key} in fallback LLM environment"
141 )
142 return default
144 # If a default was provided, return it instead of raising an exception
145 if default is not None:
146 from loguru import logger
148 logger.debug(
149 f"Setting '{key}' not found in snapshot or context, using default"
150 )
151 return default
153 # Only raise the exception if no default was provided
154 raise NoSettingsContextError(
155 f"No settings context available in thread for key '{key}'. All settings must be passed via settings_snapshot."
156 )
159def get_llm_setting_from_snapshot(
160 key, default=None, username=None, settings_snapshot=None
161):
162 """Get setting from snapshot with fallback LLM check enabled.
164 Convenience wrapper used by LLM provider modules that always need
165 ``check_fallback_llm=True``.
166 """
167 return get_setting_from_snapshot(
168 key, default, username, settings_snapshot, check_fallback_llm=True
169 )
172def get_bool_setting_from_snapshot(
173 key,
174 default=False,
175 username=None,
176 settings_snapshot=None,
177 check_fallback_llm=False,
178):
179 """Get a boolean setting from snapshot, handling string conversion.
181 This centralizes the string-to-boolean conversion logic for settings
182 retrieved from snapshots. Handles various truthy string representations
183 that may come from API requests, config files, or SQLite.
185 Args:
186 key: Setting key to retrieve
187 default: Default boolean value if setting not found
188 username: Username (unused, kept for backward compatibility)
189 settings_snapshot: Optional settings snapshot dict
190 check_fallback_llm: Whether to check LDR_USE_FALLBACK_LLM env var
192 Returns:
193 Boolean value of the setting
194 """
195 value = get_setting_from_snapshot(
196 key,
197 default,
198 username=username,
199 settings_snapshot=settings_snapshot,
200 check_fallback_llm=check_fallback_llm,
201 )
203 return to_bool(value, default)