Coverage for src/secchi/api/base.py: 69%

1"""Base protocol and factory for registry API adapters.""" 

2 

3from __future__ import annotations 

4 

5from typing import ClassVar, Protocol 

6 

7import httpx 

8 

9from secchi.models import ( 

10 Dependency, 

11 DownloadCounts, 

12 DownloadTrendPoint, 

13 PackageInfo, 

14 Registry, 

15 ReverseDependency, 

16 SearchResult, 

17 Version, 

18) 

19 

20 

21class RegistryAdapter(Protocol): 

22 """Protocol that all registry API adapters must implement. 

23 

24 The two capability methods at the bottom have real default bodies rather 

25 than `...`, so a concrete adapter that lacks a given real signal simply 

26 inherits an honest empty result instead of silently returning None. 

27 """ 

28 

29 @property 

30 def registry(self) -> Registry: ... 

31 

32 async def fetch_package(self, name: str) -> PackageInfo: ... 

33 

34 async def fetch_versions(self, name: str) -> list[Version]: ... 

35 

36 async def fetch_dependencies(self, name: str, version: str) -> list[Dependency]: ... 

37 

38 async def fetch_download_trend( 

39 self, name: str, days: int = 30 

40 ) -> list[DownloadTrendPoint]: ... 

41 

42 async def fetch_download_counts(self, name: str) -> DownloadCounts: ... 

43 

44 async def fetch_release_notes(self, name: str, version: str) -> str: ... 

45 

46 async def fetch_reverse_dependencies( 

47 self, name: str, limit: int = 5 

48 ) -> list[ReverseDependency]: 

49 """Packages depending on this one. Default: no reverse-dep API.""" 

50 return [] 

51 

52 async def fetch_reverse_dependency_count(self, name: str) -> int | None: 

53 """Total projects depending on this package. Default: no API.""" 

54 return None 

55 

56 async def fetch_version_download_breakdown(self, name: str) -> dict[int | str, int]: 

57 """Per-version download totals keyed by version id. Default: no API.""" 

58 return {} 

59 

60 async def search(self, query: str, limit: int = 10) -> list[SearchResult]: 

61 """Find packages in this registry; adapters may implement richer search.""" 

62 return [] 

63 

64 

65class AdapterBase: 

66 """Shared client binding for concrete registry adapters.""" 

67 

68 default_headers: ClassVar[dict[str, str]] = {} 

69 

70 def __init__(self, client: httpx.AsyncClient) -> None: 

71 self.client = client 

72 

73 def _client_scope(self): 

74 return _ClientLease(self.client, self.default_headers) 

75 

76 

77class _ClientLease: 

78 """Context-manager view that never closes the shared client.""" 

79 

80 def __init__( 

81 self, client: httpx.AsyncClient, headers: dict[str, str] | None = None 

82 ) -> None: 

83 self.client = client 

84 self.headers = headers or {} 

85 

86 async def __aenter__(self): 

87 return self 

88 

89 async def __aexit__(self, exc_type, exc, traceback) -> None: 

90 return None 

91 

92 @property 

93 def diagnostics(self): 

94 return getattr(self.client, "diagnostics", None) 

95 

96 async def get(self, url: str, *args, **kwargs): 

97 headers = httpx.Headers(self.client.headers) 

98 headers.update(self.headers) 

99 headers.update(kwargs.pop("headers", {}) or {}) 

100 return await self.client.get(url, *args, headers=headers, **kwargs) 

101 

102 

103def create_adapter(registry: Registry, *, client: httpx.AsyncClient) -> RegistryAdapter: 

104 """Factory: return the correct adapter for a given registry.""" 

105 from secchi.api.cran import CranAdapter 

106 from secchi.api.crates import CratesAdapter 

107 from secchi.api.golang import GoModuleAdapter 

108 from secchi.api.homebrew import HomebrewAdapter 

109 from secchi.api.npm import NpmAdapter 

110 from secchi.api.pypi import PyPIAdapter 

111 

112 adapters = { 

113 Registry.PYPI: PyPIAdapter, 

114 Registry.CRATES: CratesAdapter, 

115 Registry.NPM: NpmAdapter, 

116 Registry.HOMEBREW: HomebrewAdapter, 

117 Registry.GO: GoModuleAdapter, 

118 Registry.CRAN: CranAdapter, 

119 } 

120 cls = adapters[registry] 

121 return cls(client)