Differential D3648 Diff 12857 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,
		iter_origin_visits,
		iter_origin_visit_statuses,
		)
		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_id_get_from_revision(
		storage: StorageInterface, origin: str, revision_id: bytes
		) -> Optional[bytes]:
		"""Retrieve the most recent snapshot id targeting the revision_id for the given origin.

		Warning This is a potentially highly costly operation

		Returns
		The snapshot id if found. None otherwise.

		"""
		revision = storage.revision_get([revision_id])
		if not revision:
		return None

		for visit in iter_origin_visits(storage, origin, order=ListOrder.DESC):
		assert visit.visit is not None
		for visit_status in iter_origin_visit_statuses(
		storage, origin, visit.visit, order=ListOrder.DESC
		vlorentzUnsubmitted Not Done Inline Actions move this to a new `iter_origin_visit` generator vlorentz: move this to a new `iter_origin_visit` generator
		):
		snapshot_id = visit_status.snapshot
		if snapshot_id is None:
		continue

		snapshot = snapshot_get_all_branches(storage, snapshot_id)
		if not snapshot:
		continue
		for branch_name, branch in snapshot["branches"].items():
		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
		if (
		branch is not None
		and branch["target_type"] == TargetType.REVISION.value
		and branch["target"] == revision_id
		): # snapshot found
		return snapshot_id
		vlorentzUnsubmitted Not Done Inline Actions use snapshot_get_all_branches vlorentz: use snapshot_get_all_branches

		return None