Repository: cogent3/cogent3
Pull request #1254 (2022-12-29 07:09:17)

MAINT: upgrade integration of new data stores and writer apps


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index 643a00266..50d0a61ba 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -35,7 +35,7 @@ from .data_store import (
     DataStoreMember,
     WritableDirectoryDataStore,
 )
-from .data_store_new import get_data_source
+from .data_store_new import DataMember, get_data_source, get_unique_id
 
 
 __author__ = "Gavin Huttley"
@@ -1346,7 +1346,7 @@ def is_composable(obj):
 def _apply_to(
     self,
     dstore,
-    id_from_source: callable = get_data_source,
+    id_from_source: callable = get_unique_id,
     parallel=False,
     par_kw=None,
     logger=None,
@@ -1397,14 +1397,13 @@ def _apply_to(
     # todo this should fail if somebody provides data that cannot produce a unique_id
     inputs = {}
     for m in dstore:
-        input_id = Path(m if isinstance(m, DataStoreMember) else id_from_source(m))
-        suffixes = input_id.suffixes
-        input_id = input_id.name.replace("".join(suffixes), "")
+        input_id = Path(m.unique_id) if isinstance(m, DataMember) else m
+        input_id = id_from_source(input_id)
         if input_id in inputs:
             raise ValueError("non-unique identifier detected in data")
         if input_id in self.data_store:  # todo write a test
             continue
-        inputs[input_id] = input_id
+        inputs[input_id] = m
 
     if not dstore:  # this should just return datastore, because if all jobs are done!
         raise ValueError("dstore is empty")
@@ -1415,7 +1414,7 @@ def _apply_to(
     logger.log_message(str(self), label="composable function")
     logger.log_versions(["cogent3"])
 
-    inputs = _proxy_input(dstore)
+    inputs = _proxy_input(inputs.values())
     for result in self.as_completed(inputs, parallel=parallel, par_kw=par_kw):
         member = self.main(
             data=result.obj, identifier=id_from_source(result.source)
diff --git a/src/cogent3/app/data_store_new.py b/src/cogent3/app/data_store_new.py
index 3fcb15ee2..111b7c16a 100644
--- a/src/cogent3/app/data_store_new.py
+++ b/src/cogent3/app/data_store_new.py
@@ -494,7 +494,6 @@ class DataStoreDirectory(DataStoreABC):
         return member
 
     def write_not_completed(self, *, unique_id: str, data: str) -> DataMember:
-        (self.source / _NOT_COMPLETED_TABLE).mkdir(parents=True, exist_ok=True)
         member = self._write(
             subdir=_NOT_COMPLETED_TABLE, unique_id=unique_id, suffix="json", data=data
         )
@@ -503,7 +502,6 @@ class DataStoreDirectory(DataStoreABC):
         return member
 
     def write_log(self, *, unique_id: str, data: str) -> None:
-        (self.source / _LOG_TABLE).mkdir(parents=True, exist_ok=True)        
         _ = self._write(subdir=_LOG_TABLE, unique_id=unique_id, suffix="log", data=data)
 
     def md5(self, unique_id: str) -> Union[str, NoneType]:
@@ -524,6 +522,13 @@ class DataStoreDirectory(DataStoreABC):
         return path.read_text() if path.exists() else None
 
 
+def get_unique_id(name: str) -> str:
+    """strips any format suffixes from name"""
+    name = get_data_source(name)
+    suffixes = ".".join(sfx for sfx in get_format_suffixes(name) if sfx)
+    return re.sub(fr"[.]{suffixes}$", "", name)
+
+
 @singledispatch
 def get_data_source(data) -> str:
     source = getattr(data, "source", None)
diff --git a/src/cogent3/app/io_new.py b/src/cogent3/app/io_new.py
index 986db1200..b02baab6e 100644
--- a/src/cogent3/app/io_new.py
+++ b/src/cogent3/app/io_new.py
@@ -35,7 +35,7 @@ from .data_store_new import (
     DataStoreABC,
     DataStoreDirectory,
     Mode,
-    get_data_source,
+    get_unique_id,
     load_record_from_json,
     make_record_for_json,
 )
@@ -468,7 +468,7 @@ class write_json:
     def main(
         self, data: SerialisableType, identifier: Optional[str] = None
     ) -> IdentifierType:
-        identifier = identifier or get_data_source(data)
+        identifier = identifier or get_unique_id(data)
         if isinstance(data, NotCompleted):
             return self.data_store.write_not_completed(
                 unique_id=f"{identifier}.json", data=data.to_json()
@@ -492,7 +492,7 @@ class write_seqs:  # todo docstring
     def main(
         self, data: SeqsCollectionType, identifier: Optional[str] = None
     ) -> IdentifierType:
-        identifier = identifier or get_data_source(data)
+        identifier = identifier or get_unique_id(data)
         if isinstance(data, NotCompleted):
             return self.data_store.write_not_completed(
                 unique_id=f"{identifier}.json", data=data.to_json()
@@ -511,7 +511,7 @@ class write_tabular:  # todo doctsring
     def main(
         self, data: TabularType, identifier: Optional[str] = None
     ) -> IdentifierType:
-        identifier = identifier or get_data_source(data)
+        identifier = identifier or get_unique_id(data)
         if isinstance(data, NotCompleted):
             return self.data_store.write_not_completed(
                 unique_id=f"{identifier}.json", data=data.to_json()
@@ -548,7 +548,7 @@ class write_db:
         -------
         identifier
         """
-        identifier = identifier or get_data_source(data)
+        identifier = identifier or get_unique_id(data)
         blob = self.serialiser(data)
 
         if isinstance(data, NotCompleted):


Test patch:
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index b942b56c3..dbe3684e7 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -16,8 +16,7 @@ from scitrack import CachingLogger
 
 from cogent3 import get_app, make_aligned_seqs, open_data_store
 from cogent3.app import align, evo
-from cogent3.app import io as io_app
-from cogent3.app import io_new as io_app_new
+from cogent3.app import io_new as io_app
 from cogent3.app import sample as sample_app
 from cogent3.app import translate, tree
 from cogent3.app import typing as c3types
@@ -342,7 +341,7 @@ def test_apply_to_strings(tmp_dir, klass):
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     min_length = sample_app.min_length(10)
     outpath = tmp_dir / "test_apply_to_strings"
-    writer = io_app_new.write_seqs(klass(outpath, mode=OVERWRITE, suffix="fasta"))
+    writer = io_app.write_seqs(klass(outpath, mode=OVERWRITE, suffix="fasta"))
     process = reader + min_length + writer
     # create paths as strings
     _ = process.apply_to(dstore, id_from_source=get_data_source)
@@ -352,13 +351,13 @@ def test_apply_to_strings(tmp_dir, klass):
 def test_apply_to_non_unique_identifiers(tmp_dir):
     """should fail if non-unique names"""
     dstore = [
-        "brca1.bats.fasta",
-        "brca1.apes.fasta",
+        "brca1.fasta",
+        "brca1.fasta",
     ]
     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_new.write_seqs(
+    writer = io_app.write_seqs(
         DataStoreDirectory(outpath, mode=OVERWRITE, suffix="fasta")
     )
     process = reader + min_length + writer
@@ -371,13 +370,12 @@ def test_apply_to_logging(tmp_dir):
     dstore = open_data_store(DATA_DIR, suffix="fasta", limit=3)
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     min_length = sample_app.min_length(10)
-    outpath = os.path.join(os.getcwd(), tmp_dir, "delme.tinydb")
-    writer = io_app.write_db(outpath)
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    writer = io_app.write_db(out_dstore)
     process = reader + min_length + writer
     r = process.apply_to(dstore, show_progress=False)
     # always creates a log
     assert len(process.data_store.logs) == 1
-    process.data_store.close()
 
 
 def test_apply_to_logger(tmp_dir):
@@ -386,26 +384,26 @@ def test_apply_to_logger(tmp_dir):
     LOGGER = CachingLogger()
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     min_length = sample_app.min_length(10)
-    outpath = os.path.join(os.getcwd(), tmp_dir, "delme.tinydb")
-    writer = io_app.write_db(outpath)
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    writer = io_app.write_db(out_dstore)
     process = reader + min_length + writer
-    r = process.apply_to(dstore, show_progress=False, logger=LOGGER)
+    process.apply_to(dstore, show_progress=False, logger=LOGGER)
     assert len(process.data_store.logs) == 1
-    process.data_store.close()
 
 
-def test_apply_to_invalid_logger(tmp_dir):
+@pytest.mark.parametrize("logger_val", (True, "somepath.log"))
+def test_apply_to_invalid_logger(tmp_dir, logger_val):
     """incorrect logger value raises TypeError"""
     dstore = open_data_store(DATA_DIR, suffix="fasta", limit=3)
-    for logger_val in (True, "somepath.log"):
-        reader = io_app.load_aligned(format="fasta", moltype="dna")
-        min_length = sample_app.min_length(10)
-        outpath = os.path.join(os.getcwd(), tmp_dir, "delme.tinydb")
-        writer = io_app.write_db(outpath)
-        process = reader + min_length + writer
-        with pytest.raises(TypeError):
-            process.apply_to(dstore, show_progress=False, logger=logger_val)
-        process.data_store.close()
+    reader = io_app.load_aligned(format="fasta", moltype="dna")
+    min_length = sample_app.min_length(10)
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    writer = io_app.write_db(out_dstore)
+    process = reader + min_length + writer
+    with pytest.raises(TypeError):
+        process.apply_to(dstore, show_progress=False, logger=logger_val)
+
+    out_dstore.close()
 
 
 def test_apply_to_not_completed(tmp_dir):
@@ -414,32 +412,28 @@ def test_apply_to_not_completed(tmp_dir):
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     # trigger creation of notcompleted
     min_length = sample_app.min_length(3000)
-    outpath = os.path.join(os.getcwd(), tmp_dir, "delme.tinydb")
-    writer = io_app.write_db(outpath)
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    writer = io_app.write_db(out_dstore)
     process = reader + min_length + writer
-    r = process.apply_to(dstore, show_progress=False)
-    assert len(process.data_store.incomplete) == 3
-    process.data_store.close()
+    process.apply_to(dstore, show_progress=False)
+    assert len(out_dstore.not_completed) == 3
 
 
 def test_apply_to_not_partially_done(tmp_dir):
     """correctly applies process when result already partially done"""
     dstore = open_data_store(DATA_DIR, suffix="fasta")
     num_records = len(dstore)
-    dirname = Path(tmp_dir)
     reader = io_app.load_aligned(format="fasta", moltype="dna")
-    outpath = dirname / "delme.tinydb"
-    writer = io_app.write_db(outpath)
-    _ = writer(reader(dstore[0]))
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    writer = io_app.write_db(out_dstore)
+    _ = writer(reader(dstore[0]))  # doing the first one
     writer.data_store.close()
 
-    writer = io_app.write_db(outpath, if_exists="ignore")
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="a")
+    writer = io_app.write_db(out_dstore)
     process = reader + writer
     _ = process.apply_to(dstore, show_progress=False)
-    writer.data_store.close()
-    dstore = open_data_store(outpath)
-    assert len(dstore) == num_records
-    dstore.close()
+    assert len(out_dstore) == num_records
 
 
 def test_err_result():
@@ -1164,22 +1158,29 @@ def test_complex_type_allowed_depths(hint):
 
 
 def test_apply_to_only_appends(half_dstore1, half_dstore2):
-    half_dstore1._mode = APPEND
+    half_dstore1 = open_data_store(
+        half_dstore1.source, suffix=half_dstore1.suffix, mode=APPEND
+    )
     reader1 = io_app.load_aligned(format="fasta", moltype="dna")
     min_length1 = sample_app.min_length(10)
-    writer1 = io_app_new.write_seqs(half_dstore1)
+    writer1 = io_app.write_seqs(half_dstore1)
     process1 = reader1 + min_length1 + writer1
+
     # create paths as strings
-    dstore1 = open_data_store(half_dstore1.source, suffix="fasta")
-    dstore1 = [str(Path(m.data_store.source) / m.unique_id) for m in dstore1]
-    # check fail on append the same records
-    with pytest.raises(IOError):
+    dstore1 = [
+        str(Path(m.data_store.source) / m.unique_id) for m in half_dstore1.completed
+    ]
+    # check fail on all the same records
+    with pytest.raises(ValueError):
         _ = process1.apply_to(dstore1, id_from_source=get_data_source)
 
-    half_dstore2._mode = APPEND
+    half_dstore2 = open_data_store(
+        half_dstore2.source, suffix=half_dstore2.suffix, mode=APPEND
+    )
+
     reader2 = io_app.load_aligned(format="fasta", moltype="dna")
     min_length2 = sample_app.min_length(10)
-    writer2 = io_app_new.write_seqs(half_dstore2)
+    writer2 = io_app.write_seqs(half_dstore2)
     process2 = reader2 + min_length2 + writer2
     # check not fail on append new records
     _ = process2.apply_to(dstore1, id_from_source=get_data_source)
diff --git a/tests/test_app/test_data_store_new.py b/tests/test_app/test_data_store_new.py
index 6036899fa..f8b8ab333 100644
--- a/tests/test_app/test_data_store_new.py
+++ b/tests/test_app/test_data_store_new.py
@@ -9,8 +9,6 @@ import pytest
 
 from scitrack import get_text_hexdigest
 
-import cogent3.app.io_new as io_app
-
 from cogent3.app.composable import NotCompleted
 from cogent3.app.data_store_new import (
     _MD5_TABLE,
@@ -23,6 +21,7 @@ from cogent3.app.data_store_new import (
     convert_directory_datastore,
     convert_tinydb_to_sqlite,
     get_data_source,
+    get_unique_id,
     load_record_from_json,
 )
 from cogent3.util.table import Table
@@ -546,6 +545,14 @@ def test_get_data_source_dict(container_type, source_stype):
     assert got == "path.txt"
 
 
+@pytest.mark.parametrize(
+    "name", ("path/name.txt", "path/name.gz", "path/name.fasta.gz", "name.fasta.gz")
+)
+def test_get_unique_id(name):
+    got = get_unique_id(name)
+    assert got == "name"
+
+
 @pytest.mark.parametrize("data", (dict(), set(), dict(info=dict())))
 def test_get_data_source_none(data):
     assert get_data_source(data) is None
@@ -576,15 +583,3 @@ def test_load_record_from_json():
         assert Id == "some.json"
         assert data_ == expected
         assert compl == True
-
-
-def test_write_read_not_completed(nc_dstore):
-    nc_dstore.drop_not_completed()
-    assert len(nc_dstore.not_completed) == 0
-    nc = NotCompleted("ERROR", "test", "for tracing", source="blah")
-    writer = io_app.write_seqs(data_store=nc_dstore)
-    writer.main(nc, identifier="blah")
-    assert len(nc_dstore.not_completed) == 1
-    got = nc_dstore.not_completed[0].read()
-    assert nc.to_json() == got
-
diff --git a/tests/test_app/test_evo.py b/tests/test_app/test_evo.py
index 55624df9a..d173eb2aa 100644
--- a/tests/test_app/test_evo.py
+++ b/tests/test_app/test_evo.py
@@ -1,3 +1,5 @@
+import pathlib
+
 from os.path import dirname, join
 from tempfile import TemporaryDirectory
 from unittest import TestCase, main
@@ -5,9 +7,14 @@ from unittest.mock import MagicMock
 
 from numpy.testing import assert_allclose, assert_raises
 
-from cogent3 import load_aligned_seqs, make_aligned_seqs, make_tree
+from cogent3 import (
+    load_aligned_seqs,
+    make_aligned_seqs,
+    make_tree,
+    open_data_store,
+)
 from cogent3.app import evo as evo_app
-from cogent3.app import io
+from cogent3.app import io_new
 from cogent3.app.composable import NotCompleted
 from cogent3.app.result import (
     hypothesis_result,
@@ -866,8 +873,10 @@ class TestBootstrap(TestCase):
         m2 = evo_app.model("HKY85")
         hyp = evo_app.hypothesis(m1, m2)
         with TemporaryDirectory(dir=".") as dirname:
-            path = join(dirname, "delme.tinydb")
-            _ = io.load_db() + evo_app.bootstrap(hyp, num_reps=2) + io.write_db(path)
+            dirname = pathlib.Path(dirname)
+            out_dstore = open_data_store(dirname / "delme.sqlitedb", mode="w")
+            writer = io_new.write_db(out_dstore)
+            _ = io_new.load_db() + evo_app.bootstrap(hyp, num_reps=2) + writer
 
 
 if __name__ == "__main__":
diff --git a/tests/test_app/test_io_new.py b/tests/test_app/test_io_new.py
index ac1d7e746..f7f8c85d9 100644
--- a/tests/test_app/test_io_new.py
+++ b/tests/test_app/test_io_new.py
@@ -452,7 +452,7 @@ def test_write_db_load_db(fasta_dir, tmp_dir):
     assert data_store.record_type == get_object_provenance(orig)
 
 
-def test_write_read_db_not_completed(tmp_dir):
+def test_write_db_not_completed(tmp_dir):
     from cogent3.app.sqlite_data_store import DataStoreSqlite
 
     nc = NotCompleted("ERROR", "test", "for tracing", source="blah")