Coverage for src / lexigram / admin / services / search_service.py: 38%

1"""Global search service for cross-resource search.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from typing import Any 

7 

8 

9@dataclass 

10class SearchResult: 

11 """A single search result from any resource.""" 

12 

13 resource_name: str 

14 resource_label: str 

15 id: Any 

16 title: str 

17 subtitle: str = "" 

18 url: str = "" 

19 

20 

21@dataclass 

22class SearchResults: 

23 """Aggregated search results grouped by resource.""" 

24 

25 query: str 

26 total_count: int = 0 

27 results: list[SearchResult] = field(default_factory=list) 

28 resource_counts: dict[str, int] = field(default_factory=dict) 

29 

30 @property 

31 def has_results(self) -> bool: 

32 return self.total_count > 0 

33 

34 @property 

35 def group_count(self) -> int: 

36 return len(self.resource_counts) 

37 

38 

39class SearchService: 

40 """Searches across all registered admin resources. 

41 

42 Discovers resources, runs Resource.search() on each, aggregates 

43 and returns combined results. 

44 """ 

45 

46 def __init__(self, resource_manager: Any) -> None: 

47 self._resource_manager = resource_manager 

48 

49 async def search( 

50 self, 

51 query: str, 

52 *, 

53 limit: int = 5, 

54 per_resource: int = 5, 

55 ) -> SearchResults: 

56 """Search across all resources. 

57 

58 Args: 

59 query: The search query string. 

60 limit: Maximum total results. 

61 per_resource: Maximum results per resource. 

62 

63 Returns: 

64 Aggregated SearchResults. 

65 """ 

66 if not query or not query.strip(): 

67 return SearchResults(query=query) 

68 

69 query = query.strip() 

70 results = SearchResults(query=query) 

71 resources = self.get_searchable_resources() 

72 

73 for resource_cls in resources: 

74 try: 

75 items = await resource_cls.search(query, limit=per_resource) 

76 except Exception: # noqa: S112 

77 continue 

78 

79 if not items: 

80 continue 

81 

82 resource_name = getattr(resource_cls, "name", "") 

83 resource_label = getattr(resource_cls, "label", resource_name) 

84 resource_count = 0 

85 

86 for item in items: 

87 title = item.get("title", str(item.get("id", ""))) 

88 subtitle = item.get("subtitle", "") 

89 item_id = item.get("id", "") 

90 url = f"/admin/{resource_name}/{item_id}" 

91 

92 results.results.append( 

93 SearchResult( 

94 resource_name=resource_name, 

95 resource_label=resource_label, 

96 id=item_id, 

97 title=title, 

98 subtitle=subtitle, 

99 url=url, 

100 ) 

101 ) 

102 resource_count += 1 

103 

104 results.resource_counts[resource_name] = resource_count 

105 

106 results.total_count = len(results.results) 

107 if limit > 0 and results.total_count > limit: 

108 results.results = results.results[:limit] 

109 results.total_count = limit 

110 

111 return results 

112 

113 def get_searchable_resources(self) -> list[Any]: 

114 """Get list of resources with search_fields configured.""" 

115 resources: list[Any] = [] 

116 try: 

117 registered = self._resource_manager.get_all_resources() 

118 for r in registered: 

119 search_fields = getattr(r, "search_fields", None) or [] 

120 if search_fields: 

121 resources.append(r) 

122 except Exception: # noqa: S112 

123 pass 

124 return resources