Coverage for src / lexigram / admin / integrations / search.py: 54%

1"""Search integration — delegates resource search to a search index.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING, Any 

7 

8if TYPE_CHECKING: 

9 from lexigram.contracts.core.di import ( 

10 ContainerRegistrarProtocol, 

11 ContainerResolverProtocol, 

12 ) 

13 

14 

15@dataclass(frozen=True) 

16class SearchableSpec: 

17 """Specification for search-index-based resource searching. 

18 

19 Attributes: 

20 index_name: Name of the search index (defaults to resource name). 

21 fields: Which fields to include in the searchable document body. 

22 result_limit: Max results per query. 

23 """ 

24 

25 index_name: str | None = None 

26 fields: tuple[str, ...] = () 

27 result_limit: int = 50 

28 

29 

30class _NoOpSearch: 

31 async def search( 

32 self, index_name: str, query: str, **kwargs: Any 

33 ) -> dict[str, Any]: 

34 return {"results": [], "total": 0} 

35 

36 

37class SearchIntegration: 

38 """Adapter that routes list queries through a search index. 

39 

40 Gracefully no-ops when ``lexigram-search`` is not installed or the 

41 integration is disabled. 

42 """ 

43 

44 def __init__(self, config: Any) -> None: 

45 self._config = config 

46 self._search: Any = None 

47 self._enabled = False 

48 

49 def register(self, container: ContainerRegistrarProtocol) -> None: 

50 from lexigram.admin.config import SearchIntegrationConfig 

51 from lexigram.admin.integrations._optional import is_installed 

52 

53 cfg = self._config 

54 if not isinstance(cfg, SearchIntegrationConfig): 

55 cfg = SearchIntegrationConfig() 

56 if not cfg.enabled: 

57 self._search = _NoOpSearch() 

58 return 

59 if not is_installed("lexigram.search"): 

60 self._search = _NoOpSearch() 

61 return 

62 self._enabled = True 

63 

64 async def boot(self, container: ContainerResolverProtocol) -> None: 

65 if not self._enabled: 

66 return 

67 self._container = container 

68 

69 async def shutdown(self) -> None: 

70 pass 

71 

72 async def health_check(self) -> dict[str, Any]: 

73 return { 

74 "status": "healthy" 

75 if self._search is not None and not isinstance(self._search, _NoOpSearch) 

76 else "noop" 

77 } 

78 

79 async def query( 

80 self, index: str, query_str: str, limit: int = 50, offset: int = 0 

81 ) -> dict[str, Any]: 

82 engine = await self._get_engine() 

83 if isinstance(engine, _NoOpSearch): 

84 return {"results": [], "total": 0} 

85 result = await engine.search( 

86 index_name=index, query=query_str, limit=limit, offset=offset 

87 ) 

88 return self._unwrap(result) 

89 

90 async def _get_engine(self) -> Any: 

91 if self._search is not None: 

92 return self._search 

93 try: 

94 from lexigram.search.engine import SearchEngine 

95 

96 self._search = await self._container.resolve(SearchEngine) 

97 except Exception: 

98 self._search = _NoOpSearch() 

99 return self._search 

100 

101 @staticmethod 

102 def _unwrap(result: Any) -> dict[str, Any]: 

103 if hasattr(result, "is_ok"): 

104 if not result.is_ok(): 

105 return {"results": [], "total": 0} 

106 result = result.unwrap() 

107 if isinstance(result, dict): 

108 return { 

109 "results": list(result.get("results", [])), 

110 "total": result.get("total", 0), 

111 } 

112 return { 

113 "results": list(result.results) if hasattr(result, "results") else [], 

114 "total": result.total if hasattr(result, "total") else 0, 

115 } 

116 

117 @property 

118 def fallback_to_like(self) -> bool: 

119 return getattr(self._config, "fallback_to_like", True) 

120 

121 

122__all__ = ["SearchIntegration", "SearchableSpec"]