Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/validate.py
| # Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 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 | ||||
| from typing import Dict, Iterable, List | from typing import Dict, Iterable, List | ||||
| from swh.model.hashutil import MultiHash, hash_to_bytes, hash_to_hex | from swh.model.hashutil import MultiHash, hash_to_bytes, hash_to_hex | ||||
| from swh.model.model import Content, Directory, Release, Revision, Snapshot | from swh.model.model import ( | ||||
| Content, | |||||
| Directory, | |||||
| MetadataAuthority, | |||||
| MetadataFetcher, | |||||
| OriginVisitStatus, | |||||
| RawExtrinsicMetadata, | |||||
| Release, | |||||
| Revision, | |||||
| Snapshot, | |||||
| ) | |||||
| from swh.storage import get_storage | from swh.storage import get_storage | ||||
| from swh.storage.exc import StorageArgumentException | from swh.storage.exc import StorageArgumentException | ||||
| from swh.storage.interface import StorageInterface | from swh.storage.interface import StorageInterface | ||||
| class ValidatingProxyStorage: | class ValidatingProxyStorage: | ||||
| """Proxy for storage classes, which checks inserted objects have a correct hash. | """Proxy for storage classes, which checks inserted objects have a correct hash. | ||||
| ▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines | class ValidatingProxyStorage: | ||||
| def release_add(self, releases: List[Release]) -> Dict[str, int]: | def release_add(self, releases: List[Release]) -> Dict[str, int]: | ||||
| self._check_hashes(releases) | self._check_hashes(releases) | ||||
| return self.storage.release_add(releases) | return self.storage.release_add(releases) | ||||
| def snapshot_add(self, snapshots: List[Snapshot]) -> Dict[str, int]: | def snapshot_add(self, snapshots: List[Snapshot]) -> Dict[str, int]: | ||||
| self._check_hashes(snapshots) | self._check_hashes(snapshots) | ||||
| return self.storage.snapshot_add(snapshots) | return self.storage.snapshot_add(snapshots) | ||||
| def origin_visit_status_add( | |||||
| self, visit_statuses: List[OriginVisitStatus], db=None, cur=None, | |||||
| ) -> Dict[str, int]: | |||||
| self._check_hashes(visit_statuses) | |||||
| return self.storage.origin_visit_status_add(visit_statuses) | |||||
| def raw_extrinsic_metadata_add( | |||||
| self, metadata: List[RawExtrinsicMetadata] | |||||
| ) -> Dict[str, int]: | |||||
| self._check_hashes(metadata) | |||||
| return self.storage.raw_extrinsic_metadata_add(metadata) | |||||
| def metadata_fetcher_add(self, fetchers: List[MetadataFetcher]) -> Dict[str, int]: | |||||
| self._check_hashes(fetchers) | |||||
| return self.storage.metadata_fetcher_add(fetchers) | |||||
| def metadata_authority_add( | |||||
| self, authorities: List[MetadataAuthority] | |||||
| ) -> Dict[str, int]: | |||||
| self._check_hashes(authorities) | |||||
| return self.storage.metadata_authority_add(authorities) | |||||