Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/debian/loader.py
| # Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-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 email.utils | import email.utils | ||||
| import logging | import logging | ||||
| from os import path | from os import path | ||||
| import re | import re | ||||
| import subprocess | import subprocess | ||||
| from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple | |||||
| import attr | |||||
| from dateutil.parser import parse as parse_date | from dateutil.parser import parse as parse_date | ||||
| from debian.changelog import Changelog | from debian.changelog import Changelog | ||||
| from debian.deb822 import Dsc | from debian.deb822 import Dsc | ||||
| from typing import Any, Generator, List, Mapping, Optional, Sequence, Tuple | |||||
| from swh.loader.package.loader import PackageLoader | from swh.loader.package.loader import BasePackageInfo, PackageLoader | ||||
| from swh.loader.package.utils import download, release_name | from swh.loader.package.utils import download, release_name | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| Sha1Git, | Sha1Git, | ||||
| Person, | Person, | ||||
| Revision, | Revision, | ||||
| RevisionType, | RevisionType, | ||||
| TimestampWithTimezone, | TimestampWithTimezone, | ||||
| ) | ) | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| UPLOADERS_SPLIT = re.compile(r"(?<=\>)\s*,\s*") | UPLOADERS_SPLIT = re.compile(r"(?<=\>)\s*,\s*") | ||||
| class DebianLoader(PackageLoader): | class DebianPackageInfo(BasePackageInfo): | ||||
| raw = attr.ib(type=Dict[str, Any]) | |||||
| class DebianLoader(PackageLoader[DebianPackageInfo]): | |||||
| """Load debian origins into swh archive. | """Load debian origins into swh archive. | ||||
| """ | """ | ||||
| visit_type = "deb" | visit_type = "deb" | ||||
| def __init__(self, url: str, date: str, packages: Mapping[str, Any]): | def __init__(self, url: str, date: str, packages: Mapping[str, Any]): | ||||
| """Debian Loader implementation. | """Debian Loader implementation. | ||||
| Show All 40 Lines | class DebianLoader(PackageLoader[DebianPackageInfo]): | ||||
| def get_versions(self) -> Sequence[str]: | def get_versions(self) -> Sequence[str]: | ||||
| """Returns the keys of the packages input (e.g. | """Returns the keys of the packages input (e.g. | ||||
| stretch/contrib/0.7.2-3, etc...) | stretch/contrib/0.7.2-3, etc...) | ||||
| """ | """ | ||||
| return list(self.packages.keys()) | return list(self.packages.keys()) | ||||
| def get_package_info( | def get_package_info(self, version: str) -> Iterator[Tuple[str, DebianPackageInfo]]: | ||||
ardumont: same, Iterator... | |||||
| self, version: str | |||||
| ) -> Generator[Tuple[str, Mapping[str, Any]], None, None]: | |||||
| meta = self.packages[version] | meta = self.packages[version] | ||||
| p_info = meta.copy() | p_info = DebianPackageInfo(url=self.url, filename=None, raw=meta,) | ||||
| p_info["raw"] = meta | |||||
| yield release_name(version), p_info | yield release_name(version), p_info | ||||
| def resolve_revision_from( | def resolve_revision_from( | ||||
| self, known_package_artifacts: Mapping, artifact_metadata: Mapping | self, known_package_artifacts: Mapping, artifact_metadata: Mapping | ||||
| ) -> Optional[bytes]: | ) -> Optional[bytes]: | ||||
| return resolve_revision_from(known_package_artifacts, artifact_metadata) | return resolve_revision_from(known_package_artifacts, artifact_metadata) | ||||
| def download_package( | def download_package( | ||||
| self, p_info: Mapping[str, Any], tmpdir: str | self, p_info: DebianPackageInfo, tmpdir: str | ||||
| ) -> List[Tuple[str, Mapping]]: | ) -> List[Tuple[str, Mapping]]: | ||||
| """Contrary to other package loaders (1 package, 1 artifact), | """Contrary to other package loaders (1 package, 1 artifact), | ||||
| `a_metadata` represents the package's datafiles set to fetch: | `p_info.files` represents the package's datafiles set to fetch: | ||||
Done Inline Actionsp_info.raw now ardumont: p_info.raw now | |||||
| - <package-version>.orig.tar.gz | - <package-version>.orig.tar.gz | ||||
| - <package-version>.dsc | - <package-version>.dsc | ||||
| - <package-version>.diff.gz | - <package-version>.diff.gz | ||||
| This is delegated to the `download_package` function. | This is delegated to the `download_package` function. | ||||
| """ | """ | ||||
| all_hashes = download_package(p_info, tmpdir) | all_hashes = download_package(p_info.raw, tmpdir) | ||||
| logger.debug("all_hashes: %s", all_hashes) | logger.debug("all_hashes: %s", all_hashes) | ||||
| res = [] | res = [] | ||||
| for hashes in all_hashes.values(): | for hashes in all_hashes.values(): | ||||
| res.append((tmpdir, hashes)) | res.append((tmpdir, hashes)) | ||||
| logger.debug("res: %s", res) | logger.debug("res: %s", res) | ||||
| return res | return res | ||||
| def uncompress( | def uncompress( | ||||
| ▲ Show 20 Lines • Show All 296 Lines • Show Last 20 Lines | |||||
same, Iterator...