Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/algos/snapshot.py
| # Copyright (C) 2018-2020 The Software Heritage developers | # Copyright (C) 2018-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 List, Optional | from typing import List, Optional | ||||
| from swh.model.model import Snapshot | from swh.model.model import Snapshot, TargetType | ||||
| from swh.storage.algos.origin import origin_get_latest_visit_status | from swh.storage.algos.origin import origin_get_latest_visit_status | ||||
| from swh.storage.interface import ListOrder, StorageInterface | |||||
| def snapshot_get_all_branches(storage, snapshot_id): | def snapshot_get_all_branches(storage, snapshot_id): | ||||
| """Get all the branches for a given snapshot | """Get all the branches for a given snapshot | ||||
| Args: | Args: | ||||
| storage (swh.storage.storage.Storage): the storage instance | storage (swh.storage.storage.Storage): the storage instance | ||||
| snapshot_id (bytes): the snapshot's identifier | snapshot_id (bytes): the snapshot's identifier | ||||
| ▲ Show 20 Lines • Show All 69 Lines • ▼ Show 20 Lines | if branches_count: # partial snapshot | ||||
| snapshot_id, branches_count=branches_count | snapshot_id, branches_count=branches_count | ||||
| ) | ) | ||||
| if snapshot is None: | if snapshot is None: | ||||
| return None | return None | ||||
| snapshot.pop("next_branch") | snapshot.pop("next_branch") | ||||
| else: | else: | ||||
| snapshot = snapshot_get_all_branches(storage, snapshot_id) | snapshot = snapshot_get_all_branches(storage, snapshot_id) | ||||
| return Snapshot.from_dict(snapshot) if snapshot else None | return Snapshot.from_dict(snapshot) if snapshot else None | ||||
| def snapshot_get_from_revision( | |||||
| storage: StorageInterface, origin: str, revision_id: bytes | |||||
| ) -> Optional[bytes]: | |||||
| """Retrieve the most recent snapshot targeting the revision_id for the given origin. | |||||
| Returns | |||||
| The snapshot id if found. None otherwise. | |||||
| """ | |||||
| revision = storage.revision_get([revision_id]) | |||||
| if not revision: | |||||
| return None | |||||
| next_page_token_visit = None | |||||
| while True: | |||||
| visit_page = storage.origin_visit_get( | |||||
| origin, order=ListOrder.DESC, page_token=next_page_token_visit | |||||
| ) | |||||
| next_page_token_visit = visit_page.next_page_token | |||||
| for visit in visit_page.results: | |||||
| next_page_token_visit_status = None | |||||
| while True: | |||||
| visit_status_page = storage.origin_visit_status_get( | |||||
| origin, | |||||
| visit.visit, | |||||
| order=ListOrder.DESC, | |||||
| page_token=next_page_token_visit_status, | |||||
| ) | |||||
| next_page_token_visit_status = visit_status_page.next_page_token | |||||
| for visit_status in visit_status_page.results: | |||||
| snapshot_id = visit_status.snapshot | |||||
| if snapshot_id is None: | |||||
| continue | |||||
| detail_snapshot = storage.snapshot_get(snapshot_id) | |||||
| if not detail_snapshot: | |||||
| continue | |||||
| for branch_name, branch in detail_snapshot["branches"].items(): | |||||
| if ( | |||||
| branch is not None | |||||
| and branch["target_type"] == TargetType.REVISION.value | |||||
| and branch["target"] == revision_id | |||||
| ): # snapshot found | |||||
| return snapshot_id | |||||
| if next_page_token_visit_status is None: | |||||
| break | |||||
| if next_page_token_visit is None: | |||||
| break | |||||
| return None | |||||