Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/npm/loader.py
| Show All 23 Lines | from swh.model.model import ( | ||||
| Sha1Git, | Sha1Git, | ||||
| ) | ) | ||||
| from swh.loader.package.loader import ( | from swh.loader.package.loader import ( | ||||
| BasePackageInfo, | BasePackageInfo, | ||||
| PackageLoader, | PackageLoader, | ||||
| RawExtrinsicMetadataCore, | RawExtrinsicMetadataCore, | ||||
| ) | ) | ||||
| from swh.loader.package.utils import api_info, release_name | from swh.loader.package.utils import api_info, cached_method, release_name | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| EMPTY_PERSON = Person(fullname=b"", name=None, email=None) | EMPTY_PERSON = Person(fullname=b"", name=None, email=None) | ||||
| ▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines | def __init__(self, url: str): | ||||
| """ | """ | ||||
| super().__init__(url=url) | super().__init__(url=url) | ||||
| package_name = url.split("https://www.npmjs.com/package/")[1] | package_name = url.split("https://www.npmjs.com/package/")[1] | ||||
| safe_name = quote(package_name, safe="") | safe_name = quote(package_name, safe="") | ||||
| self.provider_url = f"https://replicate.npmjs.com/{safe_name}/" | self.provider_url = f"https://replicate.npmjs.com/{safe_name}/" | ||||
| self._info: Dict[str, Any] = {} | self._info: Dict[str, Any] = {} | ||||
| self._versions = None | self._versions = None | ||||
| @property | @cached_method | ||||
| def info(self) -> Dict[str, Any]: | def _raw_info(self) -> bytes: | ||||
| return api_info(self.provider_url) | |||||
| @cached_method | |||||
| def info(self) -> Dict: | |||||
| """Return the project metadata information (fetched from npm registry) | """Return the project metadata information (fetched from npm registry) | ||||
| """ | """ | ||||
| if not self._info: | return json.loads(self._raw_info()) | ||||
| self._raw_info = api_info(self.provider_url) | |||||
| self._info = json.loads(self._raw_info) | |||||
| return self._info | |||||
| def get_versions(self) -> Sequence[str]: | def get_versions(self) -> Sequence[str]: | ||||
| return sorted(list(self.info["versions"].keys())) | return sorted(list(self.info()["versions"].keys())) | ||||
| def get_default_version(self) -> str: | def get_default_version(self) -> str: | ||||
| return self.info["dist-tags"].get("latest", "") | return self.info()["dist-tags"].get("latest", "") | ||||
| def get_metadata_authority(self): | def get_metadata_authority(self): | ||||
| return MetadataAuthority( | return MetadataAuthority( | ||||
| type=MetadataAuthorityType.FORGE, url="https://npmjs.com/", metadata={}, | type=MetadataAuthorityType.FORGE, url="https://npmjs.com/", metadata={}, | ||||
| ) | ) | ||||
| def get_extrinsic_snapshot_metadata(self): | def get_extrinsic_snapshot_metadata(self): | ||||
| return [ | return [ | ||||
| RawExtrinsicMetadataCore( | RawExtrinsicMetadataCore( | ||||
| format="replicate-npm-package-json", metadata=self._raw_info, | format="replicate-npm-package-json", metadata=self._raw_info(), | ||||
| ), | ), | ||||
| ] | ] | ||||
| def get_package_info(self, version: str) -> Iterator[Tuple[str, NpmPackageInfo]]: | def get_package_info(self, version: str) -> Iterator[Tuple[str, NpmPackageInfo]]: | ||||
| p_info = NpmPackageInfo.from_metadata( | p_info = NpmPackageInfo.from_metadata( | ||||
| project_metadata=self.info, version=version | project_metadata=self.info(), version=version | ||||
| ) | ) | ||||
| yield release_name(version), p_info | yield release_name(version), p_info | ||||
| def resolve_revision_from( | def resolve_revision_from( | ||||
| self, known_artifacts: Dict, p_info: NpmPackageInfo | self, known_artifacts: Dict, p_info: NpmPackageInfo | ||||
| ) -> Optional[bytes]: | ) -> Optional[bytes]: | ||||
| return artifact_to_revision_id(known_artifacts, p_info) | return artifact_to_revision_id(known_artifacts, p_info) | ||||
| ▲ Show 20 Lines • Show All 205 Lines • Show Last 20 Lines | |||||