Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/opam/loader.py
| # Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 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 io | import io | ||||
| import os | import os | ||||
| from subprocess import PIPE, Popen, call | from subprocess import PIPE, Popen, call | ||||
| from typing import Iterator, List, Optional, Tuple | from typing import Iterator, List, Optional, Tuple | ||||
| import attr | import attr | ||||
| from swh.loader.package.loader import BasePackageInfo, PackageLoader | from swh.loader.package.loader import ( | ||||
| BasePackageInfo, | |||||
| PackageLoader, | |||||
| RawExtrinsicMetadataCore, | |||||
| ) | |||||
| from swh.loader.package.utils import cached_method | from swh.loader.package.utils import cached_method | ||||
| from swh.model.model import Person, Revision, RevisionType, Sha1Git | from swh.model.model import ( | ||||
| MetadataAuthority, | |||||
| MetadataAuthorityType, | |||||
| Person, | |||||
| Revision, | |||||
| RevisionType, | |||||
| Sha1Git, | |||||
| ) | |||||
| from swh.storage.interface import StorageInterface | from swh.storage.interface import StorageInterface | ||||
| @attr.s | @attr.s | ||||
| class OpamPackageInfo(BasePackageInfo): | class OpamPackageInfo(BasePackageInfo): | ||||
| author = attr.ib(type=Person) | author = attr.ib(type=Person) | ||||
| committer = attr.ib(type=Person) | committer = attr.ib(type=Person) | ||||
| version = attr.ib(type=str) | version = attr.ib(type=str) | ||||
| ▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines | def get_package_dir(self) -> str: | ||||
| ) | ) | ||||
| def get_package_name(self, version: str) -> str: | def get_package_name(self, version: str) -> str: | ||||
| return f"{self.opam_package}.{version}" | return f"{self.opam_package}.{version}" | ||||
| def get_package_file(self, version: str) -> str: | def get_package_file(self, version: str) -> str: | ||||
| return f"{self.get_package_dir()}/{self.get_package_name(version)}/opam" | return f"{self.get_package_dir()}/{self.get_package_name(version)}/opam" | ||||
| def get_metadata_authority(self): | |||||
| return MetadataAuthority(type=MetadataAuthorityType.FORGE, url=self.opam_url) | |||||
| @cached_method | @cached_method | ||||
| def _compute_versions(self) -> List[str]: | def _compute_versions(self) -> List[str]: | ||||
| """Compute the versions using opam internals | """Compute the versions using opam internals | ||||
| Raises: | Raises: | ||||
| ValueError in case the lister is not able to determine the list of versions | ValueError in case the lister is not able to determine the list of versions | ||||
| Returns: | Returns: | ||||
| ▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines | def get_versions(self) -> List[str]: | ||||
| raise ValueError("Invalid opam root") | raise ValueError("Invalid opam root") | ||||
| return self._compute_versions() | return self._compute_versions() | ||||
| def get_default_version(self) -> str: | def get_default_version(self) -> str: | ||||
| """Return the most recent version of the package as default.""" | """Return the most recent version of the package as default.""" | ||||
| return self._compute_versions()[-1] | return self._compute_versions()[-1] | ||||
| def get_enclosed_single_line_field(self, field, version) -> Optional[str]: | def _opam_show_args(self, version: str): | ||||
| package_file = self.get_package_file(version) | package_file = self.get_package_file(version) | ||||
| result = opam_read( | |||||
| [ | return [ | ||||
| "opam", | "opam", | ||||
| "show", | "show", | ||||
| "--color", | "--color", | ||||
| "never", | "never", | ||||
| "--safe", | "--safe", | ||||
| "--normalise", | "--normalise", | ||||
| "--root", | "--root", | ||||
| self.opam_root, | self.opam_root, | ||||
| "--file", | "--file", | ||||
| package_file, | package_file, | ||||
| "--field", | |||||
| field, | |||||
| ] | ] | ||||
| ) | |||||
| def get_enclosed_single_line_field(self, field, version) -> Optional[str]: | |||||
| result = opam_read(self._opam_show_args(version) + ["--field", field]) | |||||
| # Sanitize the result if any (remove trailing \n and enclosing ") | # Sanitize the result if any (remove trailing \n and enclosing ") | ||||
| return result.strip().strip('"') if result else None | return result.strip().strip('"') if result else None | ||||
| def get_package_info(self, version: str) -> Iterator[Tuple[str, OpamPackageInfo]]: | def get_package_info(self, version: str) -> Iterator[Tuple[str, OpamPackageInfo]]: | ||||
| url = self.get_enclosed_single_line_field("url.src:", version) | url = self.get_enclosed_single_line_field("url.src:", version) | ||||
| if url is None: | if url is None: | ||||
| raise ValueError( | raise ValueError( | ||||
| f"can't get field url.src: for version {version} of package {self.opam_package} \ | f"can't get field url.src: for version {version} of package {self.opam_package} \ | ||||
| (at url {self.url}) from `opam show`" | (at url {self.url}) from `opam show`" | ||||
| ) | ) | ||||
| authors_field = self.get_enclosed_single_line_field("authors:", version) | authors_field = self.get_enclosed_single_line_field("authors:", version) | ||||
| fullname = b"" if authors_field is None else str.encode(authors_field) | fullname = b"" if authors_field is None else str.encode(authors_field) | ||||
| author = Person(fullname=fullname, name=None, email=None) | author = Person(fullname=fullname, name=None, email=None) | ||||
| maintainer_field = self.get_enclosed_single_line_field("maintainer:", version) | maintainer_field = self.get_enclosed_single_line_field("maintainer:", version) | ||||
| fullname = b"" if maintainer_field is None else str.encode(maintainer_field) | fullname = b"" if maintainer_field is None else str.encode(maintainer_field) | ||||
| committer = Person(fullname=fullname, name=None, email=None) | committer = Person(fullname=fullname, name=None, email=None) | ||||
| with Popen(self._opam_show_args(version) + ["--raw"], stdout=PIPE) as proc: | |||||
| assert proc.stdout is not None | |||||
| metadata = proc.stdout.read() | |||||
| yield self.get_package_name(version), OpamPackageInfo( | yield self.get_package_name(version), OpamPackageInfo( | ||||
| url=url, filename=None, author=author, committer=committer, version=version | url=url, | ||||
| filename=None, | |||||
| author=author, | |||||
| committer=committer, | |||||
| version=version, | |||||
| directory_extrinsic_metadata=[ | |||||
| RawExtrinsicMetadataCore( | |||||
| metadata=metadata, format="opam-package-definition", | |||||
| ) | |||||
| ], | |||||
| ) | ) | ||||
| def build_revision( | def build_revision( | ||||
| self, p_info: OpamPackageInfo, uncompressed_path: str, directory: Sha1Git | self, p_info: OpamPackageInfo, uncompressed_path: str, directory: Sha1Git | ||||
| ) -> Optional[Revision]: | ) -> Optional[Revision]: | ||||
| return Revision( | return Revision( | ||||
| type=RevisionType.TAR, | type=RevisionType.TAR, | ||||
| Show All 9 Lines | |||||