Repository: cogent3/cogent3
Pull request #1235 (2022-12-16 04:33:25)

ENH: connected up new data stores and apps


Patch:
diff --git a/src/cogent3/app/__init__.py b/src/cogent3/app/__init__.py
index b41f5bd65..f0ff040b8 100644
--- a/src/cogent3/app/__init__.py
+++ b/src/cogent3/app/__init__.py
@@ -63,8 +63,14 @@ def available_apps():
     for name in __all__:
         importlib.import_module(f"cogent3.app.{name}")
 
+    # exclude apps from deprecated modules
+    deprecated = ["cogent3.app.io."]
+
     rows = []
     for app, is_comp in __app_registry.items():
+        if any(app.startswith(d) for d in deprecated):
+            continue
+
         with contextlib.suppress(AttributeError):
             # probably a local scope issue in testing!
             rows.append(_get_app_attr(app, is_comp))
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index 0d9a3fade..f5635915a 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -989,9 +989,10 @@ def _call(self, val, *args, **kwargs):
     if isinstance(val, NotCompleted) and self._skip_not_completed:
         return val
 
-    # todo we should get the source information from val here
     if self.app_type is not LOADER and self.input:  # passing to connected app
         val = self.input(val, *args, **kwargs)
+        if isinstance(val, NotCompleted) and self._skip_not_completed:
+            return val
 
     type_checked = self._validate_data_type(val)
     if not type_checked:
@@ -1089,17 +1090,17 @@ def define_app(
         what type of app, typically you just want GENERIC.
     skip_not_completed
         if True (default), NotCompleted instances are returned without being
-        passed to the app main() method.
+        passed to the app.main() method.
 
     Notes
     -----
 
     Instances of ``cogent3`` apps are callable. If an exception occurs,
     the app returns a ``NotCompleted`` instance with logging information.
-    Apps defined with app_type LOADER, GENERIC or WRITER can be "composed" (summed together)
-    to produce a single callable that sequentially invokes the composed
-    apps. For example, the independent usage of app instances
-    ``app1`` and ``app2`` as
+    Apps defined with app_type ``LOADER``, ``GENERIC`` or ``WRITER`` can be
+    "composed" (summed together) to produce a single callable that
+    sequentially invokes the composed apps. For example, the independent
+    usage of app instances ``app1`` and ``app2`` as
 
     >>> app2(app1(data))
 
@@ -1132,6 +1133,13 @@ def define_app(
     Overlap between the return type hint and first argument hint is required
     for two apps to be composed together.
 
+    ``define_app`` adds a ``__call__`` method which checks an input value prior
+    to passing it to ``app.main()`` as a positional argument. The data checking
+    results in ``NotCompleted`` being returned immediately, unless
+    ``skip_not_completed==False``. If the input value type is consistent with
+    the type hint on the first argument of main it is passed to ``app.main()``.
+    If it does not match, a new ``NotCompleted`` instance is returned.
+
     An example app definition.
 
     >>> from cogent3.app.composable import define_app
@@ -1176,7 +1184,7 @@ def define_app(
     ...                                 )
 
     ``omit_seqs`` is now an app, allowing creating different variants which
-    can be composed as per ones defined via a class
+    can be composed as per ones defined via a class.
 
     >>> omit_bad = omit_seqs(quantile=0.95)
 
@@ -1399,7 +1407,9 @@ def _apply_to(
     logger.log_message(f"{taken}", label="TIME TAKEN")
     log_file_path = Path(logger.log_file_path)
     logger.shutdown()
-    self.data_store.write_log(log_file_path.name, log_file_path.read_text())
+    self.data_store.write_log(
+        unique_id=log_file_path.name, data=log_file_path.read_text()
+    )
     if cleanup:
         log_file_path.unlink(missing_ok=True)
 
diff --git a/src/cogent3/app/data_store_new.py b/src/cogent3/app/data_store_new.py
index 14b616e5d..39dd04783 100644
--- a/src/cogent3/app/data_store_new.py
+++ b/src/cogent3/app/data_store_new.py
@@ -158,11 +158,11 @@ class DataStoreABC(ABC):
         self._check_writable(unique_id)
 
     @abstractmethod
-    def write_not_completed(self, unique_id: str, data: StrOrBytes) -> None:
+    def write_not_completed(self, *, unique_id: str, data: StrOrBytes) -> None:
         self._check_writable(unique_id)
 
     @abstractmethod
-    def write_log(self, unique_id: str, data: StrOrBytes) -> None:
+    def write_log(self, *, unique_id: str, data: StrOrBytes) -> None:
         self._check_writable(unique_id)
 
     @property
@@ -486,7 +486,7 @@ class DataStoreDirectory(DataStoreABC):
             self._completed.append(member)
         return member
 
-    def write_not_completed(self, unique_id: str, data: str) -> DataMember:
+    def write_not_completed(self, *, unique_id: str, data: str) -> DataMember:
         member = self._write(
             subdir=_NOT_COMPLETED_TABLE, unique_id=unique_id, suffix="json", data=data
         )
@@ -494,7 +494,7 @@ class DataStoreDirectory(DataStoreABC):
             self._not_completed.append(member)
         return member
 
-    def write_log(self, unique_id: str, data: str) -> None:
+    def write_log(self, *, unique_id: str, data: str) -> None:
         _ = self._write(subdir=_LOG_TABLE, unique_id=unique_id, suffix="log", data=data)
 
 
diff --git a/src/cogent3/app/io.py b/src/cogent3/app/io.py
index 02bfc0a14..7990dadfe 100644
--- a/src/cogent3/app/io.py
+++ b/src/cogent3/app/io.py
@@ -99,10 +99,10 @@ def get_data_store(
         stack_level=1,
     )
 
-    return open_data_store(base_path, suffix, limit, verbose)
+    return open_data_store(base_path=base_path, suffix=suffix, limit=limit)
 
 
-def _load_seqs(path, klass, parser, moltype):
+def _load_seqs(path, klass, parser, moltype):  # pragma: no cover
     abs_path = str(path)
     if not hasattr(path, "read"):
         # we use a DataStoreMember as it's read() handles zipped compression
diff --git a/src/cogent3/app/io_new.py b/src/cogent3/app/io_new.py
index e48113430..986db1200 100644
--- a/src/cogent3/app/io_new.py
+++ b/src/cogent3/app/io_new.py
@@ -1,6 +1,5 @@
 import contextlib
 import json
-import os
 import pickle
 import zipfile
 
@@ -32,11 +31,15 @@ from .data_store import (
     ReadOnlyZippedDataStore,
 )
 from .data_store_new import (
+    READONLY,
     DataStoreABC,
+    DataStoreDirectory,
+    Mode,
     get_data_source,
     load_record_from_json,
     make_record_for_json,
 )
+from .sqlite_data_store import DataStoreSqlite
 from .typing import (
     AlignedSeqsType,
     IdentifierType,
@@ -105,73 +108,61 @@ class register_datastore_reader:
         return func
 
 
-# todo GAH update to new data stores including sqlite
 # register the main readers
 register_datastore_reader("zip")(ReadOnlyZippedDataStore)
 register_datastore_reader("tinydb")(ReadOnlyTinyDbDataStore)
-register_datastore_reader(None)(ReadOnlyDirectoryDataStore)
-
-
-def findall(base_path, suffix="fa", limit=None, verbose=False):
-    """returns glob match to suffix, path is relative to base_path
-
-    Parameters
-    ----------
-    base_path : str
-        path to directory or zipped archive
-    suffix : str
-        suffix of filenames
-    limit : int or None
-        the number of matches to return
-    """
-    if not os.path.exists(base_path):
-        raise ValueError(f"'{base_path}' does not exist")
-
-    zipped = zipfile.is_zipfile(base_path)
-    klass = _datastore_reader_map.get(".zip" if zipped else None)
-    data_store = klass(base_path, suffix=suffix, limit=limit, verbose=verbose)
-    return data_store.members
+register_datastore_reader(None)(DataStoreDirectory)
+register_datastore_reader("sqlitedb")(DataStoreSqlite)
 
 
 def open_data_store(
-    base_path: Union[str, Path], suffix=None, limit=None, verbose=False
-):
-    """returns DataStore containing glob matches to suffix in base_path
+    base_path: Union[str, Path],
+    suffix: Optional[str] = None,
+    limit: Optional[int] = None,
+    mode: Union[str, Mode] = READONLY,
+    **kwargs,
+) -> DataStoreABC:
+    """returns DataStore instance of a type specified by the path suffix
 
     Parameters
     ----------
-    base_path : str or Path
-        path to directory or zipped archive
-    suffix : str
+    base_path
+        path to directory or db
+    suffix
         suffix of filenames
-    limit : int or None
+    limit
         the number of matches to return
-    Returns
-    -------
-    ReadOnlyDirectoryDataStore or ReadOnlyZippedDataStore
+    mode
+        opening mode, either r, w, a as per file opening modes
     """
+    mode = Mode(mode)
+    if not isinstance(suffix, (str, type(None))):
+        raise ValueError(f"suffix {type(suffix)} not one of string or None")
+
+    kwargs = dict(limit=limit, mode=mode, suffix=suffix)
     base_path = Path(base_path)
     base_path = base_path.expanduser().absolute()
-    if base_path.suffix in (".tinydb", ".sqlitedb"):
-        suffix = "json"
-
-    if suffix is None:
-        raise ValueError("suffix required")
-
-    if not base_path.exists():
-        raise ValueError(f"'{base_path}' does not exist")
-
-    if type(suffix) != str:
-        raise ValueError(f"{suffix} is not a string")
+    if base_path.suffix == ".tinydb":
+        kwargs["suffix"] = "json"
+        kwargs.pop("mode")
 
-    if zipfile.is_zipfile(base_path):
+    if base_path.suffix == ".sqlitedb":
+        ds_suffix = base_path.suffix
+        kwargs.pop("suffix")
+    elif zipfile.is_zipfile(base_path):
         ds_suffix = ".zip"
+        kwargs.pop("mode")
     elif base_path.suffix:
         ds_suffix = base_path.suffix
     else:
         ds_suffix = None
+
+    if ds_suffix is None and suffix is None:
+        raise ValueError("a suffix is required if using a directory data store")
+
     klass = _datastore_reader_map[ds_suffix]
-    return klass(base_path, suffix=suffix, limit=limit)
+
+    return klass(base_path, **kwargs)
 
 
 @define_app(skip_not_completed=False)
@@ -254,6 +245,7 @@ class tabular(Enum):
 
 
 def _read_it(path):
+    path = Path(path) if isinstance(path, str) else path
     try:
         data = path.read()
     except AttributeError:


Test patch:
diff --git a/tests/test_app/test_app_mpi.py b/tests/test_app/test_app_mpi.py
index d5480df6a..7930dbf97 100644
--- a/tests/test_app/test_app_mpi.py
+++ b/tests/test_app/test_app_mpi.py
@@ -1,11 +1,8 @@
 from pathlib import Path
-from tempfile import TemporaryDirectory
-from unittest import TestCase, main, skipUnless
 
-from cogent3.app import align as align_app
-from cogent3.app import io_new as io_app
-from cogent3.app import open_data_store
-from cogent3.app.io import write_db
+import pytest
+
+from cogent3 import get_app, open_data_store
 from cogent3.util import parallel
 
 
@@ -18,38 +15,36 @@ __maintainer__ = "Gavin Huttley"
 __email__ = "Gavin.Huttley@anu.edu.au"
 __status__ = "Alpha"
 
+DATA_DIR = Path(__file__).parent.parent / "data"
 
-class MPITests(TestCase):
-    basedir = "data"
 
-    @skipUnless(parallel.USING_MPI, reason="Not using MPI")
-    def test_write_db(self):
-        """writing with overwrite in MPI should reset db"""
-        dstore = open_data_store("data", suffix="fasta")
-        members = dstore.filtered(callback=lambda x: "brca1.fasta" not in x.split("/"))
-        with TemporaryDirectory(dir=".") as dirname:
-            path = Path(dirname) / "delme.tinydb"
-            reader = io_app.load_unaligned()
-            aligner = align_app.align_to_ref()
-            writer = write_db(path, create=True, if_exists="overwrite")
-            process = reader + aligner + writer
+@pytest.fixture(scope="function")
+def tmp_dir(tmpdir_factory):
+    return Path(tmpdir_factory.mktemp("mpirun"))
 
-            r = process.apply_to(
-                members,
-                show_progress=False,
-                parallel=True,
-                par_kw=dict(use_mpi=True),
-            )
 
-            expect = [str(m) for m in process.data_store]
-            process.data_store.close()
+@pytest.mark.skipif(not parallel.USING_MPI, reason="Not using MPI")
+def test_write_db(tmp_dir):
+    """writing with overwrite in MPI should reset db"""
+    dstore = open_data_store("data", suffix="fasta")
+    members = [m for m in dstore if m.unique_id != "brca1.fasta"]
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    reader = get_app("load_unaligned")
+    aligner = get_app("align_to_ref")
+    writer = get_app("write_db", out_dstore)
+    process = reader + aligner + writer
 
-            # now get read only and check what's in there
-            result = open_data_store(path)
-            got = [str(m) for m in result]
+    r = process.apply_to(
+        members,
+        show_progress=False,
+        parallel=True,
+        par_kw=dict(use_mpi=True),
+    )
 
-            assert got == expect
+    expect = [str(m) for m in process.data_store]
 
+    # now get read only and check what's in there
+    result = open_data_store(out_dstore.source)
+    got = [str(m) for m in result]
 
-if __name__ == "__main__":
-    main()
+    assert got == expect
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index f29ca1672..a45666f0e 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -13,7 +13,7 @@ import pytest
 from numpy import array, ndarray
 from scitrack import CachingLogger
 
-from cogent3 import make_aligned_seqs, open_data_store
+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
@@ -305,11 +305,11 @@ def test_disconnect():
 def test_as_completed():
     """correctly applies iteratively"""
     dstore = open_data_store(DATA_DIR, suffix="fasta", limit=3)
-    reader = io_app.load_unaligned(format="fasta", moltype="dna")
+    reader = get_app("load_unaligned", format="fasta", moltype="dna")
     got = list(reader.as_completed(dstore))
     assert len(got) == len(dstore)
     # should also be able to apply the results to another composable func
-    min_length = sample_app.min_length(10)
+    min_length = get_app("sample.min_length", 10)
     got = list(min_length.as_completed(got))
     assert len(got) == len(dstore)
     # should work on a chained function
@@ -320,7 +320,8 @@ def test_as_completed():
     got = list(proc.as_completed([str(m) for m in dstore]))
     assert len(got) == len(dstore)
     # or a single string
-    got = list(proc.as_completed(str(dstore[0])))
+    path = str(Path(dstore[0].data_store.source) / dstore[0].unique_id)
+    got = list(proc.as_completed(path))
     assert len(got) == 1
     assert isinstance(got[0].obj, SequenceCollection)
     # raises ValueError if empty list
@@ -1118,7 +1119,7 @@ def test_apply_to_only_appends(half_dstore1, half_dstore2):
     process1 = reader1 + min_length1 + writer1
     # create paths as strings
     dstore1 = open_data_store(half_dstore1.source, suffix="fasta")
-    dstore1 = [str(m) for m in dstore1]
+    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):
         _ = process1.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 3a379cbce..b09a8368e 100644
--- a/tests/test_app/test_data_store_new.py
+++ b/tests/test_app/test_data_store_new.py
@@ -91,7 +91,7 @@ def nc_dstore(nc_dir):
     dstore = DataStoreDirectory(nc_dir, suffix="fasta", mode=OVERWRITE)
     # write one log file
     log_filename = "scitrack.log"
-    dstore.write_log(log_filename, (DATA_DIR / log_filename).read_text())
+    dstore.write_log(unique_id=log_filename, data=(DATA_DIR / log_filename).read_text())
     # write three not_completed file
     nc = [
         NotCompleted(
@@ -100,7 +100,7 @@ def nc_dstore(nc_dir):
         for i in range(3)
     ]
     for i, item in enumerate(nc):
-        dstore.write_not_completed(f"nc{i + 1}", item.to_json())
+        dstore.write_not_completed(unique_id=f"nc{i + 1}", data=item.to_json())
     assert len(dstore.not_completed) == 3
     assert len(list((nc_dir / _MD5_TABLE).glob("*.txt"))) == len(dstore)
     filenames = DATA_DIR.glob("*.fasta")
diff --git a/tests/test_app/test_io.py b/tests/test_app/test_io.py
index 34f3aa7c9..112f502f6 100644
--- a/tests/test_app/test_io.py
+++ b/tests/test_app/test_io.py
@@ -38,28 +38,6 @@ class TestIo(TestCase):
         if path.exists():
             path.unlink()
 
-    def test_define_data_store(self):
-        """returns an iterable data store"""
-        found = open_data_store(self.basedir, suffix=".fasta")
-        self.assertTrue(len(found) > 1)
-        found = open_data_store(self.basedir, suffix=".fasta", limit=2)
-        self.assertTrue(len(found) == 2)
-
-        # and with a suffix
-        found = list(open_data_store(self.basedir, suffix=".fasta*"))
-        self.assertTrue(len(found) > 2)
-
-        # with a wild-card suffix
-        found = list(open_data_store(self.basedir, suffix="*"))
-        self.assertEqual(len(os.listdir(self.basedir)), len(found))
-
-        # raises ValueError if suffix not provided or invalid
-        with self.assertRaises(ValueError):
-            _ = open_data_store(self.basedir)
-
-        with self.assertRaises(ValueError):
-            _ = open_data_store(self.basedir, 1)
-
     def test_load_db_failure_json_file(self):
         """informative load_db error message when given a json file path"""
         # todo this test has a trapped exception about being unable to delete
@@ -108,32 +86,6 @@ class TestIo(TestCase):
             w = io_app.write_db(outdir, create=True, if_exists="skip")
             w.data_store.close()
 
-    def test_write_db_parallel(self):
-        """writing with overwrite in parallel should reset db"""
-        with TemporaryDirectory(dir=".") as dirname:
-            outdir = join(dirname, "delme.tinydb")
-            dstore = open_data_store(self.basedir, suffix="fasta")
-            members = dstore.filtered(
-                callback=lambda x: "brca1.fasta" not in x.split("/")
-            )
-            reader = io_app.load_unaligned()
-            aligner = align_app.align_to_ref()
-            writer = write_db(outdir, create=True, if_exists="overwrite")
-            process = reader + aligner + writer
-
-            _ = process.apply_to(
-                members, show_progress=False, parallel=True, cleanup=True
-            )
-            expect = [str(m) for m in process.data_store]
-            process.data_store.close()
-
-            # now get read only and check what's in there
-            result = open_data_store(outdir)
-            got = [str(m) for m in result]
-            self.assertNotEqual(got, [])
-            result.close()
-            self.assertEqual(got, expect)
-
 
 if __name__ == "__main__":
     main()
diff --git a/tests/test_app/test_io_new.py b/tests/test_app/test_io_new.py
index 82fc60828..f7f8c85d9 100644
--- a/tests/test_app/test_io_new.py
+++ b/tests/test_app/test_io_new.py
@@ -1,6 +1,7 @@
 import bz2
 import gzip
 import json
+import os
 import pickle
 
 from pathlib import Path
@@ -10,7 +11,7 @@ import pytest
 
 from numpy.testing import assert_allclose
 
-from cogent3 import DNA
+from cogent3 import DNA, get_app, open_data_store
 from cogent3.app import io_new as io_app
 from cogent3.app.composable import NotCompleted, source_proxy
 from cogent3.app.data_store_new import DataMember, DataStoreDirectory, Mode
@@ -435,6 +436,7 @@ def test_pickled_compress_roundtrip(data):
 
 def test_write_db_load_db(fasta_dir, tmp_dir):
     from cogent3.app.sqlite_data_store import DataStoreSqlite
+    from cogent3.util.misc import get_object_provenance
 
     orig_dstore = DataStoreDirectory(fasta_dir, suffix="fasta")
     data_store = DataStoreSqlite(tmp_dir / "test.sqlitedb", mode="w")
@@ -447,6 +449,8 @@ def test_write_db_load_db(fasta_dir, tmp_dir):
         read = reader(m)
         assert orig == read
 
+    assert data_store.record_type == get_object_provenance(orig)
+
 
 def test_write_db_not_completed(tmp_dir):
     from cogent3.app.sqlite_data_store import DataStoreSqlite
@@ -458,3 +462,48 @@ def test_write_db_not_completed(tmp_dir):
     writer.main(nc, identifier="blah")
     assert len(data_store.not_completed) == 1
     reader = io_app.load_db()
+    got = reader(data_store.not_completed[0])
+    assert got.to_rich_dict() == nc.to_rich_dict()
+
+
+def test_write_db_parallel(tmp_dir, fasta_dir):
+    """writing with overwrite in parallel should reset db"""
+    dstore = open_data_store(fasta_dir, suffix="fasta")
+    out_dstore = open_data_store(tmp_dir / "delme.sqlitedb", mode="w")
+    members = [m for m in dstore if m.unique_id != "brca1.fasta"]
+    reader = get_app("load_unaligned")
+    aligner = get_app("align_to_ref")
+    writer = get_app("write_seqs", out_dstore)
+    process = reader + aligner + writer
+
+    _ = process.apply_to(members, show_progress=False, parallel=True, cleanup=True)
+    expect = [str(Path(m.data_store.source) / m.unique_id) for m in out_dstore]
+
+    # now get read only and check what's in there
+    result = open_data_store(out_dstore.source, suffix="fasta")
+    got = [str(Path(m.data_store.source) / m.unique_id) for m in result]
+    assert got != []
+    assert got == expect
+
+
+def test_define_data_store(fasta_dir):
+    """returns an iterable data store"""
+    found = open_data_store(fasta_dir, suffix=".fasta")
+    assert len(found) > 1
+    found = open_data_store(fasta_dir, suffix=".fasta", limit=2)
+    assert len(found) == 2
+
+    # and with a suffix
+    found = list(open_data_store(fasta_dir, suffix=".fasta*"))
+    assert len(found) > 2
+
+    # with a wild-card suffix
+    found = list(open_data_store(fasta_dir, suffix="*"))
+    assert len(os.listdir(fasta_dir)) == len(found)
+
+    # raises ValueError if suffix not provided or invalid
+    with pytest.raises(ValueError):
+        _ = open_data_store(fasta_dir)
+
+    with pytest.raises(ValueError):
+        _ = open_data_store(fasta_dir, 1)