Repository: cogent3/cogent3
Pull request #1162 (2022-10-05 20:10:47)

DEV: Add new functions in DataStoreDirectory

lgtm-com[bot]: This pull request **introduces 2 alerts** when merging 5652e359463417ba6452c822b86d09e9a93124df into 4287cd0363cd8aed7c59cee89182e4699051a9d1 - [view on LGTM.com](https://lgtm.com/projects/g/cogent3/cogent3/rev/pr-8866b94f3a0a4ce8f0f4a0fd72881986d643f4bd)

**new alerts:**

* 1 for Signature mismatch in overriding method
* 1 for First parameter of a method is not named &#39;self&#39;

Patch:
diff --git a/src/cogent3/app/data_store_new.py b/src/cogent3/app/data_store_new.py
index 92557fbcb..9bae42cad 100644
--- a/src/cogent3/app/data_store_new.py
+++ b/src/cogent3/app/data_store_new.py
@@ -1,10 +1,11 @@
 from __future__ import annotations
 
-import glob
 import re
+import reprlib
 import shutil
 
 from abc import ABC, abstractmethod
+from collections import defaultdict
 from enum import Enum
 from functools import singledispatch
 from pathlib import Path
@@ -12,13 +13,15 @@ from pathlib import Path
 from scitrack import get_text_hexdigest
 
 from cogent3.app.typing import TabularType
-from cogent3.util.io import open_
+from cogent3.util.deserialise import deserialise_not_completed
+from cogent3.util.io import get_format_suffixes, open_
 from cogent3.util.parallel import is_master_process
+from cogent3.util.table import Table
 
 
 __author__ = "Gavin Huttley"
 __copyright__ = "Copyright 2007-2022, The Cogent Project"
-__credits__ = ["Gavin Huttley"]
+__credits__ = ["Gavin Huttley", "Nick Shahmaras"]
 __license__ = "BSD-3"
 __version__ = "2022.8.24a1"
 __maintainer__ = "Gavin Huttley"
@@ -91,8 +94,10 @@ class DataStoreABC(ABC):
     def write(self, unique_id: str, data: str | bytes) -> None:
         if self._if_dest_exists is READONLY:
             raise IOError("datastore is readonly")
-        if self._if_member_exists is RAISE and unique_id in self:
-            raise IOError("member exists")
+        if self._if_member_exists in (READONLY, RAISE) and unique_id in self:
+            raise IOError(
+                f"DataStore member_exists is {self._if_member_exists!r} and {unique_id!r} exists"
+            )
 
     @abstractmethod
     def write_not_completed(self, unique_id: str, data: str | bytes) -> None:
@@ -141,6 +146,10 @@ class DataStoreABC(ABC):
     def summary_not_completed(self) -> TabularType:
         ...
 
+    @abstractmethod
+    def drop_not_completed(self):
+        ...
+
 
 class DataMember(DataMemberABC):
     def __init__(self, data_store: DataStoreABC = None, unique_id: str = None):
@@ -183,9 +192,7 @@ class DataStoreDirectory(DataStoreABC):
     def _source_check_create(self, if_dest_exists):
         if not is_master_process():
             return
-
         sub_dirs = [_NOT_COMPLETED_TABLE, _LOG_TABLE, _MD5_TABLE]
-
         if if_dest_exists is READONLY and not self.source.exists():
             raise IOError("must exist")
         elif if_dest_exists is RAISE and self.source.exists():
@@ -206,7 +213,7 @@ class DataStoreDirectory(DataStoreABC):
             path.mkdir(parents=True, exist_ok=True)
 
         for sub_dir in sub_dirs:
-            (self.source.parent / sub_dir).mkdir(parents=True, exist_ok=True)
+            (self.source / sub_dir).mkdir(parents=True, exist_ok=True)
 
     def read(self, unique_id: str) -> str:
         """reads data corresponding to identifier"""
@@ -239,14 +246,22 @@ class DataStoreDirectory(DataStoreABC):
         self._md5 = md5_setting
         return self._checksums.get(unique_id, None)
 
+    def drop_not_completed(self):
+        nc_dir = self.source / _NOT_COMPLETED_TABLE
+        md5_dir = self.source / _MD5_TABLE
+        for file in list(nc_dir.glob("*.json")):
+            file.unlink()
+            md5_file = md5_dir / f"{file.stem}.txt"
+            md5_file.unlink()
+
     @property
     def members(self) -> list[DataMember]:
-        if not self._members:  # members in completed and incomplete
-            pattern = f"{self.source}/**/*.{self.suffix}"
-            paths = list(glob.iglob(pattern, recursive=True))
+        if not self._members:  # members in completed and not_completed
             self._members = []
-            for i, path in enumerate(paths):
-                if self._limit and i >= self._limit:
+            for path in list(self.source.glob(f"*.{self.suffix}")) + list(
+                (self.source / _NOT_COMPLETED_TABLE).glob("*.json")
+            ):
+                if self._limit and len(self._members) >= self._limit:
                     break
                 self._members.append(DataMember(self, Path(path).name))
 
@@ -272,53 +287,139 @@ class DataStoreDirectory(DataStoreABC):
 
     @property
     def not_completed(self) -> list[DataMember]:
-        ic_dir = self.source / _NOT_COMPLETED_TABLE
+        nc_dir = self.source / _NOT_COMPLETED_TABLE
         return (
             [
                 DataMember(self, Path(_NOT_COMPLETED_TABLE) / m.name)
-                for m in ic_dir.glob("*.json")
+                for m in nc_dir.glob("*.json")
             ]
-            if ic_dir.exists()
+            if nc_dir.exists()
             else []
         )
 
-    def write(self, unique_id: str, data: str) -> None:
+    def _write(
+        self, subdir: str, unique_id: str, suffix: str, data: str, md5: bool
+    ) -> None:
         super().write(unique_id, data)
-        with open_(self.source / unique_id, mode="w") as out:
+        assert suffix, "Must provide suffix"
+        # check suffix compatible with this datastore
+        sfx, cmp = get_format_suffixes(unique_id)
+        if sfx != suffix:
+            unique_id = f"{unique_id}.{suffix}"
+        unique_id = (
+            unique_id.replace(self.suffix, suffix)
+            if self.suffix and self.suffix != suffix
+            else unique_id
+        )
+        with open_(self.source / subdir / unique_id, mode="w") as out:
             out.write(data)
-            if self._md5 and isinstance(data, str):
-                self._checksums[unique_id] = get_text_hexdigest(data)
+        if suffix != "log" and not self._limit or len(self) < self._limit:
+            self._members.append(DataMember(self, unique_id))
+        if md5:
+            md5 = get_text_hexdigest(data)
+            unique_id = unique_id.replace(suffix, "txt")
+            unique_id = unique_id if cmp is None else unique_id.replace(f".{cmp}", "")
+            with open_(self.source / _MD5_TABLE / unique_id, mode="w") as out:
+                out.write(md5)
+
+    def write(self, unique_id: str, data: str) -> None:
+        self._write("", unique_id, self.suffix, data, True)
 
     def write_not_completed(self, unique_id: str, data: str) -> None:
-        super().write(unique_id, data)
-        with open_(self.source / _NOT_COMPLETED_TABLE / unique_id, mode="w") as out:
-            out.write(data)
+        self._write(_NOT_COMPLETED_TABLE, unique_id, "json", data, True)
 
     def write_log(self, unique_id: str, data: str) -> None:
-        super().write(unique_id, data)
-        with open_(self.source / _LOG_TABLE / unique_id, mode="w") as out:
-            out.write(data)
+        self._write(_LOG_TABLE, unique_id, "log", data, False)
 
     def describe(self) -> TabularType:
-        ...
+        num_notcompleted = len(self.not_completed)
+        num_completed = len(self.completed)
+        num_logs = len(self.logs)
+        return Table(
+            header=["record type", "number"],
+            data=[
+                ["completed", num_completed],
+                ["not_completed", num_notcompleted],
+                ["logs", num_logs],
+            ],
+        )
 
     @property
     def summary_logs(self) -> TabularType:
-        ...
+        """returns a table summarising log files"""
+        rows = []
+        for record in self.logs:
+            data = record.read().splitlines()
+            first = data.pop(0).split("\t")
+            row = [first[0], record.name]
+            key = None
+            mapped = {}
+            for line in data:
+                line = line.split("\t")[-1].split(" : ", maxsplit=1)
+                if len(line) == 1:
+                    mapped[key] += line[0]
+                    continue
+
+                key = line[0]
+                mapped[key] = line[1]
+
+            data = mapped
+            row.extend(
+                [
+                    data["python"],
+                    data["user"],
+                    data["command_string"],
+                    data["composable function"],
+                ]
+            )
+            rows.append(row)
+        return Table(
+            header=["time", "name", "python version", "who", "command", "composable"],
+            data=rows,
+            title="summary of log files",
+        )
 
     @property
     def summary_not_completed(self) -> TabularType:
-        ...
+        """returns a table summarising not completed results"""
+        # detect last exception line
+        err_pat = re.compile(r"[A-Z][a-z]+[A-Z][a-z]+\:.+")
+        types = defaultdict(list)
+        indices = "type", "origin"
+        for member in self.not_completed:
+            record = member.read()
+            record = deserialise_not_completed(record)
+            key = tuple(getattr(record, k, None) for k in indices)
+            match = err_pat.findall(record.message)
+            types[key].append([match[-1] if match else record.message, record.source])
+
+        header = list(indices) + ["message", "num", "source"]
+        rows = []
+        maxtring = reprlib.aRepr.maxstring
+        reprlib.aRepr.maxstring = 45
+
+        for record in types:
+            messages, sources = list(zip(*types[record]))
+            messages = reprlib.repr(
+                ", ".join(m.splitlines()[-1] for m in set(messages))
+            )
+            sources = reprlib.repr(", ".join(s.splitlines()[-1] for s in sources))
+            row = list(record) + [
+                messages,
+                len(types[record]),
+                sources,
+            ]
+            rows.append(row)
+
+        reprlib.aRepr.maxstring = maxtring  # restoring original val
+        return Table(header=header, data=rows, title="not completed records")
 
     def __repr__(self):
-        notcompleted_path = Path(self.source / _NOT_COMPLETED_TABLE)
-        if not notcompleted_path.exists():
-            return "This isn't a results directory"
-        num_completed += len(self)
-        num_not_completed = len(list(notcompleted_path.glob("*")))
+        num_completed = len(self)
+        num_not_completed = len(self.not_completed)
         name = self.__class__.__name__
-        sample = f"{str(list(self[:3]))[:-1]}..." if num_completed > 3 else list(self)
-        return f"{num_completed+num_not_completed}x member {name}(source='{self.source}', members={sample})"
+        sample = f"{list(self[:2])}..." if num_completed > 2 else list(self)
+        return f"{num_completed + num_not_completed}x member {name}(source='{self.source}', members={sample})"
 
 
 @singledispatch


Test patch:
diff --git a/.github/workflows/testing_develop.yml b/.github/workflows/testing_develop.yml
index e8d584686..fb0a5b8b3 100644
--- a/.github/workflows/testing_develop.yml
+++ b/.github/workflows/testing_develop.yml
@@ -2,9 +2,9 @@ name: CI
 
 on:
   push:
-    branches: [ "develop", "composable_refactor" ]
+    branches: [ "develop", "data_store_refactor" ]
   pull_request:
-    branches: [ "develop", "composable_refactor" ]
+    branches: [ "develop", "data_store_refactor" ]
 
 jobs:
   tests:
diff --git a/tests/test_app/test_data_store_new.py b/tests/test_app/test_data_store_new.py
index 1f8df9390..e26fd0c2c 100644
--- a/tests/test_app/test_data_store_new.py
+++ b/tests/test_app/test_data_store_new.py
@@ -6,12 +6,24 @@ import pytest
 from cogent3.app.composable import NotCompleted
 from cogent3.app.data_store_new import (
     _LOG_TABLE,
+    _MD5_TABLE,
     _NOT_COMPLETED_TABLE,
     OVERWRITE,
     READONLY,
+    SKIP,
     DataStoreDirectory,
 )
 
+__author__ = "Gavin Huttley"
+__copyright__ = "Copyright 2007-2022, The Cogent Project"
+__credits__ = ["Gavin Huttley", "Nick Shahmaras"]
+__license__ = "BSD-3"
+__version__ = "2022.8.24a1"
+__maintainer__ = "Gavin Huttley"
+__email__ = "Gavin.Huttley@anu.edu.au"
+__status__ = "Alpha"
+
+
 DATA_DIR = Path(__file__).parent.parent / "data"
 
 @pytest.fixture(scope="session")
@@ -46,38 +58,64 @@ def write_dir(tmp_dir):
     return write_dir
 
 
-@pytest.fixture(scope="session")
-def ic_dir(tmp_dir):
+@pytest.fixture(scope="function")
+def nc_dir(tmp_dir):
     tmp_dir = Path(tmp_dir)
-    filenames = DATA_DIR.glob("*.fasta")
-    ic_dir = tmp_dir / "ic_dir"
-    ic_dir.mkdir(parents=True, exist_ok=True)
-    for fn in filenames:
-        dest = ic_dir / fn.name
-        dest.write_text(fn.read_text())
-    logs_dir = ic_dir / _LOG_TABLE
-    nc_dir = ic_dir / _NOT_COMPLETED_TABLE
+    nc_dir = tmp_dir / "nc_dir"
+    logs_dir = nc_dir / _LOG_TABLE
+    not_dir = nc_dir / _NOT_COMPLETED_TABLE
+    md5_dir = nc_dir / _MD5_TABLE
+    for fn in list(logs_dir.glob("*.log")):
+        fn.unlink()
+    for fn in list(not_dir.glob("*.json")):
+        fn.unlink()
+    for fn in list(md5_dir.glob("*.txt")):
+        fn.unlink()
+    for fn in list(nc_dir.glob("*.*")):
+        fn.unlink()
+    if Path(logs_dir).exists():
+        Path(logs_dir).rmdir()
+    if Path(not_dir).exists():
+        Path(not_dir).rmdir()
+    if Path(md5_dir).exists():
+        Path(md5_dir).rmdir()
+    if Path(nc_dir).exists():
+        Path(nc_dir).rmdir()
+    nc_dir.mkdir(parents=True, exist_ok=True)
     logs_dir.mkdir(exist_ok=True)
-    nc_dir.mkdir(exist_ok=True)
+    not_dir.mkdir(exist_ok=True)
     (logs_dir / "scitrack.log").write_text((DATA_DIR / "scitrack.log").read_text())
-    nc = NotCompleted("FAIL", "dummy", "dummy_message", source="dummy_source")
-    (nc_dir / "nc.json").write_text(nc.to_json())
-    return ic_dir
+    return nc_dir
 
 
 @pytest.fixture(scope="session")
 def ro_dstore(fasta_dir):
-    return DataStoreDirectory(fasta_dir, suffix=".fasta", if_dest_exists=READONLY)
+    return DataStoreDirectory(fasta_dir, suffix="fasta", if_dest_exists=READONLY)
 
 
 @pytest.fixture(scope="session")
 def w_dstore(write_dir):
-    return DataStoreDirectory(write_dir, if_dest_exists=OVERWRITE)
-
-
-@pytest.fixture(scope="session")
-def ic_dstore(ic_dir):
-    return DataStoreDirectory(ic_dir, suffix=".fasta")
+    return DataStoreDirectory(write_dir, suffix="fasta", if_dest_exists=OVERWRITE)
+
+
+@pytest.fixture(scope="function")
+def nc_dstore(nc_dir):
+    dstore = DataStoreDirectory(nc_dir, suffix="fasta", if_dest_exists=SKIP)
+    nc = [
+        NotCompleted(
+            "FAIL", f"dummy{i}", f"dummy_message{i}", source=f"dummy_source{i}"
+        )
+        for i in range(3)
+    ]
+    for i, item in enumerate(nc):
+        dstore.write_not_completed(f"nc{i + 1}", 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")
+    for fn in filenames:
+        identifier = fn.name
+        dstore.write(identifier, fn.read_text())
+    return dstore
 
 
 def test_contains(ro_dstore):
@@ -91,9 +129,9 @@ def test_iter(ro_dstore):
     assert members == ro_dstore.members
 
 
-def test_len(fasta_dir, ro_dstore):
+def test_len(ro_dstore):
     """DataStore returns correct len"""
-    expect = len(list(fasta_dir.glob("*.fasta")))
+    expect = len(list(ro_dstore.source.glob("*.fasta")))
     assert expect == len(ro_dstore) == len(ro_dstore.members)
 
 
@@ -106,15 +144,15 @@ def test_getitem(ro_dstore):
     assert last.unique_id != first.unique_id
 
 
-def test_iterall(fasta_dir, ro_dstore):
-    expect = {fn.name for fn in fasta_dir.glob("*.fasta")}
+def test_iterall(ro_dstore):
+    expect = {fn.name for fn in ro_dstore.source.glob("*.fasta")}
     got = {m.unique_id for m in ro_dstore}
     assert expect == got
 
 
-def test_read(fasta_dir, ro_dstore):
+def test_read(ro_dstore):
     """correctly read content"""
-    expect = (fasta_dir / "brca1.fasta").read_text()
+    expect = (ro_dstore.source / "brca1.fasta").read_text()
     got = ro_dstore.read("brca1.fasta")
     assert got == expect
 
@@ -146,15 +184,15 @@ def test_no_not_completed(ro_dstore):
     assert len(ro_dstore.not_completed) == 0
 
 
-def test_logs(ic_dstore):
-    assert len(ic_dstore.logs) == 1
-    log = ic_dstore.logs[0].read()
+def test_logs(nc_dstore):
+    assert len(nc_dstore.logs) == 1
+    log = nc_dstore.logs[0].read()
     assert isinstance(log, str)
 
 
-def test_not_completed(ic_dstore):
-    assert len(ic_dstore.not_completed) == 1
-    nc = ic_dstore.not_completed[0].read()
+def test_not_completed(nc_dstore):
+    assert len(nc_dstore.not_completed) == 3
+    nc = nc_dstore.not_completed[0].read()
     assert isinstance(nc, str)
 
 
@@ -181,8 +219,8 @@ def test_multi_write(fasta_dir, w_dstore):
     """correctly write multiple files to data store"""
     expect_a = Path(fasta_dir / "brca1.fasta").read_text()
     expect_b = Path(fasta_dir / "primates_brca1.fasta").read_text()
-    identifier_a = "brca1.fasta"
-    identifier_b = "primates_brca1.fasta"
+    identifier_a = "brca2.fasta"
+    identifier_b = "primates_brca2.fasta"
     w_dstore.write(identifier_a, expect_a)
     w_dstore.write(identifier_b, expect_b)
     got_a = w_dstore.read(identifier_a)
@@ -192,90 +230,42 @@ def test_multi_write(fasta_dir, w_dstore):
     assert got_b == expect_b
 
 
-'''
-def test_write_wout_suffix(w_dstore):
-    """appends suffix expected to records"""
-    with pytest.raises(ValueError):
-        w_dstore.write("1", str(dict(a=24, b="some text")))
-
-    w_dstore.write("1.fasta", str(dict(a=24, b="some text")))
-    assert len(w_dstore)== 1
-
-@skipIf(sys.platform.lower() != "darwin", "broken on linux")
-def test_md5_write(write__dir, w_dstore):
-    """tracks md5 sums of written data"""
-    expect = Path( write_dir / "brca1.fasta").read_text()
-
-    identifier = make_identifier(w_dstore, "brca1.fasta", absolute=True)
-    abs_id = w_dstore.write(identifier, expect)
-    md5 = "05a7302479c55c0b5890b50f617c5642"
-    assert w_dstore.md5(abs_id) == md5
-    assert w_dstore[0].md5 == md5
-
-    # does not have md5 if not set
-    identifier = make_identifier(w_dstore, "brca1.fasta", absolute=True)
-    abs_id = w_dstore.write(identifier, expect)
-    got = w_dstore.md5(abs_id, force=False)
-    assert got is None
-    # but if you set force=True, you get it
-    md5 = "05a7302479c55c0b5890b50f617c5642"
-    got = w_dstore.md5(abs_id, force=True)
-    assert got == md5
-
-
-def test_add_file():
-    """correctly add an arbitrarily named file"""
-    data = Path(f"data{os.sep}brca1.fasta").read_text()
-
-    log_path = os.path.join(dirname, "some.log")
-    with open(log_path, "w") as out:
-        out.write("some text")
-
-    path = os.path.join(dirname, basedir)
-    dstore = DataStoreDirectory(path, suffix=".fa", create=True)
-    _ = dstore.write("brca1.fa", data)
-    dstore.add_file(log_path)
-    assert "some.log" in dstore
-    assert os.path.exists(log_path)
-
-    log_path = os.path.join(dirname, "some.log")
-    with open(log_path, "w") as out:
-        out.write("some text")
-
-    path = os.path.join(dirname, basedir)
-    dstore = DataStoreDirectory(path, suffix=".fa", create=True)
-    _ = dstore.write("brca1.fa", data)
-    dstore.add_file(log_path, cleanup=True)
-    assert "some.log" in dstore
-    assert not os.path.exists(log_path)
-
-def test_make_identifier():
-    """correctly construct an identifier for a new member"""
-
-    if dirname.startswith(f".{os.sep}"):
-        dirname = dirname[2:]
-
-    path = os.path.join(dirname, basedir)
-    base_path = path.replace(".zip", "")
-    dstore = DataStoreDirectory(path, suffix=".json", create=True)
-    name = "brca1.fasta"
-    got = make_identifier(dstore, name, absolute = True)
-    expect = os.path.join(base_path, name.replace("fasta", "json"))
-    assert got == expect
-
-    # now using a DataMember
-    member = DataMember(
-        os.path.join(f"blah{os.sep}blah", f"2-{name}"), None
-    )
-    got = make_identifier(dstore, member, absolute=True)
-    expect = os.path.join(base_path, member.name.replace("fasta", "json"))
-    assert got == expect
-
-
-def test_summary_logs():
-    ...
-
-def test_summary_not_completed():
-   ...
-
-'''
+def test_append(w_dstore):
+    """correctly write content"""
+    identifier = "test1.fasta"
+    data = "test data"
+    w_dstore.write(identifier, data)
+    got = w_dstore.read(identifier)
+    assert got == data
+
+
+def test_notcompleted(nc_dstore):
+    assert len(nc_dstore.not_completed) == 3
+    nc_dstore.drop_not_completed()
+    assert len(nc_dstore.not_completed) == 0
+
+
+def test_no_not_completed_subdir(nc_dstore):
+    # first remove not_completed directory
+    nc_dstore.drop_not_completed()
+    Path(nc_dstore.source / _NOT_COMPLETED_TABLE).rmdir()
+    # test repr work without not_completed directory
+    assert not Path(nc_dstore.source / _NOT_COMPLETED_TABLE).exists()
+    expect = "9x member"
+    assert repr(nc_dstore).startswith(expect)
+    not_dir = nc_dstore.source / _NOT_COMPLETED_TABLE
+    not_dir.mkdir(exist_ok=True)
+
+
+def test_drop_not_completed(nc_dstore):
+    num_completed = len(nc_dstore.completed)
+    num_not_completed = len(nc_dstore.not_completed)
+    num_md5 = len(list((nc_dstore.source / _MD5_TABLE).glob("*.txt")))
+    assert num_not_completed == 3
+    assert num_completed == 6
+    assert len(nc_dstore) == 9
+    assert num_md5 == num_completed + num_not_completed
+    nc_dstore.drop_not_completed()
+    assert len(nc_dstore.not_completed) == 0
+    num_md5 = len(list((nc_dstore.source / _MD5_TABLE).glob("*.txt")))
+    assert num_md5 == num_completed