Differential D4083 Diff 15212 swh/indexer/storage/writer.py

Changeset View

Standalone View

swh/indexer/storage/writer.py

This file was added.

				# Copyright (C) 2020 The Software Heritage developers
				# See the AUTHORS file at the top-level directory of this distribution
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information

				from typing import Dict, Iterable

				import attr

				try:
				from swh.journal.writer import get_journal_writer
				except ImportError:
				get_journal_writer = None # type: ignore
				# mypy limitation, see https://github.com/python/mypy/issues/1153


				class JournalWriter:
				"""Journal writer storage collaborator. It's in charge of adding objects to
				the journal.

				"""

				def __init__(self, tool_getter, journal_writer):
				self._tool_getter = tool_getter
				if journal_writer:
				if get_journal_writer is None:
				raise EnvironmentError(
				"You need the swh.journal package to use the "
				"journal_writer feature"
				)
				self.journal = get_journal_writer(**journal_writer)
				else:
				self.journal = None

				def write_additions(self, obj_type, entries) -> None:
				if self.journal:
				# usually, all the additions in a batch are from the same indexer,
				# so this cache allows doing a single query for all the entries.
				tool_cache = {}

				for entry in entries:
				# get the tool used to generate this addition
				tool_id = entry.indexer_configuration_id
				if tool_id not in tool_cache:
				tool_cache[tool_id] = self._tool_getter(tool_id)
				entry = attr.evolve(
				entry, tool=tool_cache[tool_id], indexer_configuration_id=None
				)

				# write to kafka
				self.journal.write_addition(obj_type, entry)

				def write_deletions(self, obj_type, entries) -> None:
				if self.journal:
				raise NotImplementedError("deletions are not supported by the journal.")

				def revision_intrinsic_metadata_add(self, metadata: Iterable[Dict]) -> None:
				self.write_additions("revision_intrinsic_metadata", metadata)

				def origin_intrinsic_metadata_add(self, metadata: Iterable[Dict]) -> None:
				self.write_additions("origin_intrinsic_metadata", metadata)
				olasdUnsubmitted Not Done Inline Actions This will flush the journal writer on every message. Please avoid doing that. olasd: This will flush the journal writer on every message. Please avoid doing that.