Repository: cogent3/cogent3
Pull request #1205 (2022-11-17 16:31:18)

Data store refactor


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index af5acdb6c..3f840fd62 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -1348,8 +1348,9 @@ def _apply_to(
 
     Notes
     -----
-    If run in parallel, this instance serves as the master object and
-    aggregates results.
+    This is an append only function, meaning that if a member already exists
+    in self.data_store for an input, it is skipped.
+    If run in parallel, this instance spawns workers and aggregates results.
     """
     if not self.input:
         raise RuntimeError(f"{self!r} is not part of a composed function")
@@ -1357,9 +1358,6 @@ def _apply_to(
     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
     inputs = {}
     for m in dstore:
@@ -1368,6 +1366,8 @@ def _apply_to(
         input_id = input_id.name.replace("".join(suffixes), "")
         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
 
     if not dstore:  # this should just return datastore, because if all jobs are done!
diff --git a/src/cogent3/app/data_store_new.py b/src/cogent3/app/data_store_new.py
index 17dcfafcd..670c03d3c 100644
--- a/src/cogent3/app/data_store_new.py
+++ b/src/cogent3/app/data_store_new.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import contextlib
+import inspect
 import re
 import reprlib
 import shutil
@@ -15,7 +16,7 @@ from scitrack import get_text_hexdigest
 
 from cogent3.app.typing import TabularType
 from cogent3.core.alignment import SequenceCollection
-from cogent3.util.deserialise import deserialise_not_completed
+from cogent3.util.deserialise import deserialise_object
 from cogent3.util.io import get_format_suffixes, open_
 from cogent3.util.parallel import is_master_process
 from cogent3.util.table import Table
@@ -35,17 +36,15 @@ _LOG_TABLE = "logs"
 _MD5_TABLE = "md5"
 
 
-class IfExist(Enum):
-    SKIP = "skip"
-    OVERWRITE = "overwrite"
-    RAISE = "raise"
-    READONLY = "readonly"
+class Mode(Enum):
+    r = "r"
+    w = "w"
+    a = "a"
 
 
-SKIP = IfExist.SKIP
-OVERWRITE = IfExist.OVERWRITE
-RAISE = IfExist.RAISE
-READONLY = IfExist.READONLY
+APPEND = Mode.a
+OVERWRITE = Mode.w
+READONLY = Mode.r
 
 
 class DataMemberABC(ABC):
@@ -64,22 +63,34 @@ class DataMemberABC(ABC):
         return self.data_store.read(self.unique_id)
 
     def write(self, data: str) -> None:
-        self.data_store.write(self.unique_id, data)
+        self.data_store.write(unique_id=self.unique_id, data=data)
 
 
 class DataStoreABC(ABC):
+    def __new__(klass, *args, **kwargs):
+        obj = object.__new__(klass)
+
+        init_sig = inspect.signature(klass.__init__)
+        bargs = init_sig.bind_partial(klass, *args, **kwargs)
+        bargs.apply_defaults()
+        init_vals = bargs.arguments
+        init_vals.pop("self", None)
+
+        obj._init_vals = init_vals
+        return obj
+
     def __init__(
         self,
         source: str | Path,
-        if_dest_exists: IfExist = READONLY,
-        if_member_exists: IfExist = RAISE,
+        mode: Mode = READONLY,
     ):
-        self._if_dest_exists = IfExist(if_dest_exists)
-        self._if_member_exists = IfExist(if_member_exists)
+        self._mode = Mode(mode)
 
-    @abstractmethod
     def __repr__(self):
-        ...
+        num = len(self.members)
+        name = self.__class__.__name__
+        sample = f"{list(self[:2])}..." if num > 2 else list(self)
+        return f"{num}x member {name}(source='{self.source}', members={sample})"
 
     def __getitem__(self, index):
         return self.members[index]
@@ -96,15 +107,13 @@ class DataStoreABC(ABC):
         ...
 
     def _check_writable(self, unique_id: str):
-        if self._if_dest_exists is READONLY:
+        if self._mode is READONLY:
             raise IOError("datastore is readonly")
-        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"
-            )
+        elif unique_id in self and self._mode is APPEND:
+            raise IOError("cannot overwrite existing record in append mode")
 
     @abstractmethod
-    def write(self, unique_id: str, data: str | bytes) -> None:
+    def write(self, *, unique_id: str, data: str | bytes) -> None:
         self._check_writable(unique_id)
 
     @abstractmethod
@@ -115,14 +124,9 @@ class DataStoreABC(ABC):
     def write_log(self, unique_id: str, data: str | bytes) -> None:
         self._check_writable(unique_id)
 
-    @abstractmethod
-    def describe(self) -> TabularType:
-        ...
-
     @property
-    @abstractmethod
-    def members(self) -> list[DataMemberABC]:
-        ...
+    def members(self) -> list[DataMember]:
+        return self.completed + self.not_completed
 
     def __iter__(self):
         yield from self.members
@@ -143,19 +147,114 @@ class DataStoreABC(ABC):
         ...
 
     @property
-    @abstractmethod
     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.unique_id]
+            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
-    @abstractmethod
     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_object(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")
+
+    @property
+    def describe(self) -> TabularType:
+        title = "Directory datastore"
+        num_not_completed = 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_not_completed],
+                ["logs", num_logs],
+            ],
+            title=title,
+        )
 
     @abstractmethod
     def drop_not_completed(self):
         ...
 
+    def validate(self) -> TabularType:
+        correct_md5 = len(self.members)
+        for m in self.members:
+            data = m.read()
+            md5 = self.md5(m.unique_id)
+            if md5 != get_text_hexdigest(data):
+                correct_md5 -= 1
+        incorrect_md5 = len(self.members) - correct_md5
+
+        return Table(
+            header=["Condition", "Value"],
+            data=[
+                ["Num md5sum correct", correct_md5],
+                ["Num md5sum incorrect", incorrect_md5],
+                ["Has log", len(self.logs) > 0],
+            ],
+            title="validate status",
+        )
+
 
 class DataMember(DataMemberABC):
     def __init__(self, data_store: DataStoreABC = None, unique_id: str = None):
@@ -182,14 +281,12 @@ class DataStoreDirectory(DataStoreABC):
     def __init__(
         self,
         source,
-        if_dest_exists: IfExist = READONLY,
-        if_member_exists: IfExist = RAISE,
+        mode: Mode = READONLY,
         suffix=None,
         limit=None,
         verbose=False,
-        md5=True,
     ):
-        super().__init__(source, if_dest_exists, if_member_exists)
+        super().__init__(source, mode)
         suffix = suffix or ""
         if suffix != "*":  # wild card search for all
             suffix = re.sub(r"^[\s.*]+", "", suffix)  # tidy the suffix
@@ -200,23 +297,26 @@ class DataStoreDirectory(DataStoreABC):
         self._not_completed = []
         self._limit = limit
         self._verbose = verbose
-        self._md5 = md5
-        self._source_check_create(if_dest_exists)
+        self._source_check_create(mode)
 
-    def _source_check_create(self, if_dest_exists):
+    def __contains__(self, item: str):
+        item = f"{item}.{self.suffix}" if self.suffix not in item else item
+        return super().__contains__(item)
+
+    def _source_check_create(self, mode: Mode) -> None:
         if not is_master_process():
             return
 
         sub_dirs = [_NOT_COMPLETED_TABLE, _LOG_TABLE, _MD5_TABLE]
         source = self.source
-        if if_dest_exists is READONLY and not source.exists():
+        if mode is READONLY and not source.exists():
             raise IOError(f"'{source}' does not exist")
-        elif if_dest_exists is RAISE and source.exists():
-            raise IOError(f"'{source}' exists")
-        elif if_dest_exists is READONLY:
+        # elif mode is RAISE and source.exists():
+        #     raise IOError(f"'{source}' exists")
+        elif mode is READONLY:
             return
 
-        if source.exists() and if_dest_exists is OVERWRITE:
+        if source.exists() and mode is READONLY:
             for sub_dir in sub_dirs:
                 with contextlib.suppress((FileNotFoundError, NotADirectoryError)):
                     shutil.rmtree(source / sub_dir)
@@ -292,16 +392,10 @@ class DataStoreDirectory(DataStoreABC):
                 )
         return self._not_completed
 
-    @property
-    def members(self) -> list[DataMember]:
-        return self.completed + self.not_completed
-
-    # need to update write methods to append to _completed and _not_completed
-
     def _write(
-        self, subdir: str, unique_id: str, suffix: str, data: str, md5: bool
+        self, *, subdir: str, unique_id: str, suffix: str, data: str
     ) -> DataMember:
-        super().write(unique_id, data)
+        super().write(unique_id=unique_id, data=data)
         assert suffix, "Must provide suffix"
         # check suffix compatible with this datastore
         sfx, cmp = get_format_suffixes(unique_id)
@@ -319,126 +413,39 @@ class DataStoreDirectory(DataStoreABC):
         with open_(self.source / subdir / unique_id, mode="w") as out:
             out.write(data)
 
+        if subdir == _LOG_TABLE:
+            return None
         if subdir == _NOT_COMPLETED_TABLE:
             member = DataMember(self, Path(_NOT_COMPLETED_TABLE) / unique_id)
         elif not subdir:
             member = DataMember(self, unique_id)
-        else:
-            assert subdir == _LOG_TABLE
-            member = None
 
-        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)
+        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)
 
         return member
 
-    def write(self, unique_id: str, data: str) -> DataMember:
-        member = self._write("", unique_id, self.suffix, data, True)
+    def write(self, *, unique_id: str, data: str) -> DataMember:
+        member = self._write(
+            subdir="", unique_id=unique_id, suffix=self.suffix, data=data
+        )
         if member is not None:
             self._completed.append(member)
         return member
 
     def write_not_completed(self, unique_id: str, data: str) -> DataMember:
-        member = self._write(_NOT_COMPLETED_TABLE, unique_id, "json", data, True)
+        member = self._write(
+            subdir=_NOT_COMPLETED_TABLE, unique_id=unique_id, suffix="json", data=data
+        )
         if member is not None:
             self._not_completed.append(member)
         return member
 
     def write_log(self, unique_id: str, data: str) -> None:
-        _ = 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):
-        num = len(self.members)
-        name = self.__class__.__name__
-        sample = f"{list(self[:2])}..." if num > 2 else list(self)
-        return f"{num}x member {name}(source='{self.source}', members={sample})"
+        _ = self._write(subdir=_LOG_TABLE, unique_id=unique_id, suffix="log", data=data)
 
 
 @singledispatch
diff --git a/src/cogent3/app/io.py b/src/cogent3/app/io.py
index e2d2b8cc2..bf0382aee 100644
--- a/src/cogent3/app/io.py
+++ b/src/cogent3/app/io.py
@@ -426,7 +426,7 @@ class write_seqs_new:
             return self.data_store.write_not_completed(data.source, data.to_json())
 
         data = self._formatter(data.to_dict())
-        return self.data_store.write(identifier, data)
+        return self.data_store.write(unique_id=identifier, data=data)
 
 
 @define_app(app_type=WRITER)
diff --git a/src/cogent3/app/sqlite_data_store.py b/src/cogent3/app/sqlite_data_store.py
new file mode 100644
index 000000000..ebbfa35cd
--- /dev/null
+++ b/src/cogent3/app/sqlite_data_store.py
@@ -0,0 +1,326 @@
+from __future__ import annotations
+
+import contextlib
+import datetime
+import os
+import re
+import sqlite3
+
+from pathlib import Path
+from typing import Union
+
+from scitrack import get_text_hexdigest
+
+from cogent3.app.data_store_new import (
+    _LOG_TABLE,
+    APPEND,
+    READONLY,
+    DataMember,
+    DataStoreABC,
+)
+
+
+__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"
+
+_RESULT_TABLE = "result_table"
+_MEMORY = ":memory:"
+_mem_pattern = re.compile(r"^\s*[:]{0,1}memory[:]{0,1}\s*$")
+
+# create db
+def open_sqlite_db_rw(path: Union[str, Path]):
+    """creates a new sqlitedb for read/write at path, can be an in-memory db
+
+    Notes
+    -----
+    This function embeds the schema. There are four tables:
+    - complete: completed analysis objects
+    - incomplete: NotCompleted objects
+    - logs: log-file contents
+    - state: whether db is locked to a process
+
+    Returns
+    -------
+    Handle to a sqlite3 session
+    """
+    db = sqlite3.connect(
+        path,
+        isolation_level=None,
+        detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,
+    )
+    db.row_factory = sqlite3.Row
+    create_template = "CREATE TABLE IF NOT EXISTS {};"
+    # note it is essential to use INTEGER for the autoincrement of primary key to work
+    creates = [
+        "state(state_id INTEGER PRIMARY KEY, record_type TEXT, lock_pid INTEGER)",
+        f"{_LOG_TABLE}(log_id INTEGER PRIMARY KEY, log_name TEXT, date timestamp, data BLOB)",
+        f"{_RESULT_TABLE}(record_id TEXT PRIMARY KEY, log_id INTEGER, md5 BLOB, not_completed INTEGER, data BLOB)",
+    ]
+    for table in creates:
+        db.execute(create_template.format(table))
+    return db
+
+
+def open_sqlite_db_ro(path):
+    """returns db opened as read only
+
+    Returns
+    -------
+    Handle to a sqlite3 session
+    """
+    db = sqlite3.connect(
+        f"file:{path}?mode=ro",
+        isolation_level=None,
+        detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,
+        uri=True,
+    )
+    db.row_factory = sqlite3.Row
+    return db
+
+
+class SqliteDataStore(DataStoreABC):
+    store_suffix = "sqlitedb"
+
+    def __init__(
+        self,
+        source,
+        mode: Mode,
+        limit=None,
+        verbose=False,
+    ):
+        if _mem_pattern.search(str(source)):
+            self.source = _MEMORY
+        else:
+            source = Path(source).expanduser()
+            self.source = (
+                source
+                if source.suffix[1:] == self.store_suffix  # sliced to remove "."
+                else Path(f"{source}.{self.store_suffix}")
+            )
+
+        super().__init__(
+            source=source,
+            mode=mode,
+        )
+        if mode is not READONLY and limit is not None:
+            raise ValueError(
+                "Using limit argument is only valid for readonly datastores"
+            )
+        self._limit = limit
+        self._verbose = verbose
+        self._checksums = {}
+        self._db = None
+        self._members = []
+        self._completed = []
+        self._not_completed = []
+        self._open = False
+        self._log_id = None
+
+    def __getstate__(self):
+        return {**self._init_vals}
+
+    def __setstate__(self, state):
+        # this will reset connections to read only db's
+        obj = self.__class__(**state)
+        self.__dict__.update(obj.__dict__)
+
+    def __contains__(self, unique_id: str):
+        """whether relative identifier has been stored"""
+        # todo add limit here
+        query = f"SELECT count(*) as c FROM {_RESULT_TABLE} where record_id == ?"
+        result = self.db.execute(query, (unique_id,)).fetchone()
+        if result["c"] > 0:
+            return True
+        return False
+
+    @property
+    def db(self):
+        if self._db is None:
+            db_func = open_sqlite_db_ro if self._mode is READONLY else open_sqlite_db_rw
+            self._db = db_func(self.source)
+            self._open = True
+
+            if self._mode is not READONLY:
+                pid = os.getpid()
+                self._db.execute("INSERT INTO state(lock_pid) VALUES (?)", (pid,))
+
+        # todo: lock_id comes from process id, into state table  #new  check this on write
+
+        return self._db
+
+    def _init_log(self):
+        timestamp = datetime.datetime.now()
+        self.db.execute(f"INSERT INTO {_LOG_TABLE}(date) VALUES (?)", (timestamp,))
+        self._log_id = self._db.execute(
+            f"SELECT log_id FROM {_LOG_TABLE} where date = ?", (timestamp,)
+        ).fetchone()["log_id"]
+
+    def close(self):
+        with contextlib.suppress(sqlite3.ProgrammingError):
+            self.db.close()
+        self._open = False
+
+    @property
+    def _lock_id(self):
+        """returns lock_pid"""
+        result = self.db.execute("SELECT lock_pid FROM state").fetchone()
+        return result[0] if result else result
+
+    def read(self, identifier: str) -> str | bytes:
+        """
+        identifier string formed from Path(table_name) / identifier
+        """
+        identifier = Path(identifier)
+        table_name = str(identifier.parent)
+        if table_name not in (
+            _RESULT_TABLE,
+            _LOG_TABLE,
+        ):
+            raise ValueError(f"unknown table for {str(identifier)!r}")
+        if table_name != _LOG_TABLE:
+            cmnd = f"SELECT * FROM {_RESULT_TABLE} WHERE record_id = ?"
+            result = self.db.execute(cmnd, (identifier.name,)).fetchone()
+            return result["data"]
+
+        cmnd = f"SELECT * FROM {_LOG_TABLE} WHERE log_name = ?"
+        result = self.db.execute(cmnd, (identifier.name,)).fetchone()
+
+        return result["data"]
+
+    @property
+    def completed(self):
+        if not self._completed:
+            self._completed = self._select_members(
+                table_name=_RESULT_TABLE, not_completed=False
+            )
+        return self._completed
+
+    @property
+    def not_completed(self):
+        """returns database records of type NotCompleted"""
+        if not self._not_completed:
+            self._not_completed = self._select_members(
+                table_name=_RESULT_TABLE, not_completed=True
+            )
+        return self._not_completed
+
+    def _select_members(self, table_name: str, not_completed: bool) -> list[DataMember]:
+        limit = f"LIMIT {self._limit}" if self._limit else ""
+        cmnd = self.db.execute(
+            f"SELECT record_id FROM {table_name} WHERE not_completed=? {limit}",
+            (not_completed,),
+        )
+        return [
+            DataMember(data_store=self, unique_id=Path(table_name) / r["record_id"])
+            for r in cmnd.fetchall()
+        ]
+
+    @property
+    def logs(self):
+        """returns all log records"""
+        cmnd = self.db.execute(f"SELECT data, log_name FROM {_LOG_TABLE}")
+        return [
+            DataMember(data_store=self, unique_id=Path(_LOG_TABLE) / r["log_name"])
+            for r in cmnd.fetchall()
+        ]
+
+    def _write(
+        self, *, table_name: str, unique_id: str, data: str, not_completed: bool
+    ) -> DataMember | None:
+        """
+        Parameters
+        ----------
+        table_name: str
+            name of table to save data. It must be _RESULT_TABLE or _LOG_TABLE.
+        unique_id : str
+            unique identifier that data will be saved under.
+        data: str
+            data to be saved.
+        not_completed: bool
+            flag to identify not_completed results
+
+        Returns
+        -------
+        DataMember instance or None when writing to _LOG_TABLE
+        """
+        if self._log_id is None:
+            self._init_log()
+        if table_name == _LOG_TABLE:
+            cmnd = f"UPDATE {table_name} SET data =?, log_name =? WHERE log_id=?"
+            self.db.execute(cmnd, (data, unique_id, self._log_id))
+            return None
+        md5 = get_text_hexdigest(data)
+
+        if unique_id in self and self._mode is not APPEND:
+            cmnd = f"UPDATE {table_name} SET data= ?, log_id=?, md5=? WHERE record_id=?"
+            self.db.execute(cmnd, (data, self._log_id, md5, unique_id))
+        else:
+            cmnd = f"INSERT INTO {table_name} (record_id,data,log_id,md5,not_completed) VALUES (?,?,?,?,?)"
+            self.db.execute(cmnd, (unique_id, data, self._log_id, md5, not_completed))
+        return DataMember(self, Path(table_name) / unique_id)
+
+    def drop_not_completed(self) -> None:
+        # self.db.execute(f"DELETE FROM {_MD5_TABLE} WHERE record_id IN (SELECT record_id FROM {_RESULT_TABLE} WHERE not_completed=1)")
+        self.db.execute(f"DELETE FROM {_RESULT_TABLE} WHERE not_completed=1")
+        self._not_completed = []
+
+    def write(self, *, unique_id: str, data: str) -> DataMember:
+        super().write(unique_id=unique_id, data=data)
+        member = self._write(
+            table_name=_RESULT_TABLE,
+            unique_id=unique_id,
+            data=data,
+            not_completed=False,
+        )
+        if member is not None:
+            self._completed.append(member)
+        return member
+
+    def write_log(self, *, unique_id: str, data: str) -> None:
+        super().write_log(unique_id=unique_id, data=data)
+        _ = self._write(
+            table_name=_LOG_TABLE, unique_id=unique_id, data=data, not_completed=False
+        )
+
+    def write_not_completed(self, *, unique_id: str, data: str) -> DataMember:
+        super().write_not_completed(unique_id=unique_id, data=data)
+        member = self._write(
+            table_name=_RESULT_TABLE, unique_id=unique_id, data=data, not_completed=True
+        )
+        if member is not None:
+            self._not_completed.append(member)
+        return member
+
+    def md5(self, unique_id: str) -> str:
+        """
+        Parameters
+        ----------
+        unique_id
+            name of data store member
+        Returns
+        -------
+        md5 checksum for the member, if available, None otherwise
+        """
+        unique_id = Path(unique_id)
+        unique_id = re.sub(rf"[.]({self.store_suffix}|json)$", ".txt", unique_id.name)
+
+        cmnd = f"SELECT * FROM {_RESULT_TABLE} WHERE record_id = ?"
+        result = self.db.execute(cmnd, (unique_id,)).fetchone()
+
+        return result["md5"]
+
+    @property
+    def describe(self):
+        if self._lock_id:
+            title = f"Locked db store. Locked to pid={self._lock_id}, current pid={os.getpid()}."
+        else:
+            title = "Unlocked db store."
+        table = super().describe
+        table.title = title
+        return table


Test patch:
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index d3f51d5bd..117518d8c 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -1,10 +1,10 @@
 import inspect
 import os
-import pathlib
 import pickle
+import shutil
 
+from pathlib import Path
 from pickle import dumps, loads
-from tempfile import TemporaryDirectory
 from unittest import main
 from unittest.mock import Mock
 
@@ -19,7 +19,6 @@ from cogent3.app import io as io_app
 from cogent3.app import sample as sample_app
 from cogent3.app import translate, tree
 from cogent3.app.composable import (
-    GENERIC,
     NON_COMPOSABLE,
     WRITER,
     NotCompleted,
@@ -32,7 +31,13 @@ from cogent3.app.composable import (
     is_composable,
     user_function,
 )
-from cogent3.app.data_store_new import DataStoreDirectory, get_data_source
+from cogent3.app.data_store_new import (
+    APPEND,
+    OVERWRITE,
+    READONLY,
+    DataStoreDirectory,
+    get_data_source,
+)
 from cogent3.app.sample import min_length, omit_degenerates
 from cogent3.app.translate import select_translatable
 from cogent3.app.tree import quick_tree
@@ -54,11 +59,119 @@ __email__ = "Gavin.Huttley@anu.edu.au"
 __status__ = "Alpha"
 
 
+DATA_DIR = Path(__file__).parent.parent / "data"
+
+
 @pytest.fixture(scope="function")
 def tmp_dir(tmpdir_factory):
     return tmpdir_factory.mktemp("datastore")
 
 
+@pytest.fixture(scope="function")
+def fasta_dir(tmp_dir):
+    tmp_dir = Path(tmp_dir)
+    filenames = DATA_DIR.glob("*.fasta")
+    fasta_dir = tmp_dir / "fasta"
+    fasta_dir.mkdir(parents=True, exist_ok=True)
+    for fn in filenames:
+        dest = fasta_dir / fn.name
+        dest.write_text(fn.read_text())
+    return fasta_dir
+
+
+@pytest.fixture(scope="function")
+def write_dir1(tmp_dir):
+    tmp_dir = Path(tmp_dir)
+    write_dir1 = tmp_dir / "write1"
+    write_dir1.mkdir(parents=True, exist_ok=True)
+    yield write_dir1
+    shutil.rmtree(write_dir1)
+
+
+@pytest.fixture(scope="function")
+def write_dir2(tmp_dir):
+    tmp_dir = Path(tmp_dir)
+    write_dir2 = tmp_dir / "write2"
+    write_dir2.mkdir(parents=True, exist_ok=True)
+    yield write_dir2
+    shutil.rmtree(write_dir2)
+
+
+@pytest.fixture(scope="function")
+def ro_dstore(fasta_dir):
+    return DataStoreDirectory(fasta_dir, suffix="fasta", mode=READONLY)
+
+
+@pytest.fixture(scope="function")
+def completed_objects(ro_dstore):
+    return {f"{Path(m.unique_id).stem}": m.read() for m in ro_dstore}
+
+
+@pytest.fixture(scope="function")
+def nc_objects():
+    return {
+        f"id_{i}": NotCompleted("ERROR", "location", "message", source=f"id_{i}")
+        for i in range(3)
+    }
+
+
+@pytest.fixture(scope="function")
+def log_data():
+    path = DATA_DIR / "scitrack.log"
+    return path.read_text()
+
+
+@pytest.fixture(scope="function")
+def full_dstore(write_dir1, nc_objects, completed_objects, log_data):
+    dstore = DataStoreDirectory(write_dir1, suffix="fasta", mode=OVERWRITE)
+    for id, data in nc_objects.items():
+        dstore.write_not_completed(unique_id=id, data=data.to_json())
+
+    for id, data in completed_objects.items():
+        dstore.write(unique_id=id, data=data)
+
+    dstore.write_log(unique_id="scitrack.log", data=log_data)
+    return dstore
+
+
+@pytest.fixture(scope="function")
+def half_dstore1(write_dir1, nc_objects, completed_objects, log_data):
+    dstore = DataStoreDirectory(write_dir1, suffix="fasta", mode=OVERWRITE)
+    i = 0
+    for id, data in nc_objects.items():
+        dstore.write_not_completed(unique_id=id, data=data.to_json())
+        i += 1
+        if i >= len(nc_objects.items()) / 2:
+            break
+    i = 0
+    for id, data in completed_objects.items():
+        dstore.write(unique_id=id, data=data)
+        i += 1
+        if i >= len(completed_objects.items()) / 2:
+            break
+    dstore.write_log(unique_id="scitrack.log", data=log_data)
+    return dstore
+
+
+@pytest.fixture(scope="function")
+def half_dstore2(write_dir2, nc_objects, completed_objects, log_data):
+    dstore = DataStoreDirectory(write_dir2, suffix="fasta", mode=OVERWRITE)
+    i = -1
+    for id, data in nc_objects.items():
+        i += 1
+        if i < len(nc_objects.items()) / 2:
+            continue
+        dstore.write_not_completed(unique_id=id, data=data.to_json())
+    i = -1
+    for id, data in completed_objects.items():
+        i += 1
+        if i < len(completed_objects.items()) / 2:
+            continue
+        dstore.write(unique_id=id, data=data)
+    dstore.write_log(unique_id="scitrack.log", data=log_data)
+    return dstore
+
+
 def test_composable():
     """correctly form string"""
 
@@ -189,7 +302,7 @@ def test_disconnect():
 
 def test_as_completed():
     """correctly applies iteratively"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    dstore = io_app.get_data_store(DATA_DIR, suffix="fasta", limit=3)
     reader = io_app.load_unaligned(format="fasta", moltype="dna")
     got = list(reader.as_completed(dstore))
     assert len(got) == len(dstore)
@@ -220,14 +333,12 @@ def test_as_completed():
 @pytest.mark.parametrize("klass", (DataStoreDirectory,))
 def test_apply_to_strings(tmp_dir, klass):
     """apply_to handles strings as paths"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    dstore = io_app.get_data_store(DATA_DIR, suffix="fasta", limit=3)
     dstore = [str(m) for m in dstore]
     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.write_seqs_new(
-        klass(outpath, if_dest_exists="overwrite", suffix="fasta")
-    )
+    writer = io_app.write_seqs_new(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)
@@ -244,7 +355,7 @@ def test_apply_to_non_unique_identifiers(tmp_dir):
     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")
+        DataStoreDirectory(outpath, mode=OVERWRITE, suffix="fasta")
     )
     process = reader + min_length + writer
     with pytest.raises(ValueError):
@@ -253,7 +364,7 @@ def test_apply_to_non_unique_identifiers(tmp_dir):
 
 def test_apply_to_logging(tmp_dir):
     """correctly creates log file"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    dstore = io_app.get_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")
