Repository: cogent3/cogent3
Pull request #1175 (2022-10-24 06:19:54)

Data store refactor


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index 3c6527d4f..c87ea9590 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -1,7 +1,6 @@
 import inspect
 import json
 import os
-import pathlib
 import re
 import textwrap
 import time
@@ -11,6 +10,7 @@ import types
 from copy import deepcopy
 from enum import Enum
 from functools import wraps
+from pathlib import Path
 from typing import Any, Generator, Tuple
 from uuid import uuid4
 
@@ -399,7 +399,7 @@ class Composable(ComposableType):  # pragma: no cover
                 )
 
             if not logger.log_file_path:
-                src = pathlib.Path(self.data_store.source)
+                src = Path(self.data_store.source)
                 logger.log_file_path = str(src.parent / _make_logfile_name(self))
 
             log_file_path = str(logger.log_file_path)
@@ -422,9 +422,7 @@ class Composable(ComposableType):  # pragma: no cover
         # be logged
         inputs = {}
         for m in dstore:
-            input_id = pathlib.Path(
-                m if isinstance(m, DataStoreMember) else get_data_source(m)
-            )
+            input_id = Path(m if isinstance(m, DataStoreMember) else get_data_source(m))
             suffixes = input_id.suffixes
             input_id = input_id.name.replace("".join(suffixes), "")
             inputs[input_id] = m
@@ -456,7 +454,7 @@ class Composable(ComposableType):  # pragma: no cover
                 else:
                     input_id = get_data_source(result)
                     outcome = result
-                input_id = pathlib.Path(pathlib.Path(input_id))
+                input_id = Path(input_id)
                 suffixes = input_id.suffixes
                 input_id = input_id.name.replace("".join(suffixes), "")
             elif process is not self:
@@ -1354,15 +1352,22 @@ def _apply_to(
     if not self.input:
         raise RuntimeError(f"{self!r} is not part of a composed function")
 
-    if isinstance(dstore, (str, pathlib.Path)):  # one filename
+    if isinstance(dstore, (str, Path)):  # one filename
         dstore = [dstore]
 
     # todo run check on dstore to make sure we have identifiers for writing output
     # our default function should fail if a source cannot be determined, i.e. does not return None
 
     # todo this should fail if somebody provides data that cannot produce a unique_id
-    # skip records already in data_store
-    dstore = [e for e in dstore if get_data_source(e) not in self.data_store]
+    inputs = {}
+    for m in dstore:
+        input_id = Path(m if isinstance(m, DataStoreMember) else get_data_source(m))
+        suffixes = input_id.suffixes
+        input_id = input_id.name.replace("".join(suffixes), "")
+        if input_id in inputs:
+            raise ValueError("non-unique identifier detected in data")
+        inputs[input_id] = input_id
+
     if not dstore:  # this should just return datastore, because if all jobs are done!
         raise ValueError("dstore is empty")
 
@@ -1383,7 +1388,7 @@ def _apply_to(
 
     taken = time.time() - start
     logger.log_message(f"{taken}", label="TIME TAKEN")
-    log_file_path = pathlib.Path(logger.log_file_path)
+    log_file_path = Path(logger.log_file_path)
     logger.shutdown()
     self.data_store.write_log(log_file_path.name, log_file_path.read_text())
     if cleanup:
@@ -1396,7 +1401,7 @@ def _set_logger(self, logger=None):
     if logger is None:
         logger = CachingLogger(create_dir=True)
     if not logger.log_file_path:
-        src = pathlib.Path(self.data_store.source).parent
+        src = Path(self.data_store.source).parent
         logger.log_file_path = str(src / _make_logfile_name(self))
     self.logger = logger
 


Test patch:
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index 93f6bfe65..d05d682d9 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -235,19 +235,21 @@ def test_apply_to_strings(tmp_dir, klass):
     assert len(process.data_store.logs) == 1
 
 
-@pytest.mark.xfail
-def test_apply_to_non_unique_identifiers():
+def test_apply_to_non_unique_identifiers(tmp_dir):
     """should fail if non-unique names"""
     dstore = [
         "brca1.bats.fasta",
         "brca1.apes.fasta",
     ]
-    with TemporaryDirectory(dir=".") as dirname:
-        reader = io_app.load_aligned(format="fasta", moltype="dna")
-        min_length = sample_app.min_length(10)
-        process = reader + min_length
-        with pytest.raises(ValueError):
-            process.apply_to(dstore)
+    reader = io_app.load_aligned(format="fasta", moltype="dna")
+    min_length = sample_app.min_length(10)
+    outpath = tmp_dir / "test_apply_to_non_unique_identifiers"
+    writer = io_app.write_seqs_new(
+        DataStoreDirectory(outpath, if_dest_exists="overwrite", suffix="fasta")
+    )
+    process = reader + min_length + writer
+    with pytest.raises(ValueError):
+        process.apply_to(dstore)
 
 
 def test_apply_to_logging(tmp_dir):