add statements() method to versionedview

nomenklatura/kv.py

@@ -1,14 +1,21 @@
 import redis
-from redis.client import Redis
+import logging
 from typing import Union
 from functools import cache
+from redis.client import Redis
+from fakeredis import FakeStrictRedis
 
 from nomenklatura import settings
 
+log = logging.getLogger(__name__)
+
 
 @cache
 def get_redis() -> "Redis[bytes]":
     """Return a Redis connection configured from the environment."""
+    if settings.TESTING or not len(settings.REDIS_URL.strip()):
+        log.info("Using in-memory key-value store...")
+        return FakeStrictRedis(decode_responses=False)
     db = redis.from_url(settings.REDIS_URL, decode_responses=False)
     # for kvrocks:
     if len(db.config_get("redis-cursor-compatible")):

nomenklatura/settings.py

@@ -7,7 +7,7 @@
 DB_URL = env_str("NOMENKLATURA_DB_URL", f"sqlite:///{DB_PATH.as_posix()}")
 DB_POOL_SIZE = int(env_str("NOMENKLATURA_DB_POOL_SIZE", "5"))
 
-REDIS_URL = env_str("NOMENKLATURA_REDIS_URL", "redis://localhost:6379/0")
+REDIS_URL = env_str("NOMENKLATURA_REDIS_URL", "")
 
 STATEMENT_TABLE = env_str("NOMENKLATURA_STATEMENT_TABLE", "statement")
 STATEMENT_BATCH = int(env_str("NOMENKLATURA_STATEMENT_BATCH", "10000"))

nomenklatura/store/versioned.py

@@ -275,6 +275,24 @@ def get_inverted(self, id: str) -> Generator[Tuple[Property, CE], None, None]:
                 if value == id and prop.reverse is not None:
                     yield prop.reverse, entity
 
+    def statements(self, resolve: bool = False) -> Generator[Statement, None, None]:
+        """Iterate over all statements in the view. If `resolve` is set to `True`,
+        canonical IDs are applied to the statement and its value.
+
+        NOTE: The `external` flag of the view will be used to filter statements, too.
+        """
+        for ds, ver in self.vers:
+            for id in self.store.db.sscan_iter(b(f"ents:{ds}:{ver}")):
+                entity_id = id.decode("utf-8")
+                stmt_key = f"stmt:{ds}:{ver}:{entity_id}"
+                for stmt_text in self.store.db.smembers(b(stmt_key)):
+                    stmt = _unpack_statement(stmt_text, entity_id)
+                    if stmt.external and not self.external:
+                        continue
+                    if resolve:
+                        stmt = self.store.linker.apply_statement(stmt)
+                    yield stmt
+
     def entities(self) -> Generator[CE, None, None]:
         if len(self.vers) == 0:
             return

nomenklatura/versions.py

@@ -58,7 +58,7 @@ def __hash__(self) -> int:
 class VersionHistory(object):
     """A class to represent a history of dataset versions."""
 
-    LENGTH = 300
+    LENGTH = 100
 
     def __init__(self, items: List[Version], max_length: int = LENGTH) -> None:
         self.items = items

tests/store/test_versioned.py

@@ -28,6 +28,7 @@ def test_store_basics(test_dataset: Dataset):
     redis = fakeredis.FakeStrictRedis(version=6, decode_responses=False)
     resolver = Resolver[CompositeEntity]()
     store = VersionedRedisStore(test_dataset, resolver, db=redis)
+    assert len(list(store.view(test_dataset).statements())) == 0
     entity = CompositeEntity.from_data(test_dataset, PERSON)
     entity_ext = CompositeEntity.from_data(test_dataset, PERSON_EXT)
     assert len(list(store.view(test_dataset).entities())) == 0
@@ -36,9 +37,11 @@ def test_store_basics(test_dataset: Dataset):
     writer.flush()
     writer.release()
     assert len(list(store.view(test_dataset).entities())) == 1
+    assert len(list(store.view(test_dataset).statements())) == 3
     writer.add_entity(entity_ext)
     writer.flush()
     assert len(list(store.view(test_dataset).entities())) == 2
+    assert len(list(store.view(test_dataset).statements())) == 5
 
     merged_id = resolver.decide(
         "john-doe",
@@ -49,6 +52,8 @@ def test_store_basics(test_dataset: Dataset):
     store.update(merged_id)
     assert len(list(store.view(test_dataset).entities())) == 1
 
+    assert len(list(store.view(test_dataset).statements())) == 5
+
 
 def test_graph_query(donations_path: Path, test_dataset: Dataset):
     redis = fakeredis.FakeStrictRedis(version=6, decode_responses=False)