Differential D7942 Diff 28604 swh/graphql/resolvers/snapshot.py

Changeset View

Standalone View

swh/graphql/resolvers/snapshot.py

	# Copyright (C) 2022 The Software Heritage developers			# Copyright (C) 2022 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 swh.graphql.backends import archive			from swh.graphql.backends import archive
	from swh.graphql.utils import utils			from swh.graphql.utils import utils
	from swh.model.model import Snapshot			from swh.model.model import Snapshot

	from .base_connection import BaseConnection			from .base_connection import BaseConnection
	from .base_node import BaseSWHNode			from .base_node import BaseSWHNode


	class BaseSnapshotNode(BaseSWHNode):			class BaseSnapshotNode(BaseSWHNode):
				"""
				Base resolver for all the snapshot nodes
				"""

	def _get_snapshot_by_id(self, snapshot_id):			def _get_snapshot_by_id(self, snapshot_id):
	# Return a Snapshot model object			# Return a Snapshot model object
	# branches is initialized as empty			# branches is initialized as empty
	# Same pattern is used in directory			# Same pattern is used in directory
	return Snapshot(id=snapshot_id, branches={})			return Snapshot(id=snapshot_id, branches={})


	class SnapshotNode(BaseSnapshotNode):			class SnapshotNode(BaseSnapshotNode):
	"""			"""
	For directly accessing a snapshot with its SWHID			Node resolver for a snapshot requested directly with its SWHID
	"""			"""

	def _get_node_data(self):			def _get_node_data(self):
	""" """			""" """
	snapshot_id = self.kwargs.get("swhid").object_id			snapshot_id = self.kwargs.get("swhid").object_id
	if archive.Archive().is_snapshot_available([snapshot_id]):			if archive.Archive().is_snapshot_available([snapshot_id]):
	return self._get_snapshot_by_id(snapshot_id)			return self._get_snapshot_by_id(snapshot_id)
	return None			return None


	class VisitSnapshotNode(BaseSnapshotNode):			class VisitSnapshotNode(BaseSnapshotNode):
	"""			"""
	For accessing a snapshot from a visitstatus type			Node resolver for a snapshot requested from a visit-status
	"""			"""

	def _get_node_data(self):			def _get_node_data(self):
	"""			# self.obj is visitstatus here
	self.obj is visitstatus here			# self.obj.snapshotSWHID is the requested snapshot SWHID
	self.obj.snapshotSWHID is the requested snapshot SWHID
	"""
	snapshot_id = self.obj.snapshotSWHID.object_id			snapshot_id = self.obj.snapshotSWHID.object_id
	return self._get_snapshot_by_id(snapshot_id)			return self._get_snapshot_by_id(snapshot_id)


	class OriginSnapshotConnection(BaseConnection):			class OriginSnapshotConnection(BaseConnection):
				"""
				Connection resolver for the snapshots in an origin
				"""

	_node_class = BaseSnapshotNode			_node_class = BaseSnapshotNode

	def _get_paged_result(self):			def _get_paged_result(self):
	""" """			# self.obj is origin here
	results = archive.Archive().get_origin_snapshots(self.obj.url)			results = archive.Archive().get_origin_snapshots(self.obj.url)
	snapshots = [Snapshot(id=snapshot, branches={}) for snapshot in results]			snapshots = [Snapshot(id=snapshot, branches={}) for snapshot in results]
	# FIXME, using dummy(local) pagination, move pagination to backend			# FIXME, using dummy(local) pagination, move pagination to backend
	# To remove localpagination, just drop the paginated call			# To remove localpagination, just drop the paginated call
	# STORAGE-TODO			# STORAGE-TODO
	return utils.paginated(snapshots, self._get_first_arg(), self._get_after_arg())			return utils.paginated(snapshots, self._get_first_arg(), self._get_after_arg())