Differential D6249 Diff 22725 swh/storage/in_memory.py

Changeset View

Standalone View

swh/storage/in_memory.py

# Copyright (C) 2015-2020 The Software Heritage developers		# Copyright (C) 2015-2021 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 collections import defaultdict		from collections import defaultdict
import datetime		import datetime
import functools		import functools
import itertools		import itertools
▲ Show 20 Lines • Show All 691 Lines • ▼ Show 20 Lines	) -> Optional[ExtIDRow]:
extid=extid,		extid=extid,
extid_version=extid_version,		extid_version=extid_version,
target_type=target.object_type.value,		target_type=target.object_type.value,
target=target.object_id,		target=target.object_id,
)		)
)		)
return self._extid.get_from_primary_key(primary_key)		return self._extid.get_from_primary_key(primary_key)

def extid_get_from_extid(self, extid_type: str, extid: bytes) -> Iterable[ExtIDRow]:		def extid_get_from_extid(
		self, extid_type: str, extid: bytes,
		) -> Iterable[ExtIDRow]:
return (		return (
row		row
for pk, row in self._extid.iter_all()		for pk, row in self._extid.iter_all()
if row.extid_type == extid_type and row.extid == extid		if row.extid_type == extid_type and row.extid == extid
)		)

def extid_get_from_target(		def extid_get_from_extid_and_version(
self, target_type: str, target: bytes		self, extid_type: str, extid: bytes, extid_version: int,
		) -> Iterable[ExtIDRow]:
		return (
		row
		for pk, row in self._extid.iter_all()
		if row.extid_type == extid_type
		and row.extid == extid
		and (extid_version is None or row.extid_version == extid_version)
		)

		def _extid_get_from_target_with_type_and_version(
		self, target_type: str, target: bytes, extid_type: str, extid_version: int,
		) -> Iterable[ExtIDRow]:
		return (
		row
		for pk, row in self._extid.iter_all()
		if row.target_type == target_type
		and row.target == target
		and row.extid_version == extid_version
		and row.extid_type == extid_type
		)

		def _extid_get_from_target(
		self, target_type: str, target: bytes,
) -> Iterable[ExtIDRow]:		) -> Iterable[ExtIDRow]:
return (		return (
row		row
for pk, row in self._extid.iter_all()		for pk, row in self._extid.iter_all()
if row.target_type == target_type and row.target == target		if row.target_type == target_type and row.target == target
)		)
		vlorentzUnsubmitted Done Inline Actions what about splitting this into two functions (one for each case), to make it more readable? vlorentz: what about splitting this into two functions (one for each case), to make it more readable?
		ardumontAuthorUnsubmitted Done Inline Actions sure ardumont: sure

		def extid_get_from_target(
		self,
		target_type: str,
		target: bytes,
		extid_type: Optional[str] = None,
		extid_version: Optional[int] = None,
		) -> Iterable[ExtIDRow]:
		if (extid_version is not None and extid_type is None) or (
		extid_version is None and extid_type is not None
		):
		raise ValueError("You must provide both extid_type and extid_version")

		if extid_type is not None and extid_version is not None:
		extids = self._extid_get_from_target_with_type_and_version(
		target_type, target, extid_type, extid_version
		)
		else:
		extids = self._extid_get_from_target(target_type, target)
		return extids


class InMemoryStorage(CassandraStorage):		class InMemoryStorage(CassandraStorage):
_cql_runner: InMemoryCqlRunner # type: ignore		_cql_runner: InMemoryCqlRunner # type: ignore

def __init__(self, journal_writer=None):		def __init__(self, journal_writer=None):
self.reset()		self.reset()
self.journal_writer = JournalWriter(journal_writer)		self.journal_writer = JournalWriter(journal_writer)
self._allow_overwrite = False		self._allow_overwrite = False
self._directory_entries_insert_algo = "one-by-one"		self._directory_entries_insert_algo = "one-by-one"

def reset(self):		def reset(self):
self._cql_runner = InMemoryCqlRunner()		self._cql_runner = InMemoryCqlRunner()
self.objstorage = ObjStorage({"cls": "memory"})		self.objstorage = ObjStorage({"cls": "memory"})

def check_config(self, *, check_write: bool) -> bool:		def check_config(self, *, check_write: bool) -> bool:
return True		return True