npm: use /latest, the abbreviated doc strips repository and metadata

src/devnet/sources/npm.py

@@ -5,17 +5,18 @@ from urllib.parse import quote
 
 from ..core import Resource, UpstreamError, get
 
-API = "https://registry.npmjs.org/{name}"
+# The /latest document carries full metadata for one version. The full package
+# document adds version history and publish dates, but trips the response size
+# cap for popular packages - @types/node has thousands of versions.
+#
+# ponytail: no version list and no publish dates. Ceiling: npm alone cannot answer
+# "is this package abandoned?" - use the repo's last push or its RSS feed instead.
+# Upgrade: a second call to the abbreviated document when a caller needs history.
+API = "https://registry.npmjs.org/{name}/latest"
 
-# The abbreviated "corgi" document: every version, but only the fields an
-# installer needs. The full document is megabytes for popular packages and
-# would trip the response size cap.
-_ABBREVIATED = {"Accept": "application/vnd.npm.install-v1+json"}
-
-# ponytail: repo_slug is parsed from the `repository` field, which npm lets
-# authors write in at least six shapes. Ceiling: a monorepo `directory` field is
-# ignored, so the slug points at the repo root rather than the sub-package.
-# Upgrade: cross-check package.json `exports` when a wrong root shows up.
+# ponytail: repo_slug is parsed from `repository`, which npm lets authors write in
+# at least six shapes. Ceiling: a monorepo `directory` field is ignored, so the
+# slug points at the repo root rather than the sub-package.
 
 
 def _text(value) -> str | None:
@@ -23,9 +24,9 @@ def _text(value) -> str | None:
     if isinstance(value, str):
         return value.strip() or None
     if isinstance(value, dict):
-        for key in ("type", "name", "url"):
-            if isinstance(value.get(key), str) and value[key].strip():
-                return value[key].strip()
+        for field in ("type", "name", "url"):
+            if isinstance(value.get(field), str) and value[field].strip():
+                return value[field].strip()
     return None
 
 
@@ -47,22 +48,20 @@ def _repo_slug(repository) -> str | None:
 
 
 def package(name: str) -> dict:
-    """Current dist-tag, publish date, dependencies and release history."""
+    """Current published version, dependencies, license and source repository."""
     raw_name = str(name or "").strip().lower()
     if not raw_name or len(raw_name) > 214:
         raise UpstreamError("invalid npm package name")
 
-    payload = get(
-        API.format(name=quote(raw_name, safe="")),
-        headers=_ABBREVIATED,
-        ttl=900,
-    )
-    tags = payload.get("dist-tags") or {}
-    latest = tags.get("latest") or ""
-    versions = payload.get("versions") or {}
-    times = payload.get("time") or {}
-    manifest = versions.get(latest) or {}
+    payload = get(API.format(name=quote(raw_name, safe="")), ttl=900)
+    if not isinstance(payload, dict):
+        raise UpstreamError("unexpected npm response")
+
+    version = payload.get("version") or ""
+    dependencies = payload.get("dependencies") or {}
+    peer = payload.get("peerDependencies") or {}
     maintainers = payload.get("maintainers") or []
+    repository = payload.get("repository")
     canonical = payload.get("name") or raw_name
 
     return Resource(
@@ -73,34 +72,29 @@ def package(name: str) -> dict:
         url=f"https://www.npmjs.com/package/{canonical}",
         description=payload.get("description"),
         author={"name": _text(payload.get("author"))},
-        timestamps={
-            "published_at": times.get(latest) or None,
-            "updated_at": times.get("modified") or None,
+        timestamps={},
+        metrics={
+            "dependencies": len(dependencies),
+            "peer_dependencies": len(peer),
+            "maintainers": len(maintainers),
         },
-        metrics={"versions": len(versions), "maintainers": len(maintainers)},
         tags=[k for k in (payload.get("keywords") or []) if isinstance(k, str)][:12],
         metadata={
             "license": _text(payload.get("license")),
-            "repo_slug": _repo_slug(payload.get("repository")),
-            "repository": _text(payload.get("repository")) or (
-                payload.get("repository").get("url")
-                if isinstance(payload.get("repository"), dict)
-                else None
-            ),
-            "dist_tags": tags,
+            "repo_slug": _repo_slug(repository),
+            "repository": repository if isinstance(repository, str) else (repository or {}).get("url"),
             "homepage": _text(payload.get("homepage")),
+            "deprecated": _text(payload.get("deprecated")),
         },
         data={
-            "latest_version": latest,
-            "published_at": times.get(latest) or None,
-            "created_at": times.get("created") or None,
-            "versions": sorted(versions)[-30:],
-            "dependencies": manifest.get("dependencies") or {},
-            "peer_dependencies": manifest.get("peerDependencies") or {},
-            "engines": manifest.get("engines") or {},
+            "latest_version": version,
+            "dependencies": dependencies,
+            "peer_dependencies": peer,
+            "engines": payload.get("engines") or {},
             "maintainers": [
                 m.get("name") for m in maintainers if isinstance(m, dict) and m.get("name")
             ][:10],
+            "tarball": (payload.get("dist") or {}).get("tarball"),
         },
         provenance={"provider": "npm_registry", "url": API.format(name=raw_name)},
     ).dict()