Coverage for src/secchi/services/search.py: 100%

1"""Cross-registry package discovery and deterministic result ranking.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import logging 

7import math 

8 

9import httpx 

10 

11from secchi.api.base import create_adapter 

12from secchi.http import HttpClientFactory 

13from secchi.models import Registry, SearchResult 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18class PackageSearchService: 

19 """Search configured registries concurrently and normalize their results.""" 

20 

21 async def search( 

22 self, 

23 query: str, 

24 *, 

25 registries: list[Registry] | None = None, 

26 limit: int = 10, 

27 ) -> list[SearchResult]: 

28 selected = registries or list(Registry) 

29 

30 async with HttpClientFactory().create() as client: 

31 

32 async def search_registry(registry: Registry) -> list[SearchResult]: 

33 try: 

34 try: 

35 adapter = create_adapter(registry, client=client) 

36 except TypeError: 

37 adapter = create_adapter(registry) 

38 return await adapter.search(query, limit=limit) 

39 except ( 

40 httpx.HTTPError, 

41 OSError, 

42 ValueError, 

43 KeyError, 

44 TypeError, 

45 ) as exc: 

46 # One unavailable registry should not hide results from the others. 

47 logger.debug( 

48 "Registry search failed for %s: %s", 

49 registry.value, 

50 exc, 

51 exc_info=True, 

52 ) 

53 return [] 

54 

55 batches = await asyncio.gather( 

56 *(search_registry(registry) for registry in selected) 

57 ) 

58 results = [result for batch in batches for result in batch] 

59 results.sort(key=lambda result: self._sort_key(result, query)) 

60 return results[: limit * len(selected)] 

61 

62 @staticmethod 

63 def _sort_key(result: SearchResult, query: str) -> tuple[int, int, float, str]: 

64 exact = 0 if result.exact or result.name.casefold() == query.casefold() else 1 

65 # Registry APIs use incompatible score scales. Compress large download 

66 # scores while preserving useful ordering within a registry. 

67 normalized_score = ( 

68 math.log10(result.score + 1) if result.score > 1 else result.score 

69 ) 

70 return ( 

71 exact, 

72 0 if result.name.casefold().startswith(query.casefold()) else 1, 

73 -normalized_score, 

74 result.name.casefold(), 

75 )