Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/pypi/loader.py
# Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import os | import os | ||||
import logging | import logging | ||||
from typing import Any, Dict, Iterator, Optional, Sequence, Tuple | |||||
from typing import Any, Dict, Generator, Mapping, Optional, Sequence, Tuple | |||||
from urllib.parse import urlparse | from urllib.parse import urlparse | ||||
import attr | |||||
from pkginfo import UnpackedSDist | from pkginfo import UnpackedSDist | ||||
from swh.model.model import ( | from swh.model.model import ( | ||||
Person, | Person, | ||||
Sha1Git, | Sha1Git, | ||||
TimestampWithTimezone, | TimestampWithTimezone, | ||||
Revision, | Revision, | ||||
RevisionType, | RevisionType, | ||||
) | ) | ||||
from swh.loader.package.loader import PackageLoader | from swh.loader.package.loader import BasePackageInfo, PackageLoader | ||||
from swh.loader.package.utils import api_info, release_name, EMPTY_AUTHOR | from swh.loader.package.utils import api_info, release_name, EMPTY_AUTHOR | ||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
class PyPILoader(PackageLoader): | class PyPIPackageInfo(BasePackageInfo): | ||||
raw = attr.ib(type=Dict[str, Any]) | |||||
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) | ||||
Show All 10 Lines | def info(self) -> Dict: | ||||
return self._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_package_info( | def get_package_info(self, version: str) -> Iterator[Tuple[str, PyPIPackageInfo]]: | ||||
self, version: str | |||||
) -> Generator[Tuple[str, Mapping[str, Any]], None, None]: | |||||
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 | ||||
filename = meta["filename"] | filename = meta["filename"] | ||||
p_info = { | p_info = PyPIPackageInfo(url=meta["url"], filename=filename, raw=meta,) | ||||
"url": meta["url"], | |||||
"filename": filename, | |||||
"raw": 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 | ||||
else: | else: | ||||
for version, p_info in res: | for version, p_info in res: | ||||
yield release_name(version, p_info["filename"]), p_info | yield release_name(version, p_info.filename), p_info | ||||
def resolve_revision_from( | def resolve_revision_from( | ||||
self, known_artifacts: Dict, artifact_metadata: Dict | self, known_artifacts: Dict, artifact_metadata: Dict | ||||
) -> Optional[bytes]: | ) -> Optional[bytes]: | ||||
return artifact_to_revision_id(known_artifacts, artifact_metadata) | return artifact_to_revision_id(known_artifacts, artifact_metadata) | ||||
def build_revision( | def build_revision( | ||||
self, a_metadata: Dict, uncompressed_path: str, directory: Sha1Git | self, a_metadata: Dict, uncompressed_path: str, directory: Sha1Git | ||||
▲ Show 20 Lines • Show All 168 Lines • Show Last 20 Lines |