Coverage for src/lexigram/admin/services/search_service.py: 95%
129 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Global search service for cross-resource search."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from typing import TYPE_CHECKING, Any
8if TYPE_CHECKING:
9 from lexigram.contracts.auth import AuthorizerProtocol
12@dataclass
13class SearchResult:
14 """A single search result from any resource."""
16 resource_name: str
17 resource_label: str
18 id: Any
19 title: str
20 subtitle: str = ""
21 url: str = ""
24@dataclass
25class SearchResults:
26 """Aggregated search results grouped by resource."""
28 query: str
29 total_count: int = 0
30 results: list[SearchResult] = field(default_factory=list)
31 resource_counts: dict[str, int] = field(default_factory=dict)
33 @property
34 def has_results(self) -> bool:
35 return self.total_count > 0
37 @property
38 def group_count(self) -> int:
39 return len(self.resource_counts)
42class SearchService:
43 """Searches across all registered admin resources.
45 Discovers resources, runs Resource.search() on each, aggregates
46 and returns combined results. Resources that opted into indexed
47 search (via ``SearchableSpec``) are queried through the search
48 integration instead when one is available.
49 """
51 def __init__(
52 self,
53 resource_manager: Any,
54 authorizer: AuthorizerProtocol | None = None,
55 ) -> None:
56 self._resource_manager = resource_manager
57 self._authorizer = authorizer
59 async def search(
60 self,
61 query: str,
62 *,
63 limit: int = 5,
64 per_resource: int = 5,
65 rule: str | None = None,
66 allowed_resources: set[str] | None = None,
67 ) -> SearchResults:
68 """Search across all resources.
70 Args:
71 query: The search query string.
72 limit: Maximum total results.
73 per_resource: Maximum results per resource.
74 rule: Query-builder block JSON string applied to indexed
75 resources (ignored for LIKE-based resource search).
76 allowed_resources: Resource names the caller may view; results
77 from unlisted resources are skipped. None keeps
78 cross-resource behavior.
80 Returns:
81 Aggregated SearchResults.
82 """
83 if not query or not query.strip():
84 return SearchResults(query=query)
86 query = query.strip()
87 results = SearchResults(query=query)
88 resources = self.get_searchable_resources()
89 if allowed_resources is not None:
90 resources = [
91 r for r in resources if getattr(r, "name", "") in allowed_resources
92 ]
93 integration = self._get_search_integration()
95 for resource_cls in resources:
96 try:
97 spec = self._index_spec(resource_cls)
98 if (
99 spec is not None
100 and integration is not None
101 and integration.is_available
102 ):
103 items = await self._search_index(
104 integration,
105 resource_cls,
106 spec,
107 query,
108 per_resource,
109 rule,
110 )
111 else:
112 items = await resource_cls.search(query, limit=per_resource)
113 except Exception: # noqa: S112
114 continue
116 if not items:
117 continue
119 resource_name = getattr(resource_cls, "name", "")
120 resource_label = getattr(resource_cls, "label", resource_name)
121 resource_count = 0
123 for item in items:
124 title = item.get("title", str(item.get("id", "")))
125 subtitle = item.get("subtitle", "")
126 item_id = item.get("id", "")
127 url = f"/admin/{resource_name}/{item_id}"
129 results.results.append(
130 SearchResult(
131 resource_name=resource_name,
132 resource_label=resource_label,
133 id=item_id,
134 title=title,
135 subtitle=subtitle,
136 url=url,
137 )
138 )
139 resource_count += 1
141 results.resource_counts[resource_name] = resource_count
143 results.total_count = len(results.results)
144 if limit > 0 and results.total_count > limit:
145 results.results = results.results[:limit]
146 results.total_count = limit
148 return results
150 def get_searchable_resources(self) -> list[Any]:
151 """Get list of resources with search_fields or an index spec."""
152 resources: list[Any] = []
153 try:
154 registered = self._resource_manager.get_all_resources()
155 for r in registered:
156 search_fields = getattr(r, "search_fields", None) or []
157 if search_fields:
158 resources.append(r)
159 continue
160 spec = self._index_spec(r)
161 if spec is not None and spec.index_name:
162 resources.append(r)
163 except Exception: # noqa: S112, S110
164 pass
165 return resources
167 async def allowed_resources_for(self, user: Any) -> set[str] | None:
168 """Resolve the searchable resources the user may view.
170 Args:
171 user: Authenticated user, or None for anonymous requests.
173 Returns:
174 The set of viewable resource names, or None when there is no
175 permission context (no user, or no authorizer configured) —
176 callers then keep the unfiltered cross-resource behavior.
177 """
178 if user is None or self._authorizer is None:
179 return None
180 allowed: set[str] = set()
181 for resource in self.get_searchable_resources():
182 name = getattr(resource, "name", "")
183 if name and await self._authorizer.can_view(user, name):
184 allowed.add(name)
185 return allowed
187 def get_search_field_catalog(self) -> list[dict[str, Any]]:
188 """Build a query-builder field catalog from searchable resources.
190 Collects the searchable field names of every resource that is
191 searchable (``search_fields`` and/or ``SearchableSpec.fields``),
192 deduplicated, labeled for display.
194 Returns:
195 A list of ``{"name", "label"}`` catalog entries (possibly
196 empty when no resource exposes searchable fields).
197 """
198 catalog: list[dict[str, Any]] = []
199 try:
200 for resource in self.get_searchable_resources():
201 names: list[str] = list(getattr(resource, "search_fields", None) or [])
202 spec = self._index_spec(resource)
203 if spec is not None:
204 names += list(getattr(spec, "fields", None) or [])
205 for name in dict.fromkeys(names):
206 catalog.append(
207 {"name": name, "label": name.replace("_", " ").title()}
208 )
209 except Exception: # noqa: S112
210 return []
211 return catalog
213 @staticmethod
214 def _index_spec(resource: Any) -> Any | None:
215 """Return the resource's SearchableSpec, or None when not opted in."""
216 spec_fn = getattr(resource, "search_spec", None)
217 if not spec_fn:
218 return None
219 try:
220 return spec_fn()
221 except Exception: # noqa: S112
222 return None
224 @staticmethod
225 def _get_search_integration() -> Any:
226 """Return the registered SearchIntegration instance, or None."""
227 from lexigram.admin.integrations import get as get_integration
229 return get_integration("SearchIntegration")
231 async def _search_index(
232 self,
233 integration: Any,
234 resource: Any,
235 spec: Any,
236 query: str,
237 limit: int,
238 rule: str | None = None,
239 ) -> list[dict[str, Any]]:
240 """Query the search index for *resource* and shape docs into hits.
242 Index results carry the original document in ``SearchResult.data``
243 (the ``SearchableSpec.fields`` plus ``id``); ``search_title_field``/
244 ``name``/``title`` resolve the display title, mirroring
245 ``Resource.search()``'s hit shape. Backends that return plain dicts
246 are handled as well.
247 """
248 result = await integration.query(spec.index_name, query, limit=limit, rule=rule)
249 raw = result.get("results", []) if isinstance(result, dict) else []
250 title_field = getattr(resource, "search_title_field", "name")
251 hits: list[dict[str, Any]] = []
252 for item in raw:
253 doc: Any
254 if isinstance(item, dict):
255 item_id = item.get("id", "")
256 doc = item
257 else:
258 item_id = getattr(item, "id", None)
259 doc = getattr(item, "data", None)
260 if not item_id or not isinstance(doc, dict):
261 continue
262 title = (
263 doc.get(title_field)
264 or doc.get("name")
265 or doc.get("title")
266 or str(item_id)
267 )
268 subtitle = doc.get("email") or doc.get("description") or ""
269 hits.append({"id": str(item_id), "title": title, "subtitle": subtitle})
270 return hits