Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/client.py
| Show First 20 Lines • Show All 539 Lines • ▼ Show 20 Lines | def compute_headers(self, info: Dict[str, Any]) -> Dict[str, Any]: | ||||
| "CONTENT-TYPE": "application/atom+xml;type=entry", | "CONTENT-TYPE": "application/atom+xml;type=entry", | ||||
| "X_CHECK_SWHID": info["swhid"], | "X_CHECK_SWHID": info["swhid"], | ||||
| } | } | ||||
| def compute_method(self, *args, **kwargs) -> str: | def compute_method(self, *args, **kwargs) -> str: | ||||
| return "put" | return "put" | ||||
| class CreateMetadataOnlyDepositClient(BaseCreateDepositClient): | |||||
| """Create metadata-only deposit.""" | |||||
| def compute_information(self, *args, **kwargs) -> Dict[str, Any]: | |||||
| return { | |||||
| "headers": {"CONTENT-TYPE": "application/atom+xml;type=entry",}, | |||||
| "filepath": kwargs["metadata_path"], | |||||
| } | |||||
| def parse_result_ok(self, xml_content): | |||||
| """Given an xml content as string, returns a deposit dict. | |||||
| """ | |||||
| data = parse_xml(xml_content) | |||||
| keys = [ | |||||
| "deposit_id", | |||||
| "deposit_status", | |||||
| "deposit_date", | |||||
| ] | |||||
| return {key: data.get("swh:" + key) for key in keys} | |||||
| class CreateMultipartDepositClient(BaseCreateDepositClient): | class CreateMultipartDepositClient(BaseCreateDepositClient): | ||||
| """Create a multipart deposit client.""" | """Create a multipart deposit client.""" | ||||
| def _multipart_info(self, info, info_meta): | def _multipart_info(self, info, info_meta): | ||||
| files = [ | files = [ | ||||
| ( | ( | ||||
| "file", | "file", | ||||
| (info["filename"], open(info["filepath"], "rb"), info["content-type"]), | (info["filename"], open(info["filepath"], "rb"), info["content-type"]), | ||||
| ▲ Show 20 Lines • Show All 151 Lines • ▼ Show 20 Lines | ): | ||||
| archive_path=archive, | archive_path=archive, | ||||
| metadata_path=metadata, | metadata_path=metadata, | ||||
| replace=replace, | replace=replace, | ||||
| ) | ) | ||||
| if "error" in r: | if "error" in r: | ||||
| return r | return r | ||||
| return self.deposit_status(collection, deposit_id) | return self.deposit_status(collection, deposit_id) | ||||
| def deposit_metadata_only( | |||||
| self, collection: str, metadata: Optional[str] = None, | |||||
| ): | |||||
| assert metadata is not None | |||||
| return CreateMetadataOnlyDepositClient( | |||||
| url=self.base_url, auth=self.auth | |||||
| ).execute(collection, metadata_path=metadata) | |||||