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 | ||||
| import datetime | |||||
| import contextlib | import contextlib | ||||
| from typing import Dict, Iterable, List, Union | from typing import Dict, Iterable, List, Union | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| SkippedContent, Content, Directory, Revision, Release, Snapshot, | BaseModel, SkippedContent, Content, Directory, Revision, Release, Snapshot, | ||||
| OriginVisit, Origin | OriginVisit, Origin | ||||
| ) | ) | ||||
| from . import get_storage | from . import get_storage | ||||
| from .exc import StorageArgumentException | from .exc import StorageArgumentException | ||||
| VALIDATION_EXCEPTIONS = ( | VALIDATION_EXCEPTIONS = ( | ||||
| ▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines | def release_add(self, releases: Iterable[Dict]) -> Dict: | ||||
| return self.storage.release_add(releases) | return self.storage.release_add(releases) | ||||
| def snapshot_add(self, snapshots: Iterable[Dict]) -> Dict: | def snapshot_add(self, snapshots: Iterable[Dict]) -> Dict: | ||||
| with convert_validation_exceptions(): | with convert_validation_exceptions(): | ||||
| snapshots = [Snapshot.from_dict(s) for s in snapshots] | snapshots = [Snapshot.from_dict(s) for s in snapshots] | ||||
| return self.storage.snapshot_add(snapshots) | return self.storage.snapshot_add(snapshots) | ||||
| def origin_visit_add( | def origin_visit_add( | ||||
| self, origin, date, type) -> Dict[str, Union[str, int]]: | self, origin_url: str, | ||||
| date: Union[datetime.datetime], type: str) -> Dict[str, BaseModel]: | |||||
| with convert_validation_exceptions(): | with convert_validation_exceptions(): | ||||
| visit = OriginVisit(origin=origin, date=date, type=type, | visit = OriginVisit(origin=origin_url, date=date, type=type, | ||||
| status='ongoing', snapshot=None) | status='ongoing', snapshot=None) | ||||
| return self.storage.origin_visit_add( | return self.storage.origin_visit_add( | ||||
| visit.origin, visit.date, visit.type) | visit.origin, visit.date, visit.type) | ||||
| def origin_add(self, origins: Iterable[Dict]) -> List[Dict]: | def origin_add(self, origins: Iterable[Dict]) -> List[Dict]: | ||||
| with convert_validation_exceptions(): | with convert_validation_exceptions(): | ||||
| origins = [Origin.from_dict(o) for o in origins] | origins = [Origin.from_dict(o) for o in origins] | ||||
| return self.storage.origin_add(origins) | return self.storage.origin_add(origins) | ||||
| def origin_add_one(self, origin: Dict) -> int: | def origin_add_one(self, origin: Dict) -> int: | ||||
| with convert_validation_exceptions(): | with convert_validation_exceptions(): | ||||
| origin = Origin.from_dict(origin) | origin = Origin.from_dict(origin) | ||||
| return self.storage.origin_add_one(origin) | return self.storage.origin_add_one(origin) | ||||