Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/nixguix/loader.py
Show All 21 Lines | |||||
from swh.loader.package.utils import EMPTY_AUTHOR | from swh.loader.package.utils import EMPTY_AUTHOR | ||||
from swh.loader.package.loader import BasePackageInfo, PackageLoader | from swh.loader.package.loader import BasePackageInfo, PackageLoader | ||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
@attr.s | |||||
class NixGuixPackageInfo(BasePackageInfo): | class NixGuixPackageInfo(BasePackageInfo): | ||||
raw = attr.ib(type=Dict[str, Any]) | raw = attr.ib(type=Dict[str, Any]) | ||||
integrity = attr.ib(type=str) | |||||
"""Hash of the archive, formatted as in the Subresource Integrity | |||||
specification.""" | |||||
@classmethod | |||||
def from_metadata(cls, metadata: Dict[str, Any]) -> "NixGuixPackageInfo": | |||||
return cls( | |||||
url=metadata["url"], | |||||
filename=None, | |||||
integrity=metadata["integrity"], | |||||
raw=metadata, | |||||
) | |||||
class NixGuixLoader(PackageLoader[NixGuixPackageInfo]): | class NixGuixLoader(PackageLoader[NixGuixPackageInfo]): | ||||
"""Load sources from a sources.json file. This loader is used to load | """Load sources from a sources.json file. This loader is used to load | ||||
sources used by functional package manager (eg. Nix and Guix). | sources used by functional package manager (eg. Nix and Guix). | ||||
""" | """ | ||||
visit_type = "nixguix" | visit_type = "nixguix" | ||||
Show All 24 Lines | class NixGuixLoader(PackageLoader[NixGuixPackageInfo]): | ||||
# Note: this could be renamed get_artifact_info in the PackageLoader | # Note: this could be renamed get_artifact_info in the PackageLoader | ||||
# base class. | # base class. | ||||
def get_package_info(self, url) -> Iterator[Tuple[str, NixGuixPackageInfo]]: | def get_package_info(self, url) -> Iterator[Tuple[str, NixGuixPackageInfo]]: | ||||
# TODO: try all mirrors and not only the first one. A source | # TODO: try all mirrors and not only the first one. A source | ||||
# can be fetched from several urls, called mirrors. We | # can be fetched from several urls, called mirrors. We | ||||
# currently only use the first one, but if the first one | # currently only use the first one, but if the first one | ||||
# fails, we should try the second one and so on. | # fails, we should try the second one and so on. | ||||
integrity = self._integrityByUrl[url] | integrity = self._integrityByUrl[url] | ||||
p_info = NixGuixPackageInfo( | p_info = NixGuixPackageInfo.from_metadata({"url": url, "integrity": integrity}) | ||||
url=url, filename=None, raw={"url": url, "integrity": integrity}, | |||||
) | |||||
yield url, p_info | yield url, p_info | ||||
def known_artifacts(self, snapshot: Optional[Snapshot]) -> Dict[Sha1Git, BaseModel]: | def known_artifacts(self, snapshot: Optional[Snapshot]) -> Dict[Sha1Git, BaseModel]: | ||||
"""Almost same implementation as the default one except it filters out the extra | """Almost same implementation as the default one except it filters out the extra | ||||
"evaluation" branch which does not have the right metadata structure. | "evaluation" branch which does not have the right metadata structure. | ||||
""" | """ | ||||
if not snapshot: | if not snapshot: | ||||
Show All 14 Lines | def known_artifacts(self, snapshot: Optional[Snapshot]) -> Dict[Sha1Git, BaseModel]: | ||||
ret = {} | ret = {} | ||||
for revision in known_revisions: | for revision in known_revisions: | ||||
if not revision: # revision_get can return None | if not revision: # revision_get can return None | ||||
continue | continue | ||||
ret[revision["id"]] = revision["metadata"] | ret[revision["id"]] = revision["metadata"] | ||||
return ret | return ret | ||||
def resolve_revision_from( | def resolve_revision_from( | ||||
self, known_artifacts: Dict, artifact_metadata: Dict | self, known_artifacts: Dict, p_info: NixGuixPackageInfo, | ||||
) -> Optional[bytes]: | ) -> Optional[bytes]: | ||||
for rev_id, known_artifact in known_artifacts.items(): | for rev_id, known_artifact in known_artifacts.items(): | ||||
try: | try: | ||||
known_integrity = known_artifact["extrinsic"]["raw"]["integrity"] | known_integrity = known_artifact["extrinsic"]["raw"]["integrity"] | ||||
except KeyError as e: | except KeyError as e: | ||||
logger.exception( | logger.exception( | ||||
"Unexpected metadata revision structure detected: %(context)s", | "Unexpected metadata revision structure detected: %(context)s", | ||||
{ | { | ||||
"context": { | "context": { | ||||
"revision": hashutil.hash_to_hex(rev_id), | "revision": hashutil.hash_to_hex(rev_id), | ||||
"reason": str(e), | "reason": str(e), | ||||
"known_artifact": known_artifact, | "known_artifact": known_artifact, | ||||
} | } | ||||
}, | }, | ||||
) | ) | ||||
# metadata field for the revision is not as expected by the loader | # metadata field for the revision is not as expected by the loader | ||||
# nixguix. We consider this not the right revision and continue checking | # nixguix. We consider this not the right revision and continue checking | ||||
# the other revisions | # the other revisions | ||||
continue | continue | ||||
else: | else: | ||||
if artifact_metadata["integrity"] == known_integrity: | if p_info.integrity == known_integrity: | ||||
return rev_id | return rev_id | ||||
return None | return None | ||||
def extra_branches(self) -> Dict[bytes, Mapping[str, Any]]: | def extra_branches(self) -> Dict[bytes, Mapping[str, Any]]: | ||||
"""We add a branch to the snapshot called 'evaluation' pointing to the | """We add a branch to the snapshot called 'evaluation' pointing to the | ||||
revision used to generate the sources.json file. This revision | revision used to generate the sources.json file. This revision | ||||
is specified in the sources.json file itself. For the nixpkgs | is specified in the sources.json file itself. For the nixpkgs | ||||
origin, this revision is coming from the | origin, this revision is coming from the | ||||
Show All 11 Lines | def extra_branches(self) -> Dict[bytes, Mapping[str, Any]]: | ||||
return { | return { | ||||
b"evaluation": { | b"evaluation": { | ||||
"target_type": "revision", | "target_type": "revision", | ||||
"target": hashutil.hash_to_bytes(self.revision), | "target": hashutil.hash_to_bytes(self.revision), | ||||
} | } | ||||
} | } | ||||
def build_revision( | def build_revision( | ||||
self, a_metadata: Dict, uncompressed_path: str, directory: Sha1Git | self, p_info: NixGuixPackageInfo, uncompressed_path: str, directory: Sha1Git | ||||
) -> Optional[Revision]: | ) -> Optional[Revision]: | ||||
return Revision( | return Revision( | ||||
type=RevisionType.TAR, | type=RevisionType.TAR, | ||||
message=b"", | message=b"", | ||||
author=EMPTY_AUTHOR, | author=EMPTY_AUTHOR, | ||||
date=None, | date=None, | ||||
committer=EMPTY_AUTHOR, | committer=EMPTY_AUTHOR, | ||||
committer_date=None, | committer_date=None, | ||||
parents=(), | parents=(), | ||||
directory=directory, | directory=directory, | ||||
synthetic=True, | synthetic=True, | ||||
metadata={ | metadata={ | ||||
"extrinsic": { | "extrinsic": { | ||||
"provider": self.provider_url, | "provider": self.provider_url, | ||||
"when": self.visit_date.isoformat(), | "when": self.visit_date.isoformat(), | ||||
"raw": a_metadata, | "raw": p_info.raw, | ||||
}, | }, | ||||
}, | }, | ||||
) | ) | ||||
def retrieve_sources(url: str) -> Dict[str, Any]: | def retrieve_sources(url: str) -> Dict[str, Any]: | ||||
response = requests.get(url, allow_redirects=True) | response = requests.get(url, allow_redirects=True) | ||||
if response.status_code != 200: | if response.status_code != 200: | ||||
▲ Show 20 Lines • Show All 62 Lines • Show Last 20 Lines |