Coverage for src/qdrant_loader_mcp_server/search/engine/search.py: 95%

1""" 

2Search Operations - Main Search Functionality. 

3 

4This module implements the core search operations including 

5basic document search using the hybrid search engine. 

6""" 

7 

8from typing import TYPE_CHECKING 

9 

10if TYPE_CHECKING: 

11 from .core import SearchEngine 

12 

13from ...utils.logging import LoggingConfig 

14from ..components.models.hybrid import HybridSearchResult 

15 

16logger = LoggingConfig.get_logger(__name__) 

17 

18 

19class SearchOperations: 

20 """Handles core search operations.""" 

21 

22 def __init__(self, engine: "SearchEngine"): 

23 """Initialize with search engine reference.""" 

24 self.engine = engine 

25 

26 async def search( 

27 self, 

28 query: str, 

29 source_types: list[str] | None = None, 

30 limit: int = 5, 

31 project_ids: list[str] | None = None, 

32 ) -> list[HybridSearchResult]: 

33 """Search for documents using hybrid search. 

34 

35 Args: 

36 query: Search query text 

37 source_types: Optional list of source types to filter by 

38 limit: Maximum number of results to return 

39 project_ids: Optional list of project IDs to filter by 

40 """ 

41 hybrid = getattr(self.engine, "hybrid_search", None) 

42 if not hybrid: 

43 raise RuntimeError("Search engine not initialized") 

44 

45 logger.debug( 

46 "Performing search", 

47 query=query, 

48 source_types=source_types, 

49 limit=limit, 

50 project_ids=project_ids, 

51 ) 

52 

53 try: 

54 results = await hybrid.search( 

55 query=query, 

56 source_types=source_types, 

57 limit=limit, 

58 project_ids=project_ids, 

59 ) 

60 

61 logger.info( 

62 "Search completed", 

63 query=query, 

64 result_count=len(results), 

65 project_ids=project_ids, 

66 ) 

67 

68 return results 

69 except Exception as e: 

70 logger.error("Search failed", error=str(e), query=query) 

71 raise