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

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

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.core.di import ( 

9 ContainerRegistrarProtocol, 

10 ContainerResolverProtocol, 

11 ) 

12 

13from lexigram.contracts.search import SearchEngineProtocol 

14 

15 

16class _NoOpSearch: 

17 async def search( 

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

19 ) -> dict[str, Any]: 

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

21 

22 

23class SearchIntegration: 

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

25 

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

27 integration is disabled. 

28 """ 

29 

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

31 self._config = config 

32 self._search: Any = None 

33 self._enabled = False 

34 

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

36 from lexigram.admin.config import SearchIntegrationConfig 

37 from lexigram.admin.integrations._optional import is_installed 

38 

39 cfg = self._config 

40 if not isinstance(cfg, SearchIntegrationConfig): 

41 cfg = SearchIntegrationConfig() 

42 if not cfg.enabled: 

43 self._search = _NoOpSearch() 

44 return 

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

46 self._search = _NoOpSearch() 

47 return 

48 self._enabled = True 

49 

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

51 if not self._enabled: 

52 return 

53 self._container = container 

54 

55 async def shutdown(self) -> None: 

56 pass 

57 

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

59 return {"status": "healthy" if self.is_available else "noop"} 

60 

61 async def query( 

62 self, 

63 index: str, 

64 query_str: str, 

65 limit: int = 50, 

66 offset: int = 0, 

67 filters: dict[str, Any] | None = None, 

68 rule: str | None = None, 

69 ) -> dict[str, Any]: 

70 """Run a query against the search index. 

71 

72 Args: 

73 index: Index name to search. 

74 query_str: Full-text query string. 

75 limit: Maximum results to return. 

76 offset: Result offset for pagination. 

77 filters: Canonical search filter dict (``{field: value}``, 

78 ``{field: {"op": value}}``, ``{"$or": [...]}``, ...). 

79 rule: Query-builder block JSON string; merged into *filters* 

80 with AND semantics by the engine backend. 

81 

82 Returns: 

83 A ``{"results": [...], "total": n}`` dict. 

84 """ 

85 engine = await self._get_engine() 

86 if isinstance(engine, _NoOpSearch): 

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

88 result = await engine.search( 

89 index_name=index, 

90 query=query_str, 

91 filters=filters, 

92 rule=rule, 

93 limit=limit, 

94 offset=offset, 

95 ) 

96 return self._unwrap(result) 

97 

98 async def _get_engine(self) -> Any: 

99 if self._search is not None: 

100 return self._search 

101 try: 

102 self._search = await self._container.resolve(SearchEngineProtocol) 

103 except Exception: 

104 self._search = _NoOpSearch() 

105 return self._search 

106 

107 @staticmethod 

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

109 if hasattr(result, "is_ok"): 

110 if not result.is_ok(): 

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

112 result = result.unwrap() 

113 if isinstance(result, dict): 

114 return { 

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

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

117 } 

118 return { 

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

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

121 } 

122 

123 @property 

124 def is_available(self) -> bool: 

125 """True when a real search engine is resolved (not the no-op).""" 

126 return self._search is not None and not isinstance(self._search, _NoOpSearch) 

127 

128 @property 

129 def fallback_to_like(self) -> bool: 

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

131 

132 

133__all__ = ["SearchIntegration"]