Differential D3314 Diff 11732 swh/storage/algos/snapshot.py

Changeset View

Standalone View

swh/storage/algos/snapshot.py

# Copyright (C) 2018 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 Iterable, Optional

		from swh.model.model import Snapshot

		from swh.storage.algos.origin import origin_get_latest_visit_status


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
Returns:		Returns:
Show All 9 Lines	def snapshot_get_all_branches(storage, snapshot_id):

next_branch = ret.pop("next_branch", None)		next_branch = ret.pop("next_branch", None)
while next_branch:		while next_branch:
data = storage.snapshot_get_branches(snapshot_id, branches_from=next_branch)		data = storage.snapshot_get_branches(snapshot_id, branches_from=next_branch)
ret["branches"].update(data["branches"])		ret["branches"].update(data["branches"])
next_branch = data.get("next_branch")		next_branch = data.get("next_branch")

return ret		return ret


		def snapshot_get_latest(
		storage, origin: str, allowed_statuses: Optional[Iterable[str]] = None
		) -> Optional[Snapshot]:
		"""Get the latest snapshot for the given origin, optionally only from visits that have
		one of the given allowed_statuses.

		The branches of the snapshot are iterated in the lexicographical
		order of their names.

		Args:
		storage: Storage instance
		origin: the origin's URL
		allowed_statuses: list of visit statuses considered
		to find the latest snapshot for the visit. For instance,
		``allowed_statuses=['full']`` will only consider visits that
		have successfully run to completion.
		Returns:
		The snapshot object if one is found matching the criteria or None.

		"""
		visit_and_status = origin_get_latest_visit_status(
		storage, origin, allowed_statuses=allowed_statuses, require_snapshot=True
		)

		if not visit_and_status:
		return None

		_, visit_status = visit_and_status
		if not visit_status.snapshot:
		return None

		snapshot = snapshot_get_all_branches(storage, visit_status.snapshot)
		return Snapshot.from_dict(snapshot)
		olasdUnsubmitted Not Done Inline Actions Raise a ValueError instead? olasd: Raise a ValueError instead?
		ardumontAuthorUnsubmitted Done Inline Actions Right! ardumont: Right!