Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/deposit/loader.py
| Show All 22 Lines | from swh.model.model import ( | ||||
| MetadataAuthorityType, | MetadataAuthorityType, | ||||
| MetadataFetcher, | MetadataFetcher, | ||||
| ) | ) | ||||
| from swh.loader.package.loader import ( | from swh.loader.package.loader import ( | ||||
| BasePackageInfo, | BasePackageInfo, | ||||
| PackageLoader, | PackageLoader, | ||||
| RawExtrinsicMetadataCore, | RawExtrinsicMetadataCore, | ||||
| ) | ) | ||||
| from swh.loader.package.utils import download | from swh.loader.package.utils import cached_method, download | ||||
| logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
| @attr.s | @attr.s | ||||
| class DepositPackageInfo(BasePackageInfo): | class DepositPackageInfo(BasePackageInfo): | ||||
| filename = attr.ib(type=str) # instead of Optional[str] | filename = attr.ib(type=str) # instead of Optional[str] | ||||
| ▲ Show 20 Lines • Show All 66 Lines • ▼ Show 20 Lines | def __init__(self, url: str, deposit_id: str): | ||||
| deposit_id: Deposit identity | deposit_id: Deposit identity | ||||
| """ | """ | ||||
| super().__init__(url=url) | super().__init__(url=url) | ||||
| config_deposit = self.config["deposit"] | config_deposit = self.config["deposit"] | ||||
| self.deposit_id = deposit_id | self.deposit_id = deposit_id | ||||
| self.client = ApiClient(url=config_deposit["url"], auth=config_deposit["auth"]) | self.client = ApiClient(url=config_deposit["url"], auth=config_deposit["auth"]) | ||||
| self.metadata: Dict[str, Any] = {} | |||||
| def get_versions(self) -> Sequence[str]: | def get_versions(self) -> Sequence[str]: | ||||
| # only 1 branch 'HEAD' with no alias since we only have 1 snapshot | # only 1 branch 'HEAD' with no alias since we only have 1 snapshot | ||||
| # branch | # branch | ||||
| return ["HEAD"] | return ["HEAD"] | ||||
| def get_metadata_authority(self) -> MetadataAuthority: | def get_metadata_authority(self) -> MetadataAuthority: | ||||
| provider = self.metadata["origin_metadata"]["provider"] | provider = self.metadata()["origin_metadata"]["provider"] | ||||
| assert provider["provider_type"] == "deposit_client" | assert provider["provider_type"] == "deposit_client" | ||||
| return MetadataAuthority( | return MetadataAuthority( | ||||
| type=MetadataAuthorityType.DEPOSIT_CLIENT, | type=MetadataAuthorityType.DEPOSIT_CLIENT, | ||||
| url=provider["provider_url"], | url=provider["provider_url"], | ||||
| metadata={ | metadata={ | ||||
| "name": provider["provider_name"], | "name": provider["provider_name"], | ||||
| **(provider["metadata"] or {}), | **(provider["metadata"] or {}), | ||||
| }, | }, | ||||
| ) | ) | ||||
| def get_metadata_fetcher(self) -> MetadataFetcher: | def get_metadata_fetcher(self) -> MetadataFetcher: | ||||
| tool = self.metadata["origin_metadata"]["tool"] | tool = self.metadata()["origin_metadata"]["tool"] | ||||
| return MetadataFetcher( | return MetadataFetcher( | ||||
| name=tool["name"], version=tool["version"], metadata=tool["configuration"], | name=tool["name"], version=tool["version"], metadata=tool["configuration"], | ||||
| ) | ) | ||||
| def get_package_info( | def get_package_info( | ||||
| self, version: str | self, version: str | ||||
| ) -> Iterator[Tuple[str, DepositPackageInfo]]: | ) -> Iterator[Tuple[str, DepositPackageInfo]]: | ||||
| p_info = DepositPackageInfo.from_metadata( | p_info = DepositPackageInfo.from_metadata( | ||||
| self.metadata, url=self.url, filename="archive.zip", | self.metadata(), url=self.url, filename="archive.zip", | ||||
| ) | ) | ||||
| yield "HEAD", p_info | yield "HEAD", p_info | ||||
| def download_package( | def download_package( | ||||
| self, p_info: DepositPackageInfo, tmpdir: str | self, p_info: DepositPackageInfo, tmpdir: str | ||||
| ) -> List[Tuple[str, Mapping]]: | ) -> List[Tuple[str, Mapping]]: | ||||
| """Override to allow use of the dedicated deposit client | """Override to allow use of the dedicated deposit client | ||||
| Show All 22 Lines | ) -> Optional[Revision]: | ||||
| "provider": self.client.metadata_url(self.deposit_id), | "provider": self.client.metadata_url(self.deposit_id), | ||||
| "when": self.visit_date.isoformat(), | "when": self.visit_date.isoformat(), | ||||
| "raw": p_info.raw_info, | "raw": p_info.raw_info, | ||||
| }, | }, | ||||
| }, | }, | ||||
| ) | ) | ||||
| def get_extrinsic_origin_metadata(self) -> List[RawExtrinsicMetadataCore]: | def get_extrinsic_origin_metadata(self) -> List[RawExtrinsicMetadataCore]: | ||||
| origin_metadata = self.metadata["origin_metadata"] | origin_metadata = self.metadata()["origin_metadata"] | ||||
| return [ | return [ | ||||
| RawExtrinsicMetadataCore( | RawExtrinsicMetadataCore( | ||||
| format="sword-v2-atom-codemeta-v2-in-json", | format="sword-v2-atom-codemeta-v2-in-json", | ||||
| metadata=json.dumps(origin_metadata["metadata"]).encode(), | metadata=json.dumps(origin_metadata["metadata"]).encode(), | ||||
| ) | ) | ||||
| ] | ] | ||||
| @cached_method | |||||
| def metadata(self): | |||||
| """Returns metadata from the deposit server""" | |||||
| return self.client.metadata_get(self.deposit_id) | |||||
| def load(self) -> Dict: | def load(self) -> Dict: | ||||
| # First making sure the deposit is known prior to trigger a loading | # First making sure the deposit is known prior to trigger a loading | ||||
| try: | try: | ||||
| self.metadata = self.client.metadata_get(self.deposit_id) | self.metadata() | ||||
| except ValueError: | except ValueError: | ||||
| logger.error(f"Unknown deposit {self.deposit_id}, ignoring") | logger.error(f"Unknown deposit {self.deposit_id}, ignoring") | ||||
| return {"status": "failed"} | return {"status": "failed"} | ||||
| # Then usual loading | # Then usual loading | ||||
| r = super().load() | r = super().load() | ||||
| success = r["status"] != "failed" | success = r["status"] != "failed" | ||||
| # Update deposit status | # Update deposit status | ||||
| ▲ Show 20 Lines • Show All 123 Lines • Show Last 20 Lines | |||||