@@ -267,7 +378,7 @@ def test_apply_to_logging(tmp_dir):
 
 def test_apply_to_logger(tmp_dir):
     """correctly uses user provided logger"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    dstore = io_app.get_data_store(DATA_DIR, suffix="fasta", limit=3)
     LOGGER = CachingLogger()
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     min_length = sample_app.min_length(10)
@@ -281,7 +392,7 @@ def test_apply_to_logger(tmp_dir):
 
 def test_apply_to_invalid_logger(tmp_dir):
     """incorrect logger value raises TypeError"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    dstore = io_app.get_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)
@@ -295,7 +406,7 @@ def test_apply_to_invalid_logger(tmp_dir):
 
 def test_apply_to_not_completed(tmp_dir):
     """correctly creates notcompleted"""
-    dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    dstore = io_app.get_data_store(DATA_DIR, suffix="fasta", limit=3)
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     # trigger creation of notcompleted
     min_length = sample_app.min_length(3000)
@@ -309,9 +420,9 @@ def test_apply_to_not_completed(tmp_dir):
 
 def test_apply_to_not_partially_done(tmp_dir):
     """correctly applies process when result already partially done"""
-    dstore = io_app.get_data_store("data", suffix="fasta")
+    dstore = io_app.get_data_store(DATA_DIR, suffix="fasta")
     num_records = len(dstore)
-    dirname = pathlib.Path(tmp_dir)
+    dirname = Path(tmp_dir)
     reader = io_app.load_aligned(format="fasta", moltype="dna")
     outpath = dirname / "delme.tinydb"
     writer = io_app.write_db(outpath)
@@ -997,5 +1108,27 @@ def test_handles_None():
     __app_registry.pop(get_object_provenance(none_out), None)
 
 
+def test_apply_to_only_appends(half_dstore1, half_dstore2):
+    half_dstore1._mode = APPEND
+    reader1 = io_app.load_aligned(format="fasta", moltype="dna")
+    min_length1 = sample_app.min_length(10)
+    writer1 = io_app.write_seqs_new(half_dstore1)
+    process1 = reader1 + min_length1 + writer1
+    # create paths as strings
+    dstore1 = io_app.get_data_store(half_dstore1.source, suffix="fasta")
+    dstore1 = [str(m) 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)
+
+    half_dstore2._mode = APPEND
+    reader2 = io_app.load_aligned(format="fasta", moltype="dna")
+    min_length2 = sample_app.min_length(10)
+    writer2 = io_app.write_seqs_new(half_dstore2)
+    process2 = reader2 + min_length2 + writer2
+    # check not fail on append new records
+    _ = process2.apply_to(dstore1, id_from_source=get_data_source)
+
+
 if __name__ == "__main__":
     main()
diff --git a/tests/test_app/test_data_store_new.py b/tests/test_app/test_data_store_new.py
index 8c5c2900b..88d28f449 100644
--- a/tests/test_app/test_data_store_new.py
+++ b/tests/test_app/test_data_store_new.py
@@ -9,14 +9,15 @@ from scitrack import get_text_hexdigest
 
 from cogent3.app.composable import NotCompleted
 from cogent3.app.data_store_new import (
-    _LOG_TABLE,
     _MD5_TABLE,
     _NOT_COMPLETED_TABLE,
+    APPEND,
     OVERWRITE,
     READONLY,
-    SKIP,
+    DataMember,
     DataStoreDirectory,
 )
+from cogent3.util.table import Table
 
 
 __author__ = "Gavin Huttley"
@@ -75,17 +76,17 @@ def nc_dir(tmp_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", mode=READONLY)
 
 
 @pytest.fixture(scope="function")
 def w_dstore(write_dir):
-    return DataStoreDirectory(write_dir, suffix="fasta", if_dest_exists=OVERWRITE)
+    return DataStoreDirectory(write_dir, suffix="fasta", mode=OVERWRITE)
 
 
 @pytest.fixture(scope="function")
 def nc_dstore(nc_dir):
-    dstore = DataStoreDirectory(nc_dir, suffix="fasta", if_dest_exists=SKIP)
+    dstore = DataStoreDirectory(nc_dir, suffix="fasta", mode=OVERWRITE)  # SKIP)
     # write one log file
     log_filename = "scitrack.log"
     dstore.write_log(log_filename, (DATA_DIR / log_filename).read_text())
@@ -104,13 +105,53 @@ def nc_dstore(nc_dir):
     # write six fasta file
     for fn in filenames:
         identifier = fn.name
-        dstore.write(identifier, fn.read_text())
+        dstore.write(unique_id=identifier, data=fn.read_text())
     return dstore
 
 
+@pytest.fixture(scope="function")
+def completed_objects(ro_dstore):
+    return {f"{Path(m.unique_id).stem}": m.read() for m in ro_dstore}
+
+
+@pytest.fixture(scope="function")
+def nc_objects():
+    return {
+        f"id_{i}": NotCompleted("ERROR", "location", "message", source=f"id_{i}")
+        for i in range(3)
+    }
+
+
+@pytest.fixture(scope="function")
+def log_data():
+    path = DATA_DIR / "scitrack.log"
+    return path.read_text()
+
+
+@pytest.fixture(scope="function")
+def full_dstore(write_dir, nc_objects, completed_objects, log_data):
+    dstore = DataStoreDirectory(write_dir, suffix="fasta", mode=OVERWRITE)
+    for id, data in nc_objects.items():
+        dstore.write_not_completed(unique_id=id, data=data.to_json())
+
+    for id, data in completed_objects.items():
+        dstore.write(unique_id=id, data=data)
+
+    dstore.write_log(unique_id="scitrack.log", data=log_data)
+    return dstore
+
+
+def test_fail_try_append(full_dstore, completed_objects):
+    full_dstore._mode = APPEND
+    id, data = list(completed_objects.items())[0]
+    with pytest.raises(IOError):
+        full_dstore.write(unique_id=id, data=data)
+
+
 def test_contains(ro_dstore):
     """correctly identify when a data store contains a member"""
     assert "brca1.fasta" in ro_dstore
+    assert "brca1" in ro_dstore
 
 
 def test_iter(ro_dstore):
@@ -119,6 +160,12 @@ def test_iter(ro_dstore):
     assert members == ro_dstore.members
 
 
+def test_members(ro_dstore):
+    for m in ro_dstore:
+        assert isinstance(m, DataMember)
+    assert len(ro_dstore) == len(ro_dstore.completed) + len(ro_dstore.not_completed)
+
+
 def test_len(ro_dstore):
     """DataStore returns correct len"""
     expect = len(list(ro_dstore.source.glob("*.fasta")))
@@ -186,9 +233,23 @@ def test_not_completed(nc_dstore):
     assert isinstance(nc, str)
 
 
+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
+
+
 def test_write_read_only_datastore(ro_dstore):
     with pytest.raises(IOError):
-        ro_dstore.write("brca1.fasta", "test data")
+        ro_dstore.write(unique_id="brca1.fasta", data="test data")
 
 
 def test_write_read_only_member(ro_dstore):
@@ -200,7 +261,7 @@ def test_write(fasta_dir, w_dstore):
     """correctly write content"""
     expect = Path(fasta_dir / "brca1.fasta").read_text()
     identifier = "brca1.fasta"
-    w_dstore.write(identifier, expect)
+    w_dstore.write(unique_id=identifier, data=expect)
     got = w_dstore.read(identifier)
     assert got == expect
 
@@ -211,8 +272,8 @@ def test_multi_write(fasta_dir, w_dstore):
     expect_b = Path(fasta_dir / "primates_brca1.fasta").read_text()
     identifier_a = "brca2.fasta"
     identifier_b = "primates_brca2.fasta"
-    w_dstore.write(identifier_a, expect_a)
-    w_dstore.write(identifier_b, expect_b)
+    w_dstore.write(unique_id=identifier_a, data=expect_a)
+    w_dstore.write(unique_id=identifier_b, data=expect_b)
     got_a = w_dstore.read(identifier_a)
     got_b = w_dstore.read(identifier_b)
     # check that both bits of data match
@@ -224,17 +285,11 @@ def test_append(w_dstore):
     """correctly write content"""
     identifier = "test1.fasta"
     data = "test data"
-    w_dstore.write(identifier, data)
+    w_dstore.write(unique_id=identifier, data=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):
     expect = f"{len(nc_dstore.completed)+len(nc_dstore.not_completed)}x member"
     assert repr(nc_dstore).startswith(expect)
@@ -251,20 +306,6 @@ def test_no_not_completed_subdir(nc_dstore):
     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
-
-
 def test_limit_datastore(nc_dstore):
     assert len(nc_dstore) == len(nc_dstore.completed) + len(nc_dstore.not_completed)
     nc_dstore._limit = len(nc_dstore.completed)
@@ -278,3 +319,22 @@ def test_md5_sum(nc_dstore):
         data = m.read()
         md5 = nc_dstore.md5(m.unique_id)
         assert md5 == get_text_hexdigest(data)
+
+
+def test_summary_logs(full_dstore):
+    # log summary has a row per log file and a column for each property
+    got = full_dstore.summary_logs
+    assert got.shape == (1, 6)
+    assert isinstance(got, Table)
+
+
+def test_summary_not_completed(full_dstore):
+    got = full_dstore.summary_not_completed
+    assert got.shape >= (1, 1)
+    assert isinstance(got, Table)
+
+
+def test_describe(full_dstore):
+    got = full_dstore.describe
+    assert got.shape >= (3, 2)
+    assert isinstance(got, Table)
diff --git a/tests/test_app/test_io_new.py b/tests/test_app/test_io_new.py
index 8945b1e55..0889ff246 100644
--- a/tests/test_app/test_io_new.py
+++ b/tests/test_app/test_io_new.py
@@ -4,7 +4,7 @@ import pytest
 
 from cogent3.app import io as io_app
 from cogent3.app.composable import source_proxy
-from cogent3.app.data_store_new import SKIP, DataMember, DataStoreDirectory
+from cogent3.app.data_store_new import DataMember, DataStoreDirectory, Mode
 from cogent3.core.alignment import ArrayAlignment
 from cogent3.parse.sequence import PARSERS
 
@@ -54,7 +54,7 @@ def test_write_seqs(fasta_dir, tmp_dir):
     seqs = ArrayAlignment(data=data, moltype=None)
     seqs.info.source = datastore.source
     out_data_store = DataStoreDirectory(
-        tmp_dir / "test_write_seqs", if_dest_exists="overwrite", suffix="fasta"
+        tmp_dir / "test_write_seqs", mode=Mode.w, suffix="fasta"
     )
     writer = io_app.write_seqs_new(out_data_store, format="fasta")
     wrote = writer(seqs[0], datamember.unique_id)
diff --git a/tests/test_app/test_sqlite_data_store.py b/tests/test_app/test_sqlite_data_store.py
new file mode 100644
index 000000000..ca9087bd1
--- /dev/null
+++ b/tests/test_app/test_sqlite_data_store.py
@@ -0,0 +1,418 @@
+from datetime import datetime
+from pathlib import Path
+from pickle import dumps, loads
+
+import pytest
+
+from scitrack import get_text_hexdigest
+
+from cogent3.app.composable import NotCompleted
+from cogent3.app.data_store_new import (
+    OVERWRITE,
+    READONLY,
+    DataMemberABC,
+    DataStoreDirectory,
+)
+from cogent3.app.sqlite_data_store import (
+    _LOG_TABLE,
+    _RESULT_TABLE,
+    SqliteDataStore,
+    open_sqlite_db_rw,
+)
+from cogent3.util.table import Table
+
+
+__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="function")
+def tmp_dir(tmpdir_factory):
+    return tmpdir_factory.mktemp("sqldb")
+
+
+@pytest.fixture(scope="function")
+def db_dir(tmp_dir):
+    db_dir = tmp_dir / "data.sqlitedb"
+    return db_dir
+
+
+@pytest.fixture(scope="function")
+def ro_dir_dstore():
+    return DataStoreDirectory(DATA_DIR, suffix="fasta")
+
+
+@pytest.fixture(scope="function")
+def sql_dstore(ro_dir_dstore, db_dir):
+    # we now need to write these out to a path
+    sql_dstore = SqliteDataStore(db_dir, mode=OVERWRITE)
+    for m in ro_dir_dstore:
+        sql_dstore.write(data=m.read(), unique_id=m.unique_id)
+    return sql_dstore
+
+
+@pytest.fixture(scope="function")
+def ro_sql_dstore(sql_dstore):
+    # we now need to write these out to a path
+    sql_dstore._mode = READONLY
+    return sql_dstore
+
+
+@pytest.fixture(scope="function")
+def completed_objects(ro_dir_dstore):
+    return {f"{Path(m.unique_id).stem}": m.read() for m in ro_dir_dstore}
+
+
+@pytest.fixture(scope="function")
+def nc_objects():
+    return {
+        f"id_{i}": NotCompleted("ERROR", "location", "message", source=f"id_{i}")
+        for i in range(3)
+    }
+
+
+@pytest.fixture(scope="function")
+def log_data():
+    path = Path(__file__).parent.parent / "data" / "scitrack.log"
+    return path.read_text()
+
+
+@pytest.fixture(scope="function")
+def full_dstore_directory(db_dir, nc_objects, completed_objects, log_data):
+    dstore = DataStoreDirectory(db_dir, suffix="fasta", mode=OVERWRITE)
+    for id, data in nc_objects.items():
+        dstore.write_not_completed(unique_id=id, data=data.to_json())
+
+    for id, data in completed_objects.items():
+        dstore.write(unique_id=id, data=data)
+
+    dstore.write_log(unique_id="scitrack.log", data=log_data)
+    return dstore
+
+
+@pytest.fixture(scope="function")
+def full_dstore_sqlite(db_dir, nc_objects, completed_objects, log_data):
+    dstore = SqliteDataStore(db_dir, mode=OVERWRITE)
+    for id, data in nc_objects.items():
+        dstore.write_not_completed(unique_id=id, data=data.to_json())
+    for id, data in completed_objects.items():
+        dstore.write(unique_id=id, data=data)
+    dstore.write_log(unique_id="scitrack.log", data=log_data)
+    return dstore
+
+
+@pytest.fixture(scope="function")
+def dstore_on_disk(full_dstore_sqlite):
+    path = full_dstore_sqlite.source
+    full_dstore_sqlite.close()
+    return path
+
+
+def test_open_existing(dstore_on_disk):
+    ro = SqliteDataStore(dstore_on_disk, mode=READONLY)
+    assert len(ro) > 0
+    assert len(ro.completed) > 0
+    assert len(ro.not_completed) > 0
+    assert len(ro.logs) > 0
+    for attr in ("summary_logs", "summary_not_completed", "describe"):
+        assert isinstance(getattr(ro, attr), Table)
+
+
+def test_open_to_append(dstore_on_disk):
+    ro = SqliteDataStore(dstore_on_disk, mode=OVERWRITE)
+
+
+def test_db_creation():
+    db = SqliteDataStore(":memory:", mode=OVERWRITE)
+    db = db.db
+    result = db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
+    assert len(result) == 3
+    created_names = {r["name"] for r in result}
+    assert created_names == {
+        _LOG_TABLE,
+        _RESULT_TABLE,
+        "state",
+    }
+    rows = db.execute(f"Select * from {_LOG_TABLE}").fetchall()
+    assert len(rows) == 0
+
+
+def test_db_init_log():
+    dstore = SqliteDataStore(":memory:", mode=OVERWRITE)
+    dstore._init_log()
+    rows = dstore.db.execute(f"Select * from {_LOG_TABLE}").fetchall()
+    assert len(rows) == 1
+    assert rows[0]["date"].date() == datetime.today().date()
+
+
+def test_open_sqlite_db_rw():
+    db = open_sqlite_db_rw(":memory:")
+    # should make tables for completed, not completed, md5, log and state
+    result = db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
+    assert len(result) == 3
+    created_names = {r["name"] for r in result}
+    assert created_names == {
+        _LOG_TABLE,
+        _RESULT_TABLE,
+        "state",
+    }
+
+
+def test_rw_sql_dstore_mem(completed_objects):
+    """in memory dstore"""
+    db = SqliteDataStore(":memory:", mode=OVERWRITE)
+    for unique_id, obj in completed_objects.items():
+        db.write(data=obj, unique_id=unique_id)
+    expect = len(completed_objects)
+    query = f"SELECT count(*) as c FROM {_RESULT_TABLE} WHERE not_completed=?"
+    got = db.db.execute(query, (0,)).fetchone()["c"]
+    assert got == expect, f"Failed for {_RESULT_TABLE} number of rows"
+    assert len(db.completed) == expect
+
+
+def test_not_completed(nc_objects):
+    db = SqliteDataStore(":memory:", mode=OVERWRITE)
+    for unique_id, obj in nc_objects.items():
+        db.write_not_completed(data=obj.to_json(), unique_id=unique_id)
+    expect = len(nc_objects)
+    query = f"SELECT count(*) as c FROM {_RESULT_TABLE} WHERE not_completed=?"
+    got = db.db.execute(query, (1,)).fetchone()["c"]
+    assert got == expect, f"Failed for {_RESULT_TABLE} number of rows"
+    assert len(db.not_completed) == expect
+
+
+def test_logdata(log_data):
+    db = SqliteDataStore(":memory:", mode=OVERWRITE)
+    db.write_log(data=log_data, unique_id="scitrack.log")
+    query = f"select count(*) as c from {_LOG_TABLE}"
+    got = db.db.execute(query).fetchone()["c"]
+    assert got == 1
+
+
+def test_drop_not_completed(nc_objects):
+    db = SqliteDataStore(":memory:", mode=OVERWRITE)
+    for unique_id, obj in nc_objects.items():
+        db.write_not_completed(data=obj.to_json(), unique_id=unique_id)
+    expect = len(nc_objects)
+    assert len(db.not_completed) == expect
+    db.drop_not_completed()
+    assert len(db.not_completed) == 0
+
+
+def test_contains(sql_dstore):
+    """correctly identify when a data store contains a member"""
+    assert "brca1.fasta" in sql_dstore
+
+
+def test_iter(sql_dstore):
+    """DataStore objects allow iteration over members"""
+    members = list(sql_dstore)
+    assert members == sql_dstore.members
+
+
+def test_members(sql_dstore):
+    for m in sql_dstore:
+        assert isinstance(m, DataMemberABC)
+    assert len(sql_dstore) == len(sql_dstore.completed) + len(sql_dstore.not_completed)
+
+
+def test_len(sql_dstore, ro_dir_dstore):
+    """DataStore returns correct len"""
+    expect = len(list(ro_dir_dstore.source.glob("*.fasta")))
+    assert expect == len(sql_dstore) == len(sql_dstore.members)
+
+
+def test_md5_sum(sql_dstore):
+    for m in sql_dstore.members:
+        data = m.read()
+        md5 = sql_dstore.md5(m.unique_id)
+        assert md5 == get_text_hexdigest(data)
+
+
+def test_iterall(sql_dstore, ro_dir_dstore):
+    expect = {fn.name for fn in ro_dir_dstore.source.glob("*.fasta")}
+    got = {Path(m.unique_id).name for m in sql_dstore}
+    assert expect == got
+
+
+def test_read(full_dstore_sqlite):
+    """correctly read content"""
+    c = full_dstore_sqlite.completed[0]
+    nc = full_dstore_sqlite.not_completed[0]
+    log = full_dstore_sqlite.logs[0]
+    for r in (c, nc, log):
+        assert isinstance(r.read(), str)
+
+
+def test_read_log(sql_dstore, log_data):
+    """correctly read content"""
+    sql_dstore.write_log(unique_id="brca1.fasta", data=log_data)
+
+    got = sql_dstore.read(f"{_LOG_TABLE}/brca1.fasta")
+    assert got == log_data
+
+
+def test_write(sql_dstore, ro_dir_dstore):
+    """correctly write content"""
+    expect = Path(ro_dir_dstore.source / "brca1.fasta").read_text()
+    identifier = "result_table/brca1.fasta"
+    sql_dstore.write(unique_id=identifier, data=expect)
+    got = sql_dstore.read(identifier)
+    assert got == expect
+
+
+def test_write_if_member_exists(sql_dstore, ro_dir_dstore):
+    """correctly write content"""
+    expect = Path(ro_dir_dstore.source / "brca1.fasta").read_text()
+    identifier = "result_table/brca1.fasta"
+    sql_dstore.write(unique_id=identifier, data=expect)
+    got = sql_dstore.read(identifier)
+    assert got == expect
+    sql_dstore._mode = OVERWRITE
+    identifier = "result_table/brca1.fasta"
+    sql_dstore.write(unique_id=identifier, data=expect)
+    got = sql_dstore.read(identifier)
+    assert got == expect
+
+
+def test_summary_logs(full_dstore_sqlite):
+    # log summary has a row per log file and a column for each property
+    got = full_dstore_sqlite.summary_logs
+    assert got.shape == (1, 6)
+    assert isinstance(got, Table)
+
+
+def test_summary_not_completed(full_dstore_sqlite):
+    got = full_dstore_sqlite.summary_not_completed
+    assert got.shape >= (1, 1)
+    assert isinstance(got, Table)
+
+
+def test_no_not_completed_subdir(full_dstore_sqlite):
+    expect = f"{len(full_dstore_sqlite.completed)+len(full_dstore_sqlite.not_completed)}x member"
+    assert repr(full_dstore_sqlite).startswith(expect)
+    # first remove not_completed directory
+    full_dstore_sqlite.drop_not_completed()
+    # test repr work without not_completed directory
+    expect = f"{len(full_dstore_sqlite.completed)}x member"
+    assert repr(full_dstore_sqlite).startswith(expect)
+    expect = f"{len(full_dstore_sqlite)}x member"
+    assert repr(full_dstore_sqlite).startswith(expect)
+    assert len(full_dstore_sqlite) == len(full_dstore_sqlite.completed)
+
+
+def test_describe(full_dstore_sqlite):
+    got = full_dstore_sqlite.describe
+    assert got.shape >= (3, 2)
+    assert isinstance(got, Table)
+
+
+def test_pickleable_roundtrip(full_dstore_sqlite):
+    """pickling of data stores should be reversible"""
+    re_dstore = loads(dumps(full_dstore_sqlite))
+    assert str(re_dstore) == str(full_dstore_sqlite)
+    assert re_dstore[0].read() == full_dstore_sqlite[0].read()
+
+
+def test_pickleable_member_roundtrip(full_dstore_sqlite):
+    """pickling of data store members should be reversible"""
+    re_member = loads(dumps(full_dstore_sqlite[0]))
+    data = re_member.read()
+    assert len(data) > 0
+
+
+def test_getitem(full_dstore_sqlite):
+    with pytest.raises(IndexError):
+        _ = full_dstore_sqlite[len(full_dstore_sqlite)]
+
+    last = full_dstore_sqlite[-1]
+    first = full_dstore_sqlite[0]
+    assert last.unique_id != first.unique_id
+
+
+def test_empty_data_store(db_dir):
+    dstore = SqliteDataStore(db_dir, mode=OVERWRITE)
+    assert 0 == len(dstore)
+
+
+def test_no_logs(db_dir):
+    dstore = SqliteDataStore(db_dir, mode=OVERWRITE)
+    assert len(dstore.logs) == 0
+
+
+def test_limit_datastore(full_dstore_sqlite):
+    assert len(full_dstore_sqlite) == len(full_dstore_sqlite.completed) + len(
+        full_dstore_sqlite.not_completed
+    )
+    full_dstore_sqlite._limit = len(full_dstore_sqlite.completed)
+    full_dstore_sqlite.drop_not_completed()
+    assert len(full_dstore_sqlite) == full_dstore_sqlite._limit
+    assert len(full_dstore_sqlite) == len(full_dstore_sqlite.completed)
+
+
+def test_validate(full_dstore_sqlite):
+    r = full_dstore_sqlite.validate()
+    assert r.shape == (3, 2)
+
+
+def test_no_not_completed(sql_dstore):
+    assert len(sql_dstore.not_completed) == 0
+
+
+def test_write_read_only_datastore(ro_sql_dstore):
+    with pytest.raises(IOError):
+        ro_sql_dstore.write(unique_id="brca1.fasta", data="test data")
+
+
+def test_write_read_only_member(ro_sql_dstore):
+    with pytest.raises(IOError):
+        ro_sql_dstore[0].write("test data")
+
+
+def test_new_write_read(full_dstore_sqlite):
+    """correctly write content"""
+    identifier = "test1.fasta"
+    data = "test data"
+    m = full_dstore_sqlite.write(unique_id=identifier, data=data)
+    got = full_dstore_sqlite.read(m.unique_id)
+    assert got == data
+
+
+'''
+def test_append to reset log_id
+
+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 = "brca2.fasta"
+    identifier_b = "primates_brca2.fasta"
+    w_dstore.write(unique_id=identifier_a, data=expect_a)
+    w_dstore.write(unique_id=identifier_b, data=expect_b)
+    got_a = w_dstore.read(identifier_a)
+    got_b = w_dstore.read(identifier_b)
+    # check that both bits of data match
+    assert got_a == expect_a
+    assert got_b == expect_b
+'''
+
+
+def test_summary_logs(full_dstore_sqlite):
+    # log summary has a row per log file and a column for each property
+    got = full_dstore_sqlite.summary_logs
+    assert got.shape == (1, 6)
+    assert isinstance(got, Table)
+
+
+def test_summary_not_completed(full_dstore_sqlite):
+    got = full_dstore_sqlite.summary_not_completed
+    assert got.shape >= (1, 1)
+    assert isinstance(got, Table)