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

Changeset View

Standalone View

swh/graphql/resolvers/revision.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.swhids import CoreSWHID, ObjectType			from swh.model.swhids import CoreSWHID, ObjectType

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


	class BaseRevisionNode(BaseSWHNode):			class BaseRevisionNode(BaseSWHNode):
				"""
				Base resolver for all the revision nodes
				"""

	def _get_revision_by_id(self, revision_id):			def _get_revision_by_id(self, revision_id):
	return (archive.Archive().get_revisions([revision_id]) or None)[0]			return (archive.Archive().get_revisions([revision_id]) or None)[0]

	@property			@property
	def parentSWHIDs(self): # To support the schema naming convention			def parentSWHIDs(self): # To support the schema naming convention
	return [			return [
	CoreSWHID(object_type=ObjectType.REVISION, object_id=parent_id)			CoreSWHID(object_type=ObjectType.REVISION, object_id=parent_id)
	for parent_id in self._node.parents			for parent_id in self._node.parents
	]			]

	@property			@property
	def directorySWHID(self): # To support the schema naming convention			def directorySWHID(self): # To support the schema naming convention
	""" """
	return CoreSWHID(			return CoreSWHID(
	object_type=ObjectType.DIRECTORY, object_id=self._node.directory			object_type=ObjectType.DIRECTORY, object_id=self._node.directory
	)			)

	@property			@property
	def type(self):			def type(self):
	return self._node.type.value			return self._node.type.value

	def is_type_of(self):			def is_type_of(self):
	"""			# is_type_of is required only when resolving a UNION type
	is_type_of is required only when resolving			# This is for ariadne to return the right type
	a UNION type
	This is for ariadne to return the right type
	"""
	return "Revision"			return "Revision"


	class RevisionNode(BaseRevisionNode):			class RevisionNode(BaseRevisionNode):
	"""			"""
	When the revision is requested directly with its SWHID			Node resolver for a revision requested directly with its SWHID
	"""			"""

	def _get_node_data(self):			def _get_node_data(self):
	return self._get_revision_by_id(self.kwargs.get("swhid").object_id)			return self._get_revision_by_id(self.kwargs.get("swhid").object_id)


	class TargetRevisionNode(BaseRevisionNode):			class TargetRevisionNode(BaseRevisionNode):
	"""			"""
	When a revision is requested as a target			Node resolver for a revision requested as a target

	self.obj could be a snapshotbranch or a release
	self.obj.targetHash is the requested revision id here
	"""			"""

	def _get_node_data(self):			def _get_node_data(self):
				# self.obj could be a snapshotbranch or a release
				# self.obj.targetHash is the requested revision id
	return self._get_revision_by_id(self.obj.targetHash)			return self._get_revision_by_id(self.obj.targetHash)


	class ParentRevisionConnection(BaseConnection):			class ParentRevisionConnection(BaseConnection):
	"""			"""
	When parent revisions is requested from a			Connection resolver for parent revisions in a revision
	revision
	self.obj is the current(child) revision
	self.obj.parentSWHIDs is the list of
	parent SWHIDs
	"""			"""

	_node_class = BaseRevisionNode			_node_class = BaseRevisionNode

	def _get_paged_result(self):			def _get_paged_result(self):
				# self.obj is the current(child) revision
				# self.obj.parentSWHIDs is the list of parent SWHIDs

	# 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 (pagination)			# STORAGE-TODO (pagination)
	parents = archive.Archive().get_revisions(			parents = archive.Archive().get_revisions(
	[x.object_id for x in self.obj.parentSWHIDs]			[x.object_id for x in self.obj.parentSWHIDs]
	)			)
	return utils.paginated(parents, self._get_first_arg(), self._get_after_arg())			return utils.paginated(parents, self._get_first_arg(), self._get_after_arg())


	class LogRevisionConnection(BaseConnection):			class LogRevisionConnection(BaseConnection):
	"""			"""
	When revisionslog is requested from a			Connection resolver for the log (list of revisions) in a revision
	revision
	self.obj is the current revision id
	"""			"""

	_node_class = BaseRevisionNode			_node_class = BaseRevisionNode

	def _get_paged_result(self):			def _get_paged_result(self):
				# self.obj is the current revision id

	# STORAGE-TODO (date in revisionlog is a dict)			# STORAGE-TODO (date in revisionlog is a dict)
	log = archive.Archive().get_revision_log([self.obj.swhid.object_id])			log = archive.Archive().get_revision_log([self.obj.swhid.object_id])
	# 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 (pagination)			# STORAGE-TODO (pagination)
	return utils.paginated(log, self._get_first_arg(), self._get_after_arg())			return utils.paginated(log, self._get_first_arg(), self._get_after_arg())