Coverage for src/local_deep_research/web_search_engines/search_engines_config.py: 92%
163 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 ..security.log_sanitizer import scrub_error
13from ..utilities.db_utils import get_settings_manager
14from .search_engine_base import _is_api_key_placeholder
17def _get_setting(
18 key: str,
19 default_value: Any = None,
20 db_session: Optional[Session] = None,
21 settings_snapshot: Optional[Dict[str, Any]] = None,
22 username: Optional[str] = None,
23) -> Any:
24 """
25 Get a setting from either a database session or settings snapshot.
27 Args:
28 key: The setting key
29 default_value: Default value if setting not found
30 db_session: Database session for direct access
31 settings_snapshot: Settings snapshot for thread context
32 username: Username for backward compatibility
34 Returns:
35 The setting value or default_value if not found
36 """
37 # Try settings snapshot first (thread context)
38 if settings_snapshot:
39 try:
40 return get_setting_from_snapshot(
41 key, default_value, settings_snapshot=settings_snapshot
42 )
43 except Exception as e:
44 logger.debug(
45 f"Could not get setting {key} from snapshot: {scrub_error(e)}"
46 )
48 # Try database session if available
49 if db_session:
50 try:
51 settings_manager = get_settings_manager(db_session, username)
52 return settings_manager.get_setting(key, default_value)
53 except Exception as e:
54 logger.debug(
55 f"Could not get setting {key} from db_session: {scrub_error(e)}"
56 )
58 # Return default if all methods fail
59 logger.warning(
60 f"Could not retrieve setting '{key}', returning default: {default_value}"
61 )
62 return default_value
65def _extract_per_engine_config(
66 raw_config: Dict[str, Any],
67) -> Dict[str, Dict[str, Any]]:
68 """
69 Converts the "flat" configuration loaded from the settings database into
70 individual settings dictionaries for each engine.
72 Args:
73 raw_config: The raw "flat" configuration.
75 Returns:
76 Configuration dictionaries indexed by engine name.
78 """
79 nested_config: dict[str, Any] = {}
80 for key, value in raw_config.items():
81 if "." in key:
82 # This is a higher-level key.
83 top_level_key = key.split(".")[0]
84 lower_keys = ".".join(key.split(".")[1:])
85 nested_config.setdefault(top_level_key, {})[lower_keys] = value
86 else:
87 # This is a low-level key.
88 nested_config[key] = value
90 # Expand all the lower-level keys.
91 for key, value in nested_config.items():
92 if isinstance(value, dict):
93 # Expand the child keys.
94 nested_config[key] = _extract_per_engine_config(value)
96 return nested_config
99def search_config(
100 username: Optional[str] = None,
101 db_session: Optional[Session] = None,
102 settings_snapshot: Optional[Dict[str, Any]] = None,
103) -> Dict[str, Any]:
104 """
105 Returns the search engine configuration loaded from the database or settings snapshot.
107 Args:
108 username: Username used to scope the per-user retriever listing
109 (own registrations plus shared ones) so one user's retrievers
110 never leak into another user's engine list. Falls back to the
111 ``_username`` carried in ``settings_snapshot`` when omitted.
112 db_session: Database session for direct access (preferred for web routes)
113 settings_snapshot: Settings snapshot for thread context (preferred for background threads)
115 Returns:
116 The search engine configuration loaded from the database or snapshot.
117 """
118 # Extract search engine definitions
119 config_data = _get_setting(
120 "search.engine.web",
121 {},
122 db_session=db_session,
123 settings_snapshot=settings_snapshot,
124 username=username,
125 )
127 search_engines = _extract_per_engine_config(config_data)
129 # Inject module/class from the hardcoded engine registry.
130 # This is the single source of truth for which Python module implements
131 # each engine — these values are never read from the settings DB.
132 from .engine_registry import ENGINE_REGISTRY
134 for name, entry in ENGINE_REGISTRY.items():
135 if name in search_engines:
136 search_engines[name]["module_path"] = entry.module_path
137 search_engines[name]["class_name"] = entry.class_name
138 if entry.full_search_module:
139 search_engines[name]["full_search_module"] = (
140 entry.full_search_module
141 )
142 search_engines[name]["full_search_class"] = (
143 entry.full_search_class
144 )
146 # Add registered retrievers as available search engines. Scope the
147 # listing to the requesting user (their own registrations plus shared
148 # ones) so one user's retriever names/metadata never leak into another
149 # user's engine list. Prefer an explicit username, else the ``_username``
150 # carried in the snapshot; None lists shared retrievers only.
151 from ..search_system import username_from_snapshot
153 effective_username = username or username_from_snapshot(settings_snapshot)
154 from .retriever_registry import retriever_registry
156 for name in retriever_registry.list_registered(username=effective_username):
157 search_engines[name] = {
158 "module_path": ".engines.search_engine_retriever",
159 "class_name": "RetrieverSearchEngine",
160 "requires_api_key": False,
161 "requires_llm": False,
162 "description": f"LangChain retriever: {name}",
163 "strengths": [
164 "Domain-specific knowledge",
165 "No rate limits",
166 "Fast retrieval",
167 ],
168 "weaknesses": ["Limited to indexed content"],
169 "supports_full_search": True,
170 "is_retriever": True, # Mark as retriever for identification
171 }
173 logger.info(
174 f"Loaded {len(search_engines)} search engines from configuration file"
175 )
176 logger.info(f"\n {', '.join(sorted(search_engines.keys()))} \n")
178 # Register Library RAG as a search engine
179 library_enabled = _get_setting(
180 "search.engine.library.enabled",
181 True,
182 db_session=db_session,
183 settings_snapshot=settings_snapshot,
184 username=username,
185 )
187 if library_enabled:
188 search_engines["library"] = {
189 "module_path": ".engines.search_engine_library",
190 "class_name": "LibraryRAGSearchEngine",
191 "requires_llm": True,
192 "display_name": "Search All Collections",
193 "default_params": {},
194 "description": "Search across all your document collections using semantic search",
195 "strengths": [
196 "Searches all your curated collections of research papers and documents",
197 "Uses semantic search for better relevance",
198 "Returns documents you've already saved and reviewed",
199 ],
200 "weaknesses": [
201 "Limited to documents already in your collections",
202 "Requires documents to be indexed first",
203 ],
204 "reliability": "High - searches all your collections",
205 }
206 logger.info("Registered Library RAG as search engine")
208 # Register document collections as individual search engines
209 if library_enabled:
210 try:
211 from ..database.models.library import Collection
212 from ..database.session_context import get_user_db_session
213 from ..search_system import username_from_snapshot
215 # Get username from settings_snapshot if available
216 collection_username = (
217 username_from_snapshot(settings_snapshot)
218 if settings_snapshot
219 else username
220 )
222 if collection_username:
223 with get_user_db_session(collection_username) as session:
224 collections = session.query(Collection).all()
226 for collection in collections:
227 engine_id = f"collection_{collection.id}"
228 # Add suffix to distinguish from the all-collections search
229 display_name = f"{collection.name} (Collection)"
230 # Egress classification follows the per-collection
231 # public/private flag (default private). A "public"
232 # collection counts as a public engine (allowed under
233 # PUBLIC_ONLY); a private one is local-only. NULL
234 # (pre-migration rows) reads as private — the safe
235 # default.
236 collection_is_public = bool(
237 getattr(collection, "is_public", False)
238 )
239 # Usability flag (NOT egress): whether the LangGraph
240 # research agent offers this collection as a tool. NULL
241 # (pre-migration rows) reads as available (True). Uses
242 # the same `is not False` idiom as the rag_routes
243 # serializers so all call sites share one NULL→available
244 # default and can't drift.
245 collection_agent_enabled = (
246 getattr(collection, "agent_enabled", True)
247 is not False
248 )
249 search_engines[engine_id] = {
250 "module_path": ".engines.search_engine_collection",
251 "class_name": "CollectionSearchEngine",
252 "requires_llm": True,
253 "is_local": True,
254 "is_public": collection_is_public,
255 "agent_enabled": collection_agent_enabled,
256 "display_name": display_name,
257 "default_params": {
258 "collection_id": collection.id,
259 "collection_name": collection.name,
260 },
261 "description": (
262 collection.description
263 if collection.description
264 else f"Search documents in {collection.name} collection only"
265 ),
266 "strengths": [
267 f"Searches only documents in {collection.name}",
268 "Focused semantic search within specific topic area",
269 "Returns documents from a curated collection",
270 ],
271 "weaknesses": [
272 "Limited to documents in this collection",
273 "Smaller result pool than full library search",
274 ],
275 "reliability": "High - searches a specific collection",
276 }
278 logger.info(
279 f"Registered {len(collections)} document collections as search engines"
280 )
281 else:
282 logger.debug(
283 "No username available for collection registration"
284 )
285 except Exception:
286 logger.warning("Could not register document collections")
288 return search_engines
291def list_eligible_engine_configs(
292 settings_snapshot: Optional[Dict[str, Any]] = None,
293 *,
294 use_api_key_services: bool = True,
295 egress_context: Optional[Any] = None,
296 check_agent_enabled: bool = False,
297 check_auto_search: bool = False,
298) -> Dict[str, Any]:
299 """Return every engine from ``search_config`` that can be instantiated.
301 This is the candidate pool for surfaces that decide visibility
302 through their own flag (``agent_enabled`` for the langgraph research
303 agent) — NOT for surfaces governed by ``use_in_auto_search``. Engines
304 that need a key are dropped when no key is set (or when
305 ``use_api_key_services`` is False), so callers don't have to repeat the
306 credential check; everything else (retrievers, the built-in ``library``
307 engine, dynamic ``collection_*`` entries) flows through.
309 The name keys in the returned dict are the canonical engine names from
310 ``search_config`` (e.g. ``searxng``, ``elasticsearch``) — matching the
311 registry in ``engine_registry.py`` 1:1, so callers can compare against
312 ``search.tool`` directly.
314 Engines whose ``is_available(settings_snapshot)`` returns ``False`` are
315 excluded — this catches the case where the engine is configured but its
316 backing service is unreachable (e.g. Elasticsearch with no running
317 cluster on ``localhost:9200``). Without this filter the agent still
318 SEES the broken engine as a tool in its heartbeat and the factory logs
319 ``Failed to create search engine '<name>' (ConnectionError)`` per step.
320 Subclasses of ``BaseSearchEngine`` override ``is_available`` to opt in;
321 the default is True so existing engines are unaffected.
323 Filters for disabled state (``agent_enabled`` / ``use_in_auto_search``) and
324 egress policy are evaluated BEFORE calling ``is_available`` so unreachable or
325 disabled engines do not trigger unnecessary network probes.
327 Args:
328 settings_snapshot: Thread-safe settings snapshot.
329 use_api_key_services: When False, engines that require an API key
330 are excluded even when the key is present.
331 egress_context: Optional EgressContext to evaluate policy before probing.
332 check_agent_enabled: When True, excludes engines disabled for agent.
333 check_auto_search: When True, excludes engines disabled for auto-search.
335 Returns:
336 Dict of engine_name → config for engines that passed all checks.
337 """
338 if not settings_snapshot:
339 logger.warning(
340 "list_eligible_engine_configs called without settings_snapshot, "
341 "returning empty dict"
342 )
343 return {}
345 all_engines = search_config(settings_snapshot=settings_snapshot)
347 if egress_context is not None:
348 from ..security.egress.policy import (
349 evaluate_engine,
350 evaluate_retriever,
351 )
353 eligible: Dict[str, Any] = {}
354 for name, config in all_engines.items():
355 requires_key = config.get("requires_api_key", False)
357 if requires_key and not use_api_key_services:
358 continue
359 if requires_key:
360 api_key = _resolve_api_key(name, config, settings_snapshot)
361 if not api_key:
362 logger.debug(
363 f"Skipping {name} — requires API key but none configured"
364 )
365 continue
367 # Optional auto-search flag check prior to probe
368 if check_auto_search:
369 use_in_auto = config.get("use_in_auto_search")
370 if use_in_auto is None: 370 ↛ 375line 370 didn't jump to line 375 because the condition on line 370 was always true
371 auto_search_key = f"search.engine.web.{name}.use_in_auto_search"
372 use_in_auto = get_setting_from_snapshot(
373 auto_search_key, False, settings_snapshot=settings_snapshot
374 )
375 if not use_in_auto:
376 continue
378 # Optional agent_enabled flag check prior to probe
379 if check_agent_enabled:
380 agent_enabled = config.get("agent_enabled")
381 if agent_enabled is None:
382 agent_enabled = get_setting_from_snapshot(
383 f"search.engine.web.{name}.agent_enabled",
384 True,
385 settings_snapshot=settings_snapshot,
386 )
387 if not agent_enabled:
388 logger.debug(f"Skipping {name} — disabled for research agent")
389 continue
391 # Egress policy check prior to network probe
392 if egress_context is not None:
393 if config.get("is_retriever"): 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true
394 try:
395 from .retriever_registry import retriever_registry
397 meta = retriever_registry.get_metadata(
398 name,
399 username=getattr(egress_context, "username", None),
400 )
401 except Exception:
402 meta = None
403 decision = evaluate_retriever(
404 name, egress_context, metadata=meta
405 )
406 else:
407 decision = evaluate_engine(
408 name,
409 egress_context,
410 settings_snapshot=settings_snapshot,
411 metadata=config,
412 )
413 if not decision.allowed:
414 logger.debug(
415 f"Skipping {name} — disallowed by egress policy ({decision.reason})"
416 )
417 continue
419 # Runtime availability probe. Cheap (cached, single TCP connect at
420 # worst) and excludes engines whose backing service isn't reachable
421 # so the agent doesn't waste steps advertising broken tools.
422 # ``is_retriever`` entries go through the retriever registry's own
423 # ``is_available`` path — skip them here so we don't double-probe
424 # or fall back to the base class default for custom retriever
425 # classes that never subclass ``BaseSearchEngine``.
426 if not config.get("is_retriever", False): 426 ↛ 433line 426 didn't jump to line 433 because the condition on line 426 was always true
427 if not _engine_class_is_available(name, config, settings_snapshot):
428 logger.debug(
429 f"Skipping {name} — runtime availability probe failed"
430 )
431 continue
433 eligible[name] = config
435 return eligible
438def _engine_class_is_available(
439 name: str,
440 config: Dict[str, Any],
441 settings_snapshot: Dict[str, Any],
442) -> bool:
443 """Load the engine class for ``config`` and call its ``is_available``.
445 Returns True (fail-open) on any error — the class load can legitimately
446 fail (bad module_path, missing dependency, sandbox import restrictions)
447 and we don't want ``list_eligible_engine_configs`` to mask real config
448 problems with an unrelated "engine unavailable" log line. The factory
449 will surface the real ``Failed to create search engine`` error when
450 (and only when) the caller actually tries to instantiate it.
451 """
452 from .search_engine_base import BaseSearchEngine
454 module_path = config.get("module_path")
455 class_name = config.get("class_name")
456 if not module_path or not class_name:
457 return True
459 try:
460 from ..security.module_whitelist import get_safe_module_class
462 engine_class = get_safe_module_class(module_path, class_name)
463 except Exception as exc:
464 logger.debug(
465 "Skipping availability probe for {} — could not load engine "
466 "class ({}: {})",
467 name,
468 type(exc).__name__,
469 exc,
470 )
471 return True
473 # Only subclasses of BaseSearchEngine get the ``is_available`` contract.
474 # Library / Collection / Retriever engines either subclass it (in which
475 # case the default ``True`` applies) or don't (also default True).
476 if not isinstance(engine_class, type) or not issubclass( 476 ↛ 479line 476 didn't jump to line 479 because the condition on line 476 was never true
477 engine_class, BaseSearchEngine
478 ):
479 return True
481 try:
482 return bool(
483 engine_class.is_available(settings_snapshot=settings_snapshot)
484 )
485 except Exception as exc:
486 logger.debug(
487 "is_available raised for {} — treating as available ({})",
488 name,
489 type(exc).__name__,
490 )
491 return True
494def get_available_engines(
495 settings_snapshot: Optional[Dict[str, Any]] = None,
496 use_api_key_services: bool = True,
497 exclude_engines: Optional[set] = None,
498) -> Dict[str, Any]:
499 """
500 Return search engines that are actually usable: enabled for auto-search
501 and with valid API keys when required.
503 This is the single shared filter used by ``auto``-mode search so the
504 LangGraph agent agrees with the rest of the system on which engines
505 show up in non-agent searches. For the agent's specialised tool list,
506 use :func:`list_eligible_engine_configs` instead — the agent's surface
507 is governed by per-engine ``agent_enabled``, not by this auto-search
508 filter (see #5015).
510 Args:
511 settings_snapshot: Thread-safe settings snapshot.
512 use_api_key_services: If False, engines that require an API key are
513 excluded even when the key is present.
514 exclude_engines: Additional engine names to skip (e.g. the caller's
515 own name).
517 Returns:
518 Dict of engine_name → config for engines that passed all checks.
519 """
520 eligible = list_eligible_engine_configs(
521 settings_snapshot=settings_snapshot,
522 use_api_key_services=use_api_key_services,
523 check_auto_search=True,
524 )
525 excluded = set(exclude_engines) if exclude_engines else set()
527 return {
528 name: config
529 for name, config in eligible.items()
530 if name not in excluded
531 }
534def _resolve_api_key(
535 engine_name: str,
536 engine_config: Dict[str, Any],
537 settings_snapshot: Dict[str, Any],
538) -> Optional[str]:
539 """
540 Try to find a valid API key for *engine_name*.
542 Resolution order (mirrors ``create_search_engine``):
543 1. ``search.engine.web.<name>.api_key`` in the snapshot
544 2. ``api_key`` inside the engine config dict
546 Returns the key string or None.
547 """
548 api_key = None
549 api_key_path = f"search.engine.web.{engine_name}.api_key"
551 api_key_setting = settings_snapshot.get(api_key_path)
552 if api_key_setting:
553 api_key = (
554 api_key_setting.get("value")
555 if isinstance(api_key_setting, dict)
556 else api_key_setting
557 )
559 if not api_key:
560 api_key = engine_config.get("api_key")
562 if not api_key:
563 return None
565 # Reject common placeholder values
566 api_key_str = str(api_key).strip()
567 if _is_api_key_placeholder(api_key_str): 567 ↛ 568line 567 didn't jump to line 568 because the condition on line 567 was never true
568 return None
570 return api_key_str
573def default_search_engine(
574 username: Optional[str] = None,
575 db_session: Optional[Session] = None,
576 settings_snapshot: Optional[Dict[str, Any]] = None,
577) -> str:
578 """
579 Returns the configured default search engine.
581 Args:
582 username: Username for backward compatibility (deprecated)
583 db_session: Database session for direct access (preferred for web routes)
584 settings_snapshot: Settings snapshot for thread context (preferred for background threads)
586 Returns:
587 The configured default search engine.
588 """
589 return str(
590 _get_setting(
591 "search.engine.DEFAULT_SEARCH_ENGINE",
592 "wikipedia",
593 db_session=db_session,
594 settings_snapshot=settings_snapshot,
595 username=username,
596 )
597 )