Coverage for src / lexigram / contracts / search / protocols.py: 100%

1"""Search protocol definitions. 

2 

3Protocols for search engines, index management, searchable entities, 

4and search analytics. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

10 

11if TYPE_CHECKING: 

12 from lexigram.contracts.core import HealthCheckResult 

13 from lexigram.contracts.data import QueryResult 

14 

15 

16@runtime_checkable 

17class SearchEngineProtocol(Protocol): 

18 """Protocol for search engine backends.""" 

19 

20 async def search( 

21 self, 

22 query: str, 

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

24 sort: list[dict[str, str]] | None = None, 

25 limit: int | None = None, 

26 offset: int | None = None, 

27 ) -> QueryResult: 

28 """Execute a search query. 

29 

30 Args: 

31 query: Search query string 

32 filters: Search filters 

33 sort: Sort specifications 

34 limit: Maximum results to return 

35 offset: Results offset for pagination 

36 

37 Returns: 

38 Search results 

39 """ 

40 ... 

41 

42 async def index_document( 

43 self, 

44 document_id: str, 

45 document: dict[str, Any], 

46 index_name: str | None = None, 

47 ) -> None: 

48 """Index a document. 

49 

50 Args: 

51 document_id: Unique document identifier 

52 document: Document data to index 

53 index_name: Index name (optional) 

54 """ 

55 ... 

56 

57 async def index_many( 

58 self, 

59 documents: list[tuple[str, dict[str, Any]]], 

60 index_name: str | None = None, 

61 ) -> None: 

62 """Index multiple documents in a single bulk operation. 

63 

64 Callers should prefer this over N repeated :meth:`index_document` 

65 calls when processing batches. Backends may use a native bulk API 

66 to avoid per-document round-trip overhead. 

67 

68 Args: 

69 documents: Sequence of ``(document_id, document)`` pairs. 

70 index_name: Index name (optional, backend implementation chooses 

71 a default when ``None``). 

72 """ 

73 ... 

74 

75 async def delete_document( 

76 self, 

77 document_id: str, 

78 index_name: str | None = None, 

79 ) -> None: 

80 """Delete a document from index. 

81 

82 Args: 

83 document_id: Document identifier to delete 

84 index_name: Index name (optional) 

85 """ 

86 ... 

87 

88 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

89 """Perform health check. 

90 

91 Returns: 

92 Structured health check result. 

93 """ 

94 ... 

95 

96 

97@runtime_checkable 

98class IndexManagerProtocol(Protocol): 

99 """Protocol for managing search indices.""" 

100 

101 async def create_index( 

102 self, 

103 index_name: str, 

104 schema: dict[str, Any], 

105 ) -> None: 

106 """Create a search index. 

107 

108 Args: 

109 index_name: Name of the index to create 

110 schema: Index schema definition 

111 """ 

112 ... 

113 

114 async def delete_index(self, index_name: str) -> None: 

115 """Delete a search index. 

116 

117 Args: 

118 index_name: Name of the index to delete 

119 """ 

120 ... 

121 

122 async def get_index_info(self, index_name: str) -> dict[str, Any]: 

123 """Get information about an index. 

124 

125 Args: 

126 index_name: Name of the index 

127 

128 Returns: 

129 Index information 

130 """ 

131 ... 

132 

133 async def index_exists(self, index_name: str) -> bool: 

134 """Check whether an index exists. 

135 

136 Args: 

137 index_name: Name of the index to check. 

138 

139 Returns: 

140 ``True`` if the index exists, ``False`` otherwise. 

141 """ 

142 ... 

143 

144 

145@runtime_checkable 

146class SearchableProtocol(Protocol): 

147 """Protocol for entities that can be searched.""" 

148 

149 @property 

150 def search_document_id(self) -> str: 

151 """Unique identifier for search indexing.""" 

152 ... 

153 

154 @property 

155 def search_document(self) -> dict[str, Any]: 

156 """Document representation for search indexing.""" 

157 ... 

158 

159 @property 

160 def search_index_name(self) -> str: 

161 """Name of the search index for this entity.""" 

162 ... 

163 

164 

165@runtime_checkable 

166class SearchAnalyticsProtocol(Protocol): 

167 """Protocol for search analytics and metrics.""" 

168 

169 async def record_search( 

170 self, 

171 query: str, 

172 filters: dict[str, Any] | None, 

173 result_count: int, 

174 user_id: str | None = None, 

175 session_id: str | None = None, 

176 ) -> None: 

177 """Record a search query for analytics. 

178 

179 Args: 

180 query: Search query string 

181 filters: Applied filters 

182 result_count: Number of results returned 

183 user_id: User identifier (optional) 

184 session_id: Session identifier (optional) 

185 """ 

186 ... 

187 

188 async def get_search_metrics( 

189 self, 

190 time_range: dict[str, Any] | None = None, 

191 ) -> dict[str, Any]: 

192 """Get search analytics metrics. 

193 

194 Args: 

195 time_range: Time range for metrics (optional) 

196 

197 Returns: 

198 Search metrics data 

199 """ 

200 ... 

201 

202 

203@runtime_checkable 

204class DatabaseSearchBackendProtocol(Protocol): 

205 """Protocol for database-backed search backends.""" 

206 

207 async def connect(self) -> None: 

208 """Establish connection to the database.""" 

209 ... 

210 

211 async def close(self) -> None: 

212 """Close the database connection.""" 

213 ... 

214 

215 async def ensure_schema(self, index_name: str) -> None: 

216 """Ensure the search schema/tables exist for an index.""" 

217 ... 

218 

219 

220@runtime_checkable 

221class DocumentTransformerProtocol(Protocol): 

222 """Protocol for transforming documents for search indexing. 

223 

224 Transforms domain models into search-indexable documents with 

225 appropriate text extraction, field mapping, and metadata. 

226 

227 Example: 

228 ```python 

229 class MyDocumentTransformer: 

230 def transform(self, entity: MyEntity) -> dict[str, Any]: 

231 return { 

232 "id": entity.id, 

233 "title": entity.name, 

234 "content": self._extract_text(entity.description), 

235 "metadata": {"created_at": entity.created_at.isoformat()} 

236 } 

237 ``` 

238 """ 

239 

240 def transform(self, entity: Any) -> dict[str, Any]: 

241 """Transform a domain entity to a search document. 

242 

243 Args: 

244 entity: Domain entity to transform. 

245 

246 Returns: 

247 Search-indexable document dictionary. 

248 """ 

249 ... 

250 

251 def transform_batch(self, entities: list[Any]) -> list[dict[str, Any]]: 

252 """Transform multiple domain entities to search documents. 

253 

254 Args: 

255 entities: Domain entities to transform. 

256 

257 Returns: 

258 List of search-indexable document dictionaries. 

259 """ 

260 ... 

261 

262 def extract_text(self, content: Any) -> str: 

263 """Extract searchable text from content. 

264 

265 Args: 

266 content: Content to extract text from. 

267 

268 Returns: 

269 Extracted text string. 

270 """ 

271 ... 

272 

273 

274__all__ = [ 

275 "DatabaseSearchBackendProtocol", 

276 "DocumentTransformerProtocol", 

277 "IndexManagerProtocol", 

278 "SearchAnalyticsProtocol", 

279 "SearchEngineProtocol", 

280 "SearchableProtocol", 

281]