Differential D3648 Diff 12840 swh/storage/algos/snapshot.py

Changeset 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
		vlorentzUnsubmitted Not Done Inline Actions move this to a new `iter_origin_visit` generator vlorentz: move this to a new `iter_origin_visit` generator
		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,
		)
		vlorentzUnsubmitted Not Done Inline Actions move this to a new `iter_origin_visit_statuses` generator vlorentz: move this to a new `iter_origin_visit_statuses` generator
		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

		vlorentzUnsubmitted Not Done Inline Actions use snapshot_get_all_branches vlorentz: use snapshot_get_all_branches
		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