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 async def get(self, url: str, *args, **kwargs): 

93 headers = dict(self.client.headers) 

94 headers.update(self.headers) 

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

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

97 

98 

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

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

101 from secchi.api.cran import CranAdapter 

102 from secchi.api.crates import CratesAdapter 

103 from secchi.api.golang import GoModuleAdapter 

104 from secchi.api.homebrew import HomebrewAdapter 

105 from secchi.api.npm import NpmAdapter 

106 from secchi.api.pypi import PyPIAdapter 

107 

108 adapters = { 

109 Registry.PYPI: PyPIAdapter, 

110 Registry.CRATES: CratesAdapter, 

111 Registry.NPM: NpmAdapter, 

112 Registry.HOMEBREW: HomebrewAdapter, 

113 Registry.GO: GoModuleAdapter, 

114 Registry.CRAN: CranAdapter, 

115 } 

116 cls = adapters[registry] 

117 return cls(client)