Coverage for src/secchi/api/npm.py: 73%
141 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-04 22:15 +0000
1"""npm registry adapter — direct REST API."""
3from __future__ import annotations
5from datetime import datetime, timedelta, timezone
7import httpx
9from secchi.api.base import AdapterBase, RegistryAdapter
10from secchi.models import (
11 Dependency,
12 DownloadCounts,
13 DownloadTrendPoint,
14 PackageInfo,
15 Registry,
16 ReleaseFile,
17 SearchResult,
18 Version,
19)
21NPM_REGISTRY = "https://registry.npmjs.org"
22NPM_DOWNLOADS = "https://api.npmjs.org/downloads"
25def _npm_name(name: str) -> str:
26 """URL-encode an npm package name (handles scoped packages like @scope/name)."""
27 if "/" in name:
28 parts = name.split("/", 1)
29 return f"{parts[0]}/{parts[1]}"
30 return name
33class NpmAdapter(AdapterBase, RegistryAdapter):
34 @property
35 def registry(self) -> Registry:
36 return Registry.NPM
38 async def fetch_package(self, name: str) -> PackageInfo:
39 async with self._client_scope() as client:
40 safe_name = _npm_name(name)
41 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
42 resp.raise_for_status()
43 data = resp.json()
45 latest_tag = data.get("dist-tags", {}).get("latest", "")
46 latest_info = (
47 data.get("versions", {}).get(latest_tag, {}) if latest_tag else {}
48 )
50 description = data.get("description", "")
51 author_info = data.get("author", {})
52 author = (
53 author_info.get("name", "")
54 if isinstance(author_info, dict)
55 else str(author_info)
56 if author_info
57 else ""
58 )
60 repo_info = data.get("repository", {})
61 repo_url = (
62 repo_info.get("url", "")
63 if isinstance(repo_info, dict)
64 else str(repo_info)
65 if repo_info
66 else ""
67 )
69 homepage = data.get("homepage", "")
71 if latest_info:
72 latest_date = next(
73 (
74 Version(
75 version=v,
76 release_date=_parse_npm_time(data.get("time", {}).get(v)),
77 )
78 for v in [latest_tag]
79 if v in data.get("time", {})
80 ),
81 Version(version=latest_tag),
82 )
83 latest_release_date = latest_date.release_date
84 else:
85 latest_release_date = None
87 total_downloads = await self._fetch_total_downloads(name)
89 kind = "CLI" if latest_info.get("bin") else "Library"
90 release_files: list[ReleaseFile] = []
91 unpacked = latest_info.get("dist", {}).get("unpackedSize")
92 if unpacked:
93 release_files.append(
94 ReleaseFile(
95 packagetype="npm-package",
96 size=unpacked,
97 filename=f"{name}-{latest_tag}.tgz",
98 )
99 )
101 return PackageInfo(
102 name=data["name"],
103 registry=Registry.NPM,
104 description=description,
105 author=author,
106 license=latest_info.get("license", ""),
107 homepage=homepage,
108 repository_url=repo_url,
109 latest_version=latest_tag,
110 latest_release_date=latest_release_date,
111 total_downloads=total_downloads,
112 package_kind=kind,
113 latest_release_files=release_files,
114 )
116 async def fetch_versions(self, name: str) -> list[Version]:
117 async with self._client_scope() as client:
118 safe_name = _npm_name(name)
119 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
120 resp.raise_for_status()
121 data = resp.json()
123 versions: list[Version] = []
124 time_data = data.get("time", {})
125 for ver, info in data.get("versions", {}).items():
126 release_date = _parse_npm_time(time_data.get(ver))
127 size = (
128 info.get("dist", {}).get("unpackedSize")
129 if isinstance(info, dict)
130 else None
131 )
132 versions.append(
133 Version(
134 version=ver,
135 release_date=release_date,
136 size_bytes=size,
137 )
138 )
140 versions.sort(key=lambda v: v.release_date or datetime.min, reverse=True)
141 return versions
143 async def fetch_dependencies(self, name: str, version: str) -> list[Dependency]:
144 async with self._client_scope() as client:
145 safe_name = _npm_name(name)
146 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
147 resp.raise_for_status()
148 data = resp.json()
150 version_data = data.get("versions", {}).get(version, {})
151 deps: list[Dependency] = []
153 for dep_key, dep_label in [
154 ("dependencies", False),
155 ("devDependencies", True),
156 ("peerDependencies", True),
157 ]:
158 dep_map = version_data.get(dep_key, {})
159 if isinstance(dep_map, dict):
160 for dep_name, req in dep_map.items():
161 deps.append(
162 Dependency(
163 name=dep_name,
164 requirement=str(req),
165 optional=dep_label != "dependencies",
166 )
167 )
169 return deps
171 async def fetch_download_trend(
172 self, name: str, days: int = 30
173 ) -> list[DownloadTrendPoint]:
174 async with self._client_scope() as client:
175 try:
176 if days <= 31:
177 url = f"{NPM_DOWNLOADS}/range/last-month/{name}"
178 else:
179 end = datetime.now(timezone.utc).date()
180 start = end - timedelta(days=days)
181 url = (
182 f"{NPM_DOWNLOADS}/range/{start:%Y-%m-%d}:{end:%Y-%m-%d}/{name}"
183 )
184 resp = await client.get(url)
185 resp.raise_for_status()
186 data = resp.json()
187 except httpx.HTTPError:
188 return []
190 points = [
191 DownloadTrendPoint(date=entry["day"], count=entry["downloads"])
192 for entry in data.get("downloads", [])
193 ]
194 return points[-days:] if len(points) > days else points
196 async def fetch_download_counts(self, name: str) -> DownloadCounts:
197 async with self._client_scope() as client:
198 today = week = month = 0
199 for period, store in [
200 ("last-day", "today"),
201 ("last-week", "week"),
202 ("last-month", "month"),
203 ]:
204 try:
205 resp = await client.get(f"{NPM_DOWNLOADS}/point/{period}/{name}")
206 resp.raise_for_status()
207 val = resp.json().get("downloads", 0)
208 if store == "today":
209 today = val
210 elif store == "week":
211 week = val
212 else:
213 month = val
214 except httpx.HTTPError:
215 pass
216 return DownloadCounts(today=today, week=week, month=month)
218 async def fetch_release_notes(self, name: str, version: str) -> str:
219 async with self._client_scope() as client:
220 safe_name = _npm_name(name)
221 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
222 resp.raise_for_status()
223 data = resp.json()
225 version_data = data.get("versions", {}).get(version, {})
226 readme = version_data.get("readme", "")
227 if readme:
228 max_chars = 2000
229 return readme[:max_chars] + ("..." if len(readme) > max_chars else "")
230 return ""
232 async def _fetch_total_downloads(self, name: str) -> int:
233 async with self._client_scope() as client:
234 try:
235 resp = await client.get(f"{NPM_DOWNLOADS}/point/last-year/{name}")
236 resp.raise_for_status()
237 return resp.json().get("downloads", 0)
238 except httpx.HTTPError:
239 return 0
241 async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
242 async with self._client_scope() as client:
243 try:
244 response = await client.get(
245 f"{NPM_REGISTRY}/-/v1/search",
246 params={"text": query, "size": limit},
247 )
248 response.raise_for_status()
249 except httpx.HTTPError:
250 return []
251 results: list[SearchResult] = []
252 for item in response.json().get("objects", [])[:limit]:
253 package = item.get("package", {})
254 name = package.get("name", "")
255 score_data = item.get("score", {})
256 results.append(
257 SearchResult(
258 name=name,
259 registry=Registry.NPM,
260 version=package.get("version", ""),
261 description=package.get("description", "") or "",
262 url=package.get("links", {}).get("npm", ""),
263 score=float(score_data.get("final", 0.0) or 0.0),
264 exact=name.lower() == query.lower(),
265 )
266 )
267 return results
270def _parse_npm_time(raw: str | None) -> datetime | None:
271 if not raw:
272 return None
273 try:
274 return datetime.fromisoformat(raw.replace("Z", "+00:00"))
275 except (ValueError, TypeError):
276 return None