🐛 (manager) Properly iterate through emitted memory store

Author: simonwoerpel

ingestors/manager.py

@@ -7,7 +7,7 @@
 
 import magic
 from banal import ensure_list
-from followthemoney import StatementEntity, model
+from followthemoney import DefaultDataset, StatementEntity, model
 from followthemoney.helpers import entity_filename
 from followthemoney.namespace import Namespace
 from ftmq.store.fragments import get_fragments
@@ -107,7 +107,7 @@ def __init__(self, app: App, dataset: str, context: dict[str, Any]):
         self.context = context
         self.ns = Namespace(self.context["namespace"])
         self.work_path = ensure_path(mkdtemp(prefix="ingestor-"))
-        self.emitted = MemoryStore()
+        self.emitted = MemoryStore(dataset=DefaultDataset)
         self.archive = get_archive()
 
     def make_entity(self, schema, parent=None):
@@ -261,3 +261,6 @@ def delegate(self, ingestor_class, file_path, entity):
     def close(self):
         self.writer.flush()
         remove_directory(self.work_path)
+
+    def get_emitted(self) -> list[StatementEntity]:
+        return list(self.emitted.iterate(dataset=DefaultDataset))

ingestors/tasks.py

@@ -39,7 +39,7 @@ def ingest(job: DatasetJob) -> None:
     finally:
         manager.close()
 
-    emitted = list(manager.emitted.iterate())
+    emitted = manager.get_emitted()
     for entity in emitted:
         if entity.schema.is_a("Analyzable"):
             to_analyze.append(entity)
@@ -77,6 +77,6 @@ def ingest_path(
             manager.queue_entity(entity)
         if path.is_dir():
             DirectoryIngestor.crawl(manager, path)
-    emitted = list(manager.emitted.iterate())
+    emitted = manager.get_emitted()
     log.info(f"Emitted {len(emitted)} entities.", emitted=[e.id for e in emitted])
     manager.close()

tests/support.py

@@ -68,7 +68,7 @@ def fixture(self, fixture_path):
         return path, entity
 
     def get_emitted(self, schema=None):
-        entities = list(self.manager.db.iterate())
+        entities = self.manager.get_emitted()
         if schema is not None:
             entities = [e for e in entities if e.schema.is_a(schema)]
         return entities