Add line-level deduplication for NDJSONReader.

dabapush/Reader/NDJSONReader.py

@@ -51,10 +51,15 @@ def __init__(self, config: "NDJSONReaderConfiguration") -> None:
 
     def read(self) -> Iterator[Record]:
         """reads multiple NDJSON files and emits them line by line"""
+
         for file_record in self.records:
-            yield from file_record.split(
-                func=read_and_split, flatten_records=self.config.flatten_dicts
+            filtered_records = filter(
+                lambda x: x not in self.back_log,
+                file_record.split(
+                    func=read_and_split, flatten_records=self.config.flatten_dicts
+                ),
             )
+            yield from filtered_records
 
 
 class NDJSONReaderConfiguration(ReaderConfiguration):
@@ -63,7 +68,7 @@ class NDJSONReaderConfiguration(ReaderConfiguration):
     Attributes
     ----------
     flatten_dicts: bool
-        wether to flatten those nested dictioniaries
+        whether to flatten those nested dicts
 
     """
 

dabapush/Reader/Reader.py

@@ -79,7 +79,7 @@ def records(self) -> Iterator[Record]:
         """Generator for all files matching the pattern in the read_path."""
         if self.log_path.exists():
             with self.log_path.open("rt", encoding="utf8") as f:
-                self.back_log = (ujson.loads(_) for _ in f.readlines())
+                self.back_log = [Record(**ujson.loads(_)) for _ in f.readlines()]
         else:
             self.log_path.touch()
 

tests/Reader/test_NDJSONReader.py

@@ -14,12 +14,12 @@ def data():
     return [{"key": "value"} for _ in range(20)]
 
 
-def test_read(tmp_path: Path, data):  # pylint: disable=W0621
+def test_read(isolated_test_dir: Path, data):  # pylint: disable=W0621
     """Should read the data from the file."""
     reader = NDJSONReader(
-        NDJSONReaderConfiguration("test", read_path=str(tmp_path.resolve()))
+        NDJSONReaderConfiguration("test", read_path=str(isolated_test_dir.resolve()))
     )
-    file_path = tmp_path / "test.ndjson"
+    file_path = isolated_test_dir / "test.ndjson"
     with file_path.open("wt") as file:
         for line in data:
             json.dump(line, file)
@@ -30,3 +30,26 @@ def test_read(tmp_path: Path, data):  # pylint: disable=W0621
     for n, record in enumerate(records):
         assert record.processed_at
         assert record.payload == data[n]
+
+
+def test_read_with_backlog(isolated_test_dir: Path, data):  # pylint: disable=W0621
+    """Should only read the new data."""
+    reader = NDJSONReaderConfiguration(
+        "test", read_path=str(isolated_test_dir.resolve())
+    ).get_instance()
+    file_path = isolated_test_dir / "test.ndjson"
+    with file_path.open("wt") as file:
+        for line in data:
+            json.dump(line, file)
+            file.write("\n")
+
+    records = list(reader.read())
+    assert len(records) == 20
+
+    reader2 = NDJSONReaderConfiguration(
+        "test", read_path=str(isolated_test_dir.resolve())
+    ).get_instance()
+
+    records = list(reader2.read())
+    assert len(records) == 0
+    assert len(reader2.back_log) == 20

tests/conftest.py

@@ -0,0 +1,10 @@
+"""Common fixtures for tests."""
+
+import pytest
+
+
+@pytest.fixture()
+def isolated_test_dir(monkeypatch, tmp_path):
+    """Create an isolated test tmp-directory."""
+    monkeypatch.chdir(tmp_path)
+    yield tmp_path