Coverage for src/secchi/api/pubdev.py: 92%

1"""pub.dev adapter for Dart and Flutter packages.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from datetime import UTC, datetime 

7 

8import httpx 

9 

10from secchi.api.sparse import SparseAdapter 

11from secchi.models import ( 

12 Dependency, 

13 DownloadCounts, 

14 PackageInfo, 

15 Registry, 

16 SearchResult, 

17 Version, 

18) 

19 

20PUB_API = "https://pub.dev/api" 

21 

22_MIN_DATETIME = datetime.min.replace(tzinfo=UTC) 

23 

24 

25class PubDevAdapter(SparseAdapter): 

26 @property 

27 def registry(self) -> Registry: 

28 return Registry.PUB 

29 

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

31 async with self._client_scope() as client: 

32 response = await client.get(f"{PUB_API}/packages/{name}") 

33 response.raise_for_status() 

34 data = response.json() 

35 package_name = data.get("name", name) 

36 latest = data.get("latest", {}) 

37 pubspec = latest.get("pubspec", {}) 

38 repository = pubspec.get("repository", "") or pubspec.get("homepage", "") 

39 return PackageInfo( 

40 name=package_name, 

41 registry=Registry.PUB, 

42 description=pubspec.get("description", "").strip(), 

43 author=pubspec.get("author", "") or "", 

44 homepage=pubspec.get("homepage", "") or repository, 

45 repository_url=repository, 

46 documentation_url=pubspec.get("documentation", "") 

47 or f"https://pub.dev/packages/{package_name}", 

48 latest_version=latest.get("version", "") or pubspec.get("version", ""), 

49 latest_release_date=_parse_time(latest.get("published")), 

50 versions=_versions(data.get("versions", [])), 

51 dependencies=_dependencies(pubspec), 

52 package_kind="Flutter Package" if _is_flutter(pubspec) else "Dart Package", 

53 download_counts=await self.fetch_download_counts(package_name), 

54 ) 

55 

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

57 async with self._client_scope() as client: 

58 response = await client.get(f"{PUB_API}/packages/{name}") 

59 response.raise_for_status() 

60 data = response.json() 

61 return _versions(data.get("versions", [])) 

62 

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

64 async with self._client_scope() as client: 

65 response = await client.get(f"{PUB_API}/packages/{name}") 

66 response.raise_for_status() 

67 data = response.json() 

68 for entry in data.get("versions", []): 

69 if entry.get("version") == version: 

70 return _dependencies(entry.get("pubspec", {})) 

71 return [] 

72 

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

74 # pub.dev only exposes a rolling 30-day count, no daily/weekly split 

75 # and no historical time series. 

76 async with self._client_scope() as client: 

77 try: 

78 response = await client.get(f"{PUB_API}/packages/{name}/score") 

79 response.raise_for_status() 

80 except httpx.HTTPError: 

81 return DownloadCounts() 

82 data = response.json() 

83 return DownloadCounts(month=data.get("downloadCount30Days") or 0) 

84 

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

86 async with self._client_scope() as client: 

87 names: list[str] = [] 

88 try: 

89 response = await client.get(f"{PUB_API}/search", params={"q": query}) 

90 response.raise_for_status() 

91 names = [ 

92 entry.get("package", "") 

93 for entry in response.json().get("packages", [])[:limit] 

94 if entry.get("package") 

95 ] 

96 except (httpx.HTTPError, ValueError, KeyError, TypeError): 

97 # The search endpoint is useful for suggestions, but it is not 

98 # authoritative for exact package resolution. Fall through to 

99 # the package endpoint below so newly indexed or oddly ranked 

100 # packages can still be opened directly. 

101 pass 

102 

103 # An unqualified CLI package name must resolve even when pub.dev's 

104 # search results omit the exact package (or the search endpoint is 

105 # temporarily unavailable). 

106 if query.casefold() not in {name.casefold() for name in names}: 

107 names.insert(0, query) 

108 

109 async def describe(candidate: str) -> SearchResult | None: 

110 try: 

111 pkg_response = await client.get(f"{PUB_API}/packages/{candidate}") 

112 pkg_response.raise_for_status() 

113 except httpx.HTTPError: 

114 return None 

115 data = pkg_response.json() 

116 resolved_name = data.get("name", candidate) 

117 latest = data.get("latest", {}) 

118 pubspec = latest.get("pubspec", {}) 

119 return SearchResult( 

120 name=resolved_name, 

121 registry=Registry.PUB, 

122 version=latest.get("version", ""), 

123 description=pubspec.get("description", "").strip(), 

124 url=f"https://pub.dev/packages/{resolved_name}", 

125 score=1.0, 

126 exact=resolved_name.casefold() == query.casefold(), 

127 ) 

128 

129 results = await asyncio.gather(*(describe(name) for name in names[:limit])) 

130 return [result for result in results if result is not None] 

131 

132 

133def _is_flutter(pubspec: dict) -> bool: 

134 if "flutter" in pubspec.get("dependencies", {}): 

135 return True 

136 return "flutter" in pubspec.get("environment", {}) 

137 

138 

139def _requirement(raw: object) -> str: 

140 if isinstance(raw, str): 

141 return raw 

142 if isinstance(raw, dict): 

143 return "git" if "git" in raw else "path" if "path" in raw else "" 

144 return "" 

145 

146 

147def _dependencies(pubspec: dict) -> list[Dependency]: 

148 deps: list[Dependency] = [] 

149 for dep_name, requirement in pubspec.get("dependencies", {}).items(): 

150 deps.append(Dependency(name=dep_name, requirement=_requirement(requirement))) 

151 for dep_name, requirement in pubspec.get("dev_dependencies", {}).items(): 

152 deps.append( 

153 Dependency( 

154 name=dep_name, requirement=_requirement(requirement), optional=True 

155 ) 

156 ) 

157 return deps 

158 

159 

160def _versions(raw: list[dict]) -> list[Version]: 

161 versions: list[Version] = [] 

162 for entry in raw: 

163 version = entry.get("version") or entry.get("pubspec", {}).get("version", "") 

164 if not version: 

165 continue 

166 versions.append( 

167 Version( 

168 version=version, 

169 release_date=_parse_time(entry.get("published")), 

170 is_yanked=bool(entry.get("retracted", False)), 

171 ) 

172 ) 

173 versions.sort(key=lambda v: v.release_date or _MIN_DATETIME, reverse=True) 

174 return versions 

175 

176 

177def _parse_time(raw: str | None) -> datetime | None: 

178 if not raw: 

179 return None 

180 try: 

181 return datetime.fromisoformat(raw.replace("Z", "+00:00")) 

182 except ValueError: 

183 return None