Differential D6862 Diff 24887 swh/provenance/graph.py

Changeset View

Standalone View

swh/provenance/graph.py

	Show All 17 Lines
	from .model import DirectoryEntry, RevisionEntry			from .model import DirectoryEntry, RevisionEntry

	GRAPH_DURATION_METRIC = "swh_provenance_graph_duration_seconds"			GRAPH_DURATION_METRIC = "swh_provenance_graph_duration_seconds"
	GRAPH_OPERATIONS_METRIC = "swh_provenance_graph_operations_total"			GRAPH_OPERATIONS_METRIC = "swh_provenance_graph_operations_total"

	UTCMIN = datetime.min.replace(tzinfo=timezone.utc)			UTCMIN = datetime.min.replace(tzinfo=timezone.utc)


	class HistoryNode:
	def __init__(
	self, entry: RevisionEntry, is_head: bool = False, in_history: bool = False
	) -> None:
	self.entry = entry
	# A revision is `is_head` if it is directly pointed by an origin (ie. a head
	# revision for some snapshot)
	self.is_head = is_head
	# A revision is `in_history` if it appears in the history graph of an already
	# processed revision in the provenance database
	self.in_history = in_history
	# XXX: the current simplified version of the origin-revision layer algorithm
	# does not use this previous two flags at all. They are kept for now but might
	# be removed in the future (hence, RevisionEntry might be used instead of
	# HistoryNode).

	def __str__(self) -> str:
	return f"<{self.entry}: is_head={self.is_head}, in_history={self.in_history}>"

	def as_dict(self) -> Dict[str, Any]:
	return {
	"rev": hash_to_hex(self.entry.id),
	"is_head": self.is_head,
	"in_history": self.in_history,
	}


	class HistoryGraph:			class HistoryGraph:
	@statsd.timed(metric=GRAPH_DURATION_METRIC, tags={"method": "build_history_graph"})			@statsd.timed(metric=GRAPH_DURATION_METRIC, tags={"method": "build_history_graph"})
	def __init__(			def __init__(
	self,			self,
	provenance: ProvenanceInterface,
	archive: ArchiveInterface,			archive: ArchiveInterface,
	revision: RevisionEntry,			revision: RevisionEntry,
	) -> None:			) -> None:
	self._head = HistoryNode(			self._head = revision
	revision,			self._graph: Dict[RevisionEntry, Set[RevisionEntry]] = {}
	is_head=provenance.revision_visited(revision),
	in_history=provenance.revision_in_history(revision),
	)
	self._graph: Dict[HistoryNode, Set[HistoryNode]] = {}

	stack = [self._head]			stack = [self._head]
	while stack:			while stack:
	current = stack.pop()			current = stack.pop()

	if current not in self._graph:			if current not in self._graph:
	self._graph[current] = set()			self._graph[current] = set()
	current.entry.retrieve_parents(archive)			current.retrieve_parents(archive)
	for parent in current.entry.parents:			for parent in current.parents:
	node = HistoryNode(			self._graph[current].add(parent)
	parent,			stack.append(parent)
	is_head=provenance.revision_visited(parent),
	in_history=provenance.revision_in_history(parent),
	)
	self._graph[current].add(node)
	stack.append(node)

	@property			@property
	def head(self) -> HistoryNode:			def head(self) -> RevisionEntry:
	return self._head			return self._head

	@property			@property
	def parents(self) -> Dict[HistoryNode, Set[HistoryNode]]:			def parents(self) -> Dict[RevisionEntry, Set[RevisionEntry]]:
	return self._graph			return self._graph

	def __str__(self) -> str:			def __str__(self) -> str:
	return f"<HistoryGraph: head={self._head}, graph={self._graph}"			return f"<HistoryGraph: head={self._head}, graph={self._graph}"

	def as_dict(self) -> Dict[str, Any]:			def as_dict(self) -> Dict[str, Any]:
	return {			return {
	"head": self.head.as_dict(),			"head": hash_to_hex(self.head.id),
	"graph": {			"graph": {
	hash_to_hex(node.entry.id): sorted(			hash_to_hex(node.id): sorted(
	[parent.as_dict() for parent in parents],			[hash_to_hex(parent.id) for parent in parents]
	key=lambda d: d["rev"],
	)			)
	for node, parents in self._graph.items()			for node, parents in self._graph.items()
	},			},
	}			}


	class IsochroneNode:			class IsochroneNode:
	def __init__(			def __init__(
	▲ Show 20 Lines • Show All 145 Lines • Show Last 20 Lines