Coverage for src/local_deep_research/web_search_engines/search_engines_config.py: 89%
111 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Configuration file for search engines.
3Loads search engine definitions from the user's configuration.
4"""
6from typing import Any, Dict, Optional
7from sqlalchemy.orm import Session
9from ..security.secure_logging import logger
11from ..config.thread_settings import get_setting_from_snapshot
12from ..utilities.db_utils import get_settings_manager
13from .search_engine_base import _is_api_key_placeholder
16def _get_setting(
17 key: str,
18 default_value: Any = None,
19 db_session: Optional[Session] = None,
20 settings_snapshot: Optional[Dict[str, Any]] = None,
21 username: Optional[str] = None,
22) -> Any:
23 """
24 Get a setting from either a database session or settings snapshot.
26 Args:
27 key: The setting key
28 default_value: Default value if setting not found
29 db_session: Database session for direct access
30 settings_snapshot: Settings snapshot for thread context
31 username: Username for backward compatibility
33 Returns:
34 The setting value or default_value if not found
35 """
36 # Try settings snapshot first (thread context)
37 if settings_snapshot:
38 try:
39 return get_setting_from_snapshot(
40 key, default_value, settings_snapshot=settings_snapshot
41 )
42 except Exception as e:
43 logger.debug(f"Could not get setting {key} from snapshot: {e}")
45 # Try database session if available
46 if db_session:
47 try:
48 settings_manager = get_settings_manager(db_session, username)
49 return settings_manager.get_setting(key, default_value)
50 except Exception as e:
51 logger.debug(f"Could not get setting {key} from db_session: {e}")
53 # Return default if all methods fail
54 logger.warning(
55 f"Could not retrieve setting '{key}', returning default: {default_value}"
56 )
57 return default_value
60def _extract_per_engine_config(
61 raw_config: Dict[str, Any],
62) -> Dict[str, Dict[str, Any]]:
63 """
64 Converts the "flat" configuration loaded from the settings database into
65 individual settings dictionaries for each engine.
67 Args:
68 raw_config: The raw "flat" configuration.
70 Returns:
71 Configuration dictionaries indexed by engine name.
73 """
74 nested_config: dict[str, Any] = {}
75 for key, value in raw_config.items():
76 if "." in key:
77 # This is a higher-level key.
78 top_level_key = key.split(".")[0]
79 lower_keys = ".".join(key.split(".")[1:])
80 nested_config.setdefault(top_level_key, {})[lower_keys] = value
81 else:
82 # This is a low-level key.
83 nested_config[key] = value
85 # Expand all the lower-level keys.
86 for key, value in nested_config.items():
87 if isinstance(value, dict):
88 # Expand the child keys.
89 nested_config[key] = _extract_per_engine_config(value)
91 return nested_config
94def search_config(
95 username: Optional[str] = None,
96 db_session: Optional[Session] = None,
97 settings_snapshot: Optional[Dict[str, Any]] = None,
98) -> Dict[str, Any]:
99 """
100 Returns the search engine configuration loaded from the database or settings snapshot.
102 Args:
103 username: Username for backward compatibility (deprecated)
104 db_session: Database session for direct access (preferred for web routes)
105 settings_snapshot: Settings snapshot for thread context (preferred for background threads)
107 Returns:
108 The search engine configuration loaded from the database or snapshot.
109 """
110 # Extract search engine definitions
111 config_data = _get_setting(
112 "search.engine.web",
113 {},
114 db_session=db_session,
115 settings_snapshot=settings_snapshot,
116 username=username,
117 )
119 search_engines = _extract_per_engine_config(config_data)
121 # Inject module/class from the hardcoded engine registry.
122 # This is the single source of truth for which Python module implements
123 # each engine — these values are never read from the settings DB.
124 from .engine_registry import ENGINE_REGISTRY
126 for name, entry in ENGINE_REGISTRY.items():
127 if name in search_engines:
128 search_engines[name]["module_path"] = entry.module_path
129 search_engines[name]["class_name"] = entry.class_name
130 if entry.full_search_module:
131 search_engines[name]["full_search_module"] = (
132 entry.full_search_module
133 )
134 search_engines[name]["full_search_class"] = (
135 entry.full_search_class
136 )
138 # Add registered retrievers as available search engines
139 from .retriever_registry import retriever_registry
141 for name in retriever_registry.list_registered():
142 search_engines[name] = {
143 "module_path": ".engines.search_engine_retriever",
144 "class_name": "RetrieverSearchEngine",
145 "requires_api_key": False,
146 "requires_llm": False,
147 "description": f"LangChain retriever: {name}",
148 "strengths": [
149 "Domain-specific knowledge",
150 "No rate limits",
151 "Fast retrieval",
152 ],
153 "weaknesses": ["Limited to indexed content"],
154 "supports_full_search": True,
155 "is_retriever": True, # Mark as retriever for identification
156 }
158 logger.info(
159 f"Loaded {len(search_engines)} search engines from configuration file"
160 )
161 logger.info(f"\n {', '.join(sorted(search_engines.keys()))} \n")
163 # Register Library RAG as a search engine
164 library_enabled = _get_setting(
165 "search.engine.library.enabled",
166 True,
167 db_session=db_session,
168 settings_snapshot=settings_snapshot,
169 username=username,
170 )
172 if library_enabled:
173 search_engines["library"] = {
174 "module_path": ".engines.search_engine_library",
175 "class_name": "LibraryRAGSearchEngine",
176 "requires_llm": True,
177 "display_name": "Search All Collections",
178 "default_params": {},
179 "description": "Search across all your document collections using semantic search",
180 "strengths": [
181 "Searches all your curated collections of research papers and documents",
182 "Uses semantic search for better relevance",
183 "Returns documents you've already saved and reviewed",
184 ],
185 "weaknesses": [
186 "Limited to documents already in your collections",
187 "Requires documents to be indexed first",
188 ],
189 "reliability": "High - searches all your collections",
190 }
191 logger.info("Registered Library RAG as search engine")
193 # Register document collections as individual search engines
194 if library_enabled:
195 try:
196 from ..database.models.library import Collection
197 from ..database.session_context import get_user_db_session
199 # Get username from settings_snapshot if available
200 collection_username = (
201 settings_snapshot.get("_username")
202 if settings_snapshot
203 else username
204 )
206 if collection_username:
207 with get_user_db_session(collection_username) as session:
208 collections = session.query(Collection).all()
210 for collection in collections:
211 engine_id = f"collection_{collection.id}"
212 # Add suffix to distinguish from the all-collections search
213 display_name = f"{collection.name} (Collection)"
214 # Egress classification follows the per-collection
215 # public/private flag (default private). A "public"
216 # collection counts as a public engine (allowed under
217 # PUBLIC_ONLY); a private one is local-only. NULL
218 # (pre-migration rows) reads as private — the safe
219 # default.
220 collection_is_public = bool(
221 getattr(collection, "is_public", False)
222 )
223 # Usability flag (NOT egress): whether the LangGraph
224 # research agent offers this collection as a tool. NULL
225 # (pre-migration rows) reads as available (True). Uses
226 # the same `is not False` idiom as the rag_routes
227 # serializers so all call sites share one NULL→available
228 # default and can't drift.
229 collection_agent_enabled = (
230 getattr(collection, "agent_enabled", True)
231 is not False
232 )
233 search_engines[engine_id] = {
234 "module_path": ".engines.search_engine_collection",
235 "class_name": "CollectionSearchEngine",
236 "requires_llm": True,
237 "is_local": not collection_is_public,
238 "is_public": collection_is_public,
239 "agent_enabled": collection_agent_enabled,
240 "display_name": display_name,
241 "default_params": {
242 "collection_id": collection.id,
243 "collection_name": collection.name,
244 },
245 "description": (
246 collection.description
247 if collection.description
248 else f"Search documents in {collection.name} collection only"
249 ),
250 "strengths": [
251 f"Searches only documents in {collection.name}",
252 "Focused semantic search within specific topic area",
253 "Returns documents from a curated collection",
254 ],
255 "weaknesses": [
256 "Limited to documents in this collection",
257 "Smaller result pool than full library search",
258 ],
259 "reliability": "High - searches a specific collection",
260 }
262 logger.info(
263 f"Registered {len(collections)} document collections as search engines"
264 )
265 else:
266 logger.debug(
267 "No username available for collection registration"
268 )
269 except Exception:
270 logger.warning("Could not register document collections")
272 return search_engines
275def get_available_engines(
276 settings_snapshot: Optional[Dict[str, Any]] = None,
277 use_api_key_services: bool = True,
278 exclude_engines: Optional[set] = None,
279) -> Dict[str, Any]:
280 """
281 Return search engines that are actually usable: enabled for auto-search
282 and with valid API keys when required.
284 This is the single shared filter used by the langgraph-agent tool
285 builder so it agrees with the rest of the system on which engines are
286 available.
288 Args:
289 settings_snapshot: Thread-safe settings snapshot.
290 use_api_key_services: If False, engines that require an API key are
291 excluded even when the key is present.
292 exclude_engines: Additional engine names to skip (e.g. the caller's
293 own name).
295 Returns:
296 Dict of engine_name → config for engines that passed all checks.
297 """
298 if not settings_snapshot: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 logger.warning(
300 "get_available_engines called without settings_snapshot, "
301 "returning empty dict"
302 )
303 return {}
305 all_engines = search_config(settings_snapshot=settings_snapshot)
306 excluded = set(exclude_engines) if exclude_engines else set()
308 available: Dict[str, Any] = {}
310 for name, config in all_engines.items():
311 if name in excluded: 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true
312 continue
314 # Check use_in_auto_search setting (default False)
315 auto_search_key = f"search.engine.web.{name}.use_in_auto_search"
316 use_in_auto = get_setting_from_snapshot(
317 auto_search_key, False, settings_snapshot=settings_snapshot
318 )
319 if not use_in_auto:
320 continue
322 requires_key = config.get("requires_api_key", False)
324 # Honour the use_api_key_services flag
325 if requires_key and not use_api_key_services: 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true
326 continue
328 # Validate the API key is actually present
329 if requires_key:
330 api_key = _resolve_api_key(name, config, settings_snapshot)
331 if not api_key: 331 ↛ 337line 331 didn't jump to line 337 because the condition on line 331 was always true
332 logger.debug(
333 f"Skipping {name} — requires API key but none configured"
334 )
335 continue
337 available[name] = config
339 return available
342def _resolve_api_key(
343 engine_name: str,
344 engine_config: Dict[str, Any],
345 settings_snapshot: Dict[str, Any],
346) -> Optional[str]:
347 """
348 Try to find a valid API key for *engine_name*.
350 Resolution order (mirrors ``create_search_engine``):
351 1. ``search.engine.web.<name>.api_key`` in the snapshot
352 2. ``api_key`` inside the engine config dict
354 Returns the key string or None.
355 """
356 api_key = None
357 api_key_path = f"search.engine.web.{engine_name}.api_key"
359 api_key_setting = settings_snapshot.get(api_key_path)
360 if api_key_setting: 360 ↛ 367line 360 didn't jump to line 367 because the condition on line 360 was always true
361 api_key = (
362 api_key_setting.get("value")
363 if isinstance(api_key_setting, dict)
364 else api_key_setting
365 )
367 if not api_key: 367 ↛ 370line 367 didn't jump to line 370 because the condition on line 367 was always true
368 api_key = engine_config.get("api_key")
370 if not api_key: 370 ↛ 374line 370 didn't jump to line 374 because the condition on line 370 was always true
371 return None
373 # Reject common placeholder values
374 api_key_str = str(api_key).strip()
375 if _is_api_key_placeholder(api_key_str):
376 return None
378 return api_key_str
381def default_search_engine(
382 username: Optional[str] = None,
383 db_session: Optional[Session] = None,
384 settings_snapshot: Optional[Dict[str, Any]] = None,
385) -> str:
386 """
387 Returns the configured default search engine.
389 Args:
390 username: Username for backward compatibility (deprecated)
391 db_session: Database session for direct access (preferred for web routes)
392 settings_snapshot: Settings snapshot for thread context (preferred for background threads)
394 Returns:
395 The configured default search engine.
396 """
397 return str(
398 _get_setting(
399 "search.engine.DEFAULT_SEARCH_ENGINE",
400 "wikipedia",
401 db_session=db_session,
402 settings_snapshot=settings_snapshot,
403 username=username,
404 )
405 )