Repository: cogent3/cogent3
Pull request #1174 (2022-10-18 05:43:05)

Data store refactor


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index 11151a8ae..3c6527d4f 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -1195,32 +1195,20 @@ def define_app(klass=None, *, app_type: AppType = GENERIC):
     def wrapped(klass):
         if inspect.isfunction(klass):
             klass = _class_from_func(klass)
-
         if not inspect.isclass(klass):
             raise ValueError(f"{klass} is not a class")
 
-        # check if user defined these methods
-        # todo derive from __mapping
-        method_list = [
-            "__call__",
-            "__repr__",
-            "__str__",
-            "__new__",
-            "_validate_data_type",
-            "as_completed",
-            "_source_wrapped",
-        ]
-        excludes = ["__add__", "disconnect"]
-        if app_type is WRITER:
-            # only writers get the apply_to method
+        excludes = []
+        if not composable:
+            excludes = ["__add__", "disconnect"]
+        if app_type is not WRITER:
             excludes.extend(["apply_to", "set_logger"])
-
+        method_list = [item for item in __mapping if item not in excludes] + ["__str__"]
+        # check if user defined input for composable
         if composable and getattr(klass, "input", None):
             raise TypeError(
                 f"remove 'input' attribute in {klass.__name__!r}, this functionality provided by define_app"
             )
-        elif composable:
-            method_list.extend(excludes)
 
         for meth in method_list:
             # make sure method not defined by user before adding
@@ -1299,23 +1287,22 @@ def _as_completed(
     aggregates results. If run in serial, results are returned in the
     same order as provided.
     """
-    app = self.input if self.app_type is WRITER else self
+    app = (
+        self.input._source_wrapped if self.app_type is WRITER else self._source_wrapped
+    )
 
     if isinstance(dstore, str):
         dstore = [dstore]
-
     mapped = _proxy_input(dstore)
-
     if not mapped:
         raise ValueError("dstore is empty")
 
     if parallel:
         par_kw = par_kw or {}
-        to_do = PAR.as_completed(app._source_wrapped, mapped, **par_kw)
+        to_do = PAR.as_completed(app, mapped, **par_kw)
     else:
-        to_do = map(app._source_wrapped, mapped)
-
-    yield from to_do
+        to_do = map(app, mapped)
+    return to_do
 
 
 def is_composable(obj):
@@ -1386,17 +1373,15 @@ def _apply_to(
     logger.log_versions(["cogent3"])
 
     inputs = _proxy_input(dstore)
-
     for result in self.as_completed(inputs, parallel=parallel, par_kw=par_kw):
         member = self.main(
             data=result.obj, identifier=get_data_source(result.source)
         )  # writers must return DataMember
-        md5 = self.data_store.md5(member.unique_id)
+        md5 = member.md5
         logger.log_message(str(member), label="output")
         logger.log_message(md5, label="output md5sum")
 
     taken = time.time() - start
-
     logger.log_message(f"{taken}", label="TIME TAKEN")
     log_file_path = pathlib.Path(logger.log_file_path)
     logger.shutdown()
diff --git a/src/cogent3/app/data_store.py b/src/cogent3/app/data_store.py
index d254d4181..184a55dc9 100644
--- a/src/cogent3/app/data_store.py
+++ b/src/cogent3/app/data_store.py
@@ -325,6 +325,9 @@ class ReadOnlyDataStoreBase:
         self._md5 = md5_setting
         return result
 
+    def write_log(self, unique_id: str, data: str) -> None:
+        _ = self.write(unique_id, data)
+
 
 class ReadOnlyDirectoryDataStore(ReadOnlyDataStoreBase):
     @property
diff --git a/src/cogent3/app/data_store_new.py b/src/cogent3/app/data_store_new.py
index 90d18dc11..17dcfafcd 100644
--- a/src/cogent3/app/data_store_new.py
+++ b/src/cogent3/app/data_store_new.py
@@ -173,6 +173,10 @@ class DataMember(DataMemberABC):
     def source(self):
         return self.unique_id
 
+    @property
+    def md5(self):
+        return self.data_store.md5(self.unique_id)
+
 
 class DataStoreDirectory(DataStoreABC):
     def __init__(
@@ -243,8 +247,6 @@ class DataStoreDirectory(DataStoreABC):
         unique_id = Path(unique_id)
         unique_id = re.sub(rf"[.]({self.suffix}|json)$", ".txt", unique_id.name)
         path = self.source / _MD5_TABLE / unique_id
-        if not path.exists():
-            return None
 
         return path.read_text() if path.exists() else None
 


Test patch:
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index 43d708cba..93f6bfe65 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -53,7 +53,7 @@ __email__ = "Gavin.Huttley@anu.edu.au"
 __status__ = "Alpha"
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
 def tmp_dir(tmpdir_factory):
     return tmpdir_factory.mktemp("datastore")
 
@@ -250,36 +250,32 @@ def test_apply_to_non_unique_identifiers():
             process.apply_to(dstore)
 
 
-@pytest.mark.xfail
-def test_apply_to_logging():
+def test_apply_to_logging(tmp_dir):
     """correctly creates log file"""
     dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
-    with TemporaryDirectory(dir=".") as dirname:
-        reader = io_app.load_aligned(format="fasta", moltype="dna")
-        min_length = sample_app.min_length(10)
-        outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
-        writer = io_app.write_db(outpath)
-        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()
+    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
+    r = process.apply_to(dstore, show_progress=False)
+    # always creates a log
+    assert len(process.data_store.logs) == 1
+    process.data_store.close()
 
 
-@pytest.mark.xfail
-def test_apply_to_logger():
+def test_apply_to_logger(tmp_dir):
     """correctly uses user provided logger"""
     dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
-    with TemporaryDirectory(dir=".") as dirname:
-        LOGGER = CachingLogger()
-        reader = io_app.load_aligned(format="fasta", moltype="dna")
-        min_length = sample_app.min_length(10)
-        outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
-        writer = io_app.write_db(outpath)
-        process = reader + min_length + writer
-        r = process.apply_to(dstore, show_progress=False, logger=LOGGER)
-        assert len(process.data_store.logs) == 1
-        process.data_store.close()
+    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)
+    process = reader + min_length + writer
+    r = process.apply_to(dstore, show_progress=False, logger=LOGGER)
+    assert len(process.data_store.logs) == 1
+    process.data_store.close()
 
 
 @pytest.mark.xfail