Coverage for src/qdrant_loader_mcp_server/search/hybrid/orchestration/facets.py: 100%

1from __future__ import annotations 

2 

3from datetime import datetime 

4from typing import Any 

5 

6from ...enhanced.faceted_search import FacetedSearchResults, FacetFilter 

7 

8 

9async def search_with_facets( 

10 engine: Any, 

11 query: str, 

12 limit: int = 5, 

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

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

15 facet_filters: list[FacetFilter] | None = None, 

16 generate_facets: bool = True, 

17 session_context: dict[str, Any] | None = None, 

18 behavioral_context: list[str] | None = None, 

19) -> FacetedSearchResults: 

20 """Perform faceted search with dynamic facet generation using the provided engine.""" 

21 start_time = datetime.now() 

22 

23 # First, perform regular search (potentially with larger limit for faceting) 

24 search_limit = max(limit * 2, 50) if generate_facets else limit 

25 

26 search_results = await engine.search( 

27 query=query, 

28 limit=search_limit, 

29 source_types=source_types, 

30 project_ids=project_ids, 

31 session_context=session_context, 

32 behavioral_context=behavioral_context, 

33 ) 

34 

35 # Generate faceted results 

36 faceted_results = engine.faceted_search_engine.generate_faceted_results( 

37 results=search_results, applied_filters=facet_filters or [] 

38 ) 

39 

40 # Limit final results 

41 faceted_results.results = faceted_results.results[:limit] 

42 faceted_results.filtered_count = len(faceted_results.results) 

43 

44 search_time = (datetime.now() - start_time).total_seconds() * 1000 

45 

46 engine.logger.info( 

47 "Faceted search completed", 

48 query=query, 

49 total_results=faceted_results.total_results, 

50 filtered_results=faceted_results.filtered_count, 

51 facet_count=len(faceted_results.facets), 

52 active_filters=len(faceted_results.applied_filters), 

53 search_time_ms=round(search_time, 2), 

54 ) 

55 

56 return faceted_results