⚡️ (manager) Don't reiterate store at finish

Author: simonwoerpel

ingestors/manager.py

@@ -3,19 +3,19 @@
 from functools import cache
 from tempfile import mkdtemp
 from timeit import default_timer
-from typing import Any, Generator
+from typing import Any
 
 import magic
 from banal import ensure_list
 from followthemoney import model
 from followthemoney.helpers import entity_filename
 from followthemoney.namespace import Namespace
-from followthemoney.proxy import EntityProxy
 from ftmq.store.fragments import get_fragments
 from ftmq.store.fragments.utils import safe_fragment
 from normality import stringify
 from openaleph_procrastinate import defer
-from procrastinate import App
+from openaleph_procrastinate.app import App
+from openaleph_procrastinate.util import make_checksum_entity
 from prometheus_client import Counter, Histogram
 from rigour.mime import normalize_mimetype
 from servicelayer.archive import init_archive
@@ -106,7 +106,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 = set()
+        self.emitted = []
         self.archive = get_archive()
 
     def make_entity(self, schema, parent=None):
@@ -138,7 +138,7 @@ def apply_context(self, entity, source):
     def emit_entity(self, entity, fragment=None):
         entity = self.ns.apply(entity)
         self.writer.put(entity.to_dict(), fragment)
-        self.emitted.add(entity.id)
+        self.emitted.append(make_checksum_entity(entity, quiet=True))
 
     def emit_text_fragment(self, entity, texts, fragment):
         texts = [t for t in ensure_list(texts) if filter_text(t)]
@@ -259,6 +259,3 @@ def delegate(self, ingestor_class, file_path, entity):
     def close(self):
         self.writer.flush()
         remove_directory(self.work_path)
-
-    def iterate_emitted(self) -> Generator[EntityProxy, None, None]:
-        yield from self.db.iterate(self.emitted)

ingestors/tasks.py

@@ -39,13 +39,16 @@ def ingest(job: DatasetJob) -> None:
     finally:
         manager.close()
 
-    for entity in manager.iterate_emitted():
+    for entity in manager.emitted:
         if entity.schema.is_a("Analyzable"):
             to_analyze.append(entity)
 
         to_index.append(entity)
 
-    job.log.info(f"Emitted {len(manager.emitted)} entities.", emitted=manager.emitted)
+    job.log.info(
+        f"Emitted {len(manager.emitted)} entities.",
+        emitted=[e.id for e in manager.emitted],
+    )
     if to_analyze:
         defer.analyze(app, job.dataset, to_analyze, batch=job.batch, **job.context)
     if to_index: