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

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.diagnostics import ( 

13 DiagnosticLog, 

14 DiagnosticStatus, 

15 diagnostic_for_http_error, 

16) 

17from secchi.http import HttpClientFactory 

18from secchi.models import Registry, SearchResult 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23class PackageSearchService: 

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

25 

26 def __init__(self, diagnostics: DiagnosticLog | None = None) -> None: 

27 self.diagnostics = diagnostics 

28 

29 async def search( 

30 self, 

31 query: str, 

32 *, 

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

34 limit: int = 10, 

35 diagnostics: DiagnosticLog | None = None, 

36 ) -> list[SearchResult]: 

37 selected = registries or list(Registry) 

38 diagnostics = diagnostics or self.diagnostics 

39 failed_registries: set[Registry] = set() 

40 

41 async with HttpClientFactory(diagnostics=diagnostics).create() as client: 

42 

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

44 try: 

45 try: 

46 adapter = create_adapter(registry, client=client) 

47 except TypeError: 

48 adapter = create_adapter(registry) 

49 results = await adapter.search(query, limit=limit) 

50 if diagnostics is not None: 

51 diagnostics.record( 

52 DiagnosticStatus.SUCCESS, 

53 registry.display_name, 

54 f"Search completed ({len(results)} result(s))", 

55 ) 

56 return results 

57 except ( 

58 httpx.HTTPError, 

59 OSError, 

60 ValueError, 

61 KeyError, 

62 TypeError, 

63 ) as exc: 

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

65 failed_registries.add(registry) 

66 if diagnostics is not None: 

67 diagnostics.record( 

68 DiagnosticStatus.WARN, 

69 registry.display_name, 

70 f"Search unavailable: {diagnostic_for_http_error(exc)}", 

71 ) 

72 logger.debug( 

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

74 registry.value, 

75 exc, 

76 exc_info=True, 

77 ) 

78 return [] 

79 

80 batches = await asyncio.gather( 

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

82 ) 

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

84 if ( 

85 diagnostics is not None 

86 and selected 

87 and len(failed_registries) == len(selected) 

88 ): 

89 diagnostics.record( 

90 DiagnosticStatus.FAILURE, 

91 "SEARCH", 

92 f"All {len(selected)} selected registries failed", 

93 ) 

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

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

96 

97 @staticmethod 

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

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

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

101 # scores while preserving useful ordering within a registry. 

102 normalized_score = ( 

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

104 ) 

105 return ( 

106 exact, 

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

108 -normalized_score, 

109 result.name.casefold(), 

110 )