Differential D4189 Diff 14784 swh/indexer/origin_head.py

Changeset View

Standalone View

swh/indexer/origin_head.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

	import logging			import logging
	import re			import re
	from typing import Any, Dict, List, Optional, Tuple, Union			from typing import Any, Dict, List, Tuple, Union

	import click			import click

	from swh.indexer.indexer import OriginIndexer			from swh.indexer.indexer import OriginIndexer
	from swh.model.model import SnapshotBranch, TargetType			from swh.model.model import SnapshotBranch, 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.algos.snapshot import snapshot_get_all_branches			from swh.storage.algos.snapshot import snapshot_get_all_branches


	class OriginHeadIndexer(OriginIndexer[Optional[Dict]]):			class OriginHeadIndexer(OriginIndexer[Dict]):
	"""Origin-level indexer.			"""Origin-level indexer.

	This indexer is in charge of looking up the revision that acts as the			This indexer is in charge of looking up the revision that acts as the
	"head" of an origin.			"head" of an origin.

	In git, this is usually the commit pointed to by the 'master' branch."""			In git, this is usually the commit pointed to by the 'master' branch."""

	USE_TOOLS = False			USE_TOOLS = False

	def persist_index_computations(			def persist_index_computations(
	self, results: Any, policy_update: str			self, results: Any, policy_update: str
	) -> Dict[str, int]:			) -> Dict[str, int]:
	"""Do nothing. The indexer's results are not persistent, they			"""Do nothing. The indexer's results are not persistent, they
	should only be piped to another indexer."""			should only be piped to another indexer."""
	return {}			return {}

	# Dispatch			# Dispatch

	def index(self, origin_url):			def index(self, id: str, data: None = None, **kwargs) -> List[Dict]:
				origin_url = id
	visit_and_status = origin_get_latest_visit_status(			visit_and_status = origin_get_latest_visit_status(
	self.storage, origin_url, allowed_statuses=["full"], require_snapshot=True			self.storage, origin_url, allowed_statuses=["full"], require_snapshot=True
	)			)
	if not visit_and_status:			if not visit_and_status:
	return []			return []
	visit, visit_status = visit_and_status			visit, visit_status = visit_and_status
				assert visit_status.snapshot is not None
				ardumontUnsubmitted Done Inline Actions Please, add a test? ardumont: Please, add a test?
				vlorentzAuthorUnsubmitted Done Inline Actions actually, I'll make it an assert, because `origin_get_latest_visit_status(require_snapshot=True)` can't return a `visit_status` with a none snapshot. vlorentz: actually, I'll make it an assert, because `origin_get_latest_visit_status…
	snapshot = snapshot_get_all_branches(self.storage, visit_status.snapshot)			snapshot = snapshot_get_all_branches(self.storage, visit_status.snapshot)
	if snapshot is None:			if snapshot is None:
	return []			return []
	method = getattr(			method = getattr(
	self, "_try_get_%s_head" % visit.type, self._try_get_head_generic			self, "_try_get_%s_head" % visit.type, self._try_get_head_generic
	)			)

	rev_id = method(snapshot.branches)			rev_id = method(snapshot.branches)
	▲ Show 20 Lines • Show All 104 Lines • Show Last 20 Lines