Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/pypi/loader.py
| Show All 21 Lines | from swh.model.model import ( | ||||
| RevisionType, | RevisionType, | ||||
| ) | ) | ||||
| 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, EMPTY_AUTHOR | from swh.loader.package.utils import ( | ||||
| api_info, | |||||
| cached_method, | |||||
| release_name, | |||||
| EMPTY_AUTHOR, | |||||
| ) | |||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| @attr.s | @attr.s | ||||
| class PyPIPackageInfo(BasePackageInfo): | class PyPIPackageInfo(BasePackageInfo): | ||||
| raw_info = attr.ib(type=Dict[str, Any]) | raw_info = attr.ib(type=Dict[str, Any]) | ||||
| Show All 17 Lines | class PyPILoader(PackageLoader[PyPIPackageInfo]): | ||||
| """Load pypi origin's artifact releases into swh archive. | """Load pypi origin's artifact releases into swh archive. | ||||
| """ | """ | ||||
| visit_type = "pypi" | visit_type = "pypi" | ||||
| def __init__(self, url): | def __init__(self, url): | ||||
| super().__init__(url=url) | super().__init__(url=url) | ||||
| self._info = None | |||||
| self.provider_url = pypi_api_url(self.url) | self.provider_url = pypi_api_url(self.url) | ||||
| @property | @cached_method | ||||
| def _raw_info(self) -> bytes: | |||||
| return api_info(self.provider_url) | |||||
| @cached_method | |||||
| def info(self) -> Dict: | def info(self) -> Dict: | ||||
| """Return the project metadata information (fetched from pypi registry) | """Return the project metadata information (fetched from pypi 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 self.info["releases"].keys() | return self.info()["releases"].keys() | ||||
| def get_default_version(self) -> str: | def get_default_version(self) -> str: | ||||
| return self.info["info"]["version"] | return self.info()["info"]["version"] | ||||
| def get_metadata_authority(self): | def get_metadata_authority(self): | ||||
| p_url = urlparse(self.url) | p_url = urlparse(self.url) | ||||
| return MetadataAuthority( | return MetadataAuthority( | ||||
| type=MetadataAuthorityType.FORGE, | type=MetadataAuthorityType.FORGE, | ||||
| url=f"{p_url.scheme}://{p_url.netloc}/", | url=f"{p_url.scheme}://{p_url.netloc}/", | ||||
| metadata={}, | metadata={}, | ||||
| ) | ) | ||||
| def get_extrinsic_snapshot_metadata(self): | def get_extrinsic_snapshot_metadata(self): | ||||
| return [ | return [ | ||||
| RawExtrinsicMetadataCore( | RawExtrinsicMetadataCore( | ||||
| format="pypi-project-json", metadata=self._raw_info, | format="pypi-project-json", metadata=self._raw_info(), | ||||
| ), | ), | ||||
| ] | ] | ||||
| def get_package_info(self, version: str) -> Iterator[Tuple[str, PyPIPackageInfo]]: | def get_package_info(self, version: str) -> Iterator[Tuple[str, PyPIPackageInfo]]: | ||||
| res = [] | res = [] | ||||
| for meta in self.info["releases"][version]: | for meta in self.info()["releases"][version]: | ||||
| if meta["packagetype"] != "sdist": | if meta["packagetype"] != "sdist": | ||||
| continue | continue | ||||
| p_info = PyPIPackageInfo.from_metadata(meta) | p_info = PyPIPackageInfo.from_metadata(meta) | ||||
| res.append((version, p_info)) | res.append((version, p_info)) | ||||
| if len(res) == 1: | if len(res) == 1: | ||||
| version, p_info = res[0] | version, p_info = res[0] | ||||
| yield release_name(version), p_info | yield release_name(version), p_info | ||||
| ▲ Show 20 Lines • Show All 179 Lines • Show Last 20 Lines | |||||