23890155859

Committed 02 Apr 2026 07:52AM UTC coverage: 95.157% (-0.3%) from 95.44%

Build # 23890155859

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #7 from GavinHuttley/main

Expand type hinting

Coverage Stats

125 of 132 new or added lines in 9 files covered. (94.7%)

2063 of 2168 relevant lines covered (95.16%)

5.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.67

/src/scinexus/sqlite_data_store.py

from __future__ import annotations

import contextlib
import datetime
import os
import re
import sqlite3
import weakref
from pathlib import Path
from typing import TYPE_CHECKING, Any

from scitrack import get_text_hexdigest  # type: ignore[import-untyped]

from scinexus.data_store import (
    _LOG_TABLE,
    APPEND,
    OVERWRITE,
    READONLY,
    DataMember,
    DataMemberABC,
    DataStoreABC,
    DataStoreDirectory,
    Mode,
)
from scinexus.misc import extend_docstring_from

if TYPE_CHECKING:  # pragma: no cover
    from citeable import CitationBase

_RESULT_TABLE = "results"
_MEMORY = ":memory:"
_mem_pattern = re.compile(r"^\s*[:]{0,1}memory[:]{0,1}\s*$")
NoneType = type(None)

# dealing with python3.12 deprecation of datetime objects and their sqlite3 handling


def _datetime_to_iso(timestamp: datetime.datetime) -> str:
    """timestamp in ISO 8601 format"""
    return timestamp.isoformat()


sqlite3.register_adapter(datetime.datetime, _datetime_to_iso)


def _datetime_from_iso(data: bytes) -> datetime.datetime:
    """timestamp from ISO 8601 format"""
    return datetime.datetime.fromisoformat(data.decode())


sqlite3.register_converter("timestamp", _datetime_from_iso)


# create db
def open_sqlite_db_rw(path: str | Path) -> sqlite3.Connection:
    """creates a new sqlitedb for read/write at path, can be an in-memory db

    Notes
    -----
    This function embeds the schema. There are three tables:

    - results: analysis objects, may be completed or not completed
    - 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, is_completed INTEGER, data BLOB)",
        "citations(citation_id INTEGER PRIMARY KEY, data TEXT)",
    ]
    for table in creates:
        db.execute(create_template.format(table))
    return db


def open_sqlite_db_ro(path: str | Path) -> sqlite3.Connection:
    """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
    if not has_valid_schema(db):
        msg = "database does not have a valid schema"
        raise ValueError(msg)
    return db


def has_valid_schema(db: sqlite3.Connection) -> bool:
    # TODO: should be a full schema check
    query = "SELECT name FROM sqlite_master WHERE type='table'"
    result = db.execute(query).fetchall()
    table_names = {r["name"] for r in result}
    _required = {_RESULT_TABLE, _LOG_TABLE, "state"}
    _optional = {"citations"}
    return _required <= table_names <= (_required | _optional)


class DataStoreSqlite(DataStoreABC):
    store_suffix = "sqlitedb"

    def __init__(
        self,
        source: str | Path,
        mode: Mode | str = READONLY,
        limit: int | None = None,
        verbose: bool = False,
    ) -> None:
        if _mem_pattern.search(str(source)):
            self._source: str | Path = _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}")
            )
        self._mode = Mode(mode)
        if mode is not READONLY and limit is not None:
            msg = "Using limit argument is only valid for readonly datastores"
            raise ValueError(
                msg,
            )
        self._limit = limit
        self._verbose = verbose
        self._db: sqlite3.Connection | None = None
        self._open = False
        self._log_id: int | None = None
        weakref.finalize(self, self.close)

    def __getstate__(self) -> dict[str, object]:
        return {**self._init_vals}

    def __setstate__(self, state: dict[str, Any]) -> None:
        # this will reset connections to read only db's
        obj = self.__class__(**state)
        self.__dict__.update(obj.__dict__)

    def __del__(self) -> None:
        """close the db connection when the object is deleted"""
        self.close()

    @property
    def source(self) -> str | Path:
        """string that references connecting to data store, override in subclass constructor"""
        return self._source

    @property
    def mode(self) -> Mode:
        """string that references datastore mode, override in override in subclass constructor"""
        return self._mode

    @property
    def limit(self) -> int | None:
        return self._limit

    @property
    def db(self) -> sqlite3.Connection:
        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
            self.lock()

        if self._db is None:
            msg = "database connection is unexpectedly None"
            raise ValueError(msg)
        return self._db

    def _init_log(self) -> None:
        timestamp = datetime.datetime.now(tz=datetime.UTC)
        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) -> None:
        if getattr(self, "_db", None) is None:
            return
        if self._db is None:
            msg = "database connection is unexpectedly None"
            raise RuntimeError(msg)
        with contextlib.suppress(sqlite3.ProgrammingError):
            self._db.close()
        self._open = False

    def read(self, unique_id: str) -> str | bytes:
        """
        identifier string formed from Path(table_name) / identifier
        """
        uid_path = Path(unique_id)
        table_name = str(uid_path.parent)
        if table_name not in (
            ".",
            _LOG_TABLE,
        ):
            msg = f"unknown table for {str(uid_path)!r}"
            raise ValueError(msg)

        if table_name != _LOG_TABLE:
            cmnd = f"SELECT * FROM {_RESULT_TABLE} WHERE record_id = ?"
            result = self.db.execute(cmnd, (uid_path.name,)).fetchone()
            return result["data"]

        cmnd = f"SELECT * FROM {_LOG_TABLE} WHERE log_name = ?"
        result = self.db.execute(cmnd, (uid_path.name,)).fetchone()

        return result["data"]

    @property
    def completed(self) -> list[DataMemberABC]:
        if not self._completed:
            self._completed = self._select_members(
                table_name=_RESULT_TABLE,
                is_completed=True,
            )
        return self._completed

    @property
    def not_completed(self) -> list[DataMemberABC]:
        """returns database records of type NotCompleted"""
        if not self._not_completed:
            self._not_completed = self._select_members(
                table_name=_RESULT_TABLE,
                is_completed=False,
            )
        return self._not_completed

    def _select_members(
        self,
        *,
        table_name: str,
        is_completed: bool,
    ) -> list[DataMemberABC]:
        limit = f"LIMIT {self.limit}" if self.limit else ""
        cmnd = self.db.execute(
            f"SELECT record_id FROM {table_name} WHERE is_completed=? {limit}",
            (is_completed,),
        )
        return [
            DataMember(data_store=self, unique_id=r["record_id"])
            for r in cmnd.fetchall()
        ]

    @property
    def logs(self) -> list[DataMemberABC]:
        """returns all log records"""
        cmnd = self.db.execute(f"SELECT log_name FROM {_LOG_TABLE}")
        return [
            DataMember(data_store=self, unique_id=Path(_LOG_TABLE) / r["log_name"])
            for r in cmnd.fetchall()
            if r["log_name"]
        ]

    def _write(
        self,
        *,
        table_name: str,
        unique_id: str,
        data: str | bytes,
        is_completed: bool,
    ) -> DataMemberABC | 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.
        is_completed: bool
            flag to identify NotCompleted 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:
            # TODO how to evaluate whether writing a new log?
            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,is_completed) VALUES (?,?,?,?,?)"
            self.db.execute(cmnd, (unique_id, data, self._log_id, md5, is_completed))

        return DataMember(data_store=self, unique_id=unique_id)

    def drop_not_completed(self, *, unique_id: str | None = None) -> None:
        vals: tuple[int] | tuple[int, str]
        if not unique_id:
            cmnd = f"DELETE FROM {_RESULT_TABLE} WHERE is_completed=?"
            vals = (0,)
        else:
            cmnd = f"DELETE FROM {_RESULT_TABLE} WHERE is_completed=? AND record_id=?"
            vals = (0, unique_id)
        self.db.execute(cmnd, vals)
        self._not_completed = []

    @property
    def _lock_id(self) -> int | None:
        """returns lock_pid"""
        result = self.db.execute("SELECT lock_pid FROM state").fetchone()
        return result[0] if result else result

    @property
    def locked(self) -> bool:
        """returns if lock_pid is NULL or doesn't exist."""
        return self._lock_id is not None

    def lock(self) -> None:
        """if writable, and not locked, locks the database to this pid"""
        if self.mode is READONLY:
            return
        if self._db is None:
            msg = "database connection is unexpectedly None"
            raise RuntimeError(msg)
        result = self._db.execute("SELECT state_id,lock_pid FROM state").fetchall()
        locked = result[0]["lock_pid"] if result else None
        if locked and self.mode is OVERWRITE:
            msg = (
                f"You are trying to OVERWRITE {str(self.source)!r} which is "
                "locked. Use APPEND mode or unlock."
            )
            raise OSError(
                msg,
            )

        if result:
            # we will update an existing
            state_id = result[0]["state_id"]
            cmnd = "UPDATE state SET lock_pid=? WHERE state_id=?"
            vals = [os.getpid(), state_id]
        else:
            cmnd = "INSERT INTO state(lock_pid) VALUES (?)"
            vals = [os.getpid()]
        self._db.execute(cmnd, tuple(vals))

    def unlock(self, force: bool = False) -> None:
        """remove a lock if pid matches. If force, ignores pid. ignored if mode is READONLY"""
        if self.mode is READONLY:
            return

        lock_id = self._lock_id
        if lock_id is None:
            return

        if lock_id == os.getpid() or force:
            self.db.execute("UPDATE state SET lock_pid=NULL WHERE state_id=1")

        return

    @extend_docstring_from(DataStoreDirectory.write)
    def write(self, *, unique_id: str, data: str | bytes) -> DataMemberABC:  # type: ignore[override]
        if unique_id.startswith(_RESULT_TABLE):
            unique_id = Path(unique_id).name

        super().write(unique_id=unique_id, data=data)

        self.drop_not_completed(unique_id=unique_id)

        member = self._write(
            table_name=_RESULT_TABLE,
            unique_id=unique_id,
            data=data,
            is_completed=True,
        )
        if member is None:
            msg = "write to results table failed to produce a member"
            raise RuntimeError(msg)
        if member not in self._completed:
            self._completed.append(member)
        return member

    @extend_docstring_from(DataStoreDirectory.write_log)
    def write_log(self, *, unique_id: str, data: str | bytes) -> None:
        if unique_id.startswith(_LOG_TABLE):
            unique_id = Path(unique_id).name

        super().write_log(unique_id=unique_id, data=data)
        _ = self._write(
            table_name=_LOG_TABLE,
            unique_id=unique_id,
            data=data,
            is_completed=False,
        )

    @extend_docstring_from(DataStoreDirectory.write_not_completed)
    def write_not_completed(  # type: ignore[override]
        self, *, unique_id: str, data: str | bytes
    ) -> DataMemberABC:
        if unique_id.startswith(_RESULT_TABLE):
            unique_id = Path(unique_id).name

        super().write_not_completed(unique_id=unique_id, data=data)
        member = self._write(
            table_name=_RESULT_TABLE,
            unique_id=unique_id,
            data=data,
            is_completed=False,
        )
        if member is None:
            msg = "write to results table failed to produce a member"
            raise RuntimeError(msg)
        self._not_completed.append(member)
        return member

    def md5(self, unique_id: str) -> str | None:
        """
        Parameters
        ----------
        unique_id
            name of data store member
        Returns
        -------
        md5 checksum for the member, if available, None otherwise
        """
        cmnd = f"SELECT * FROM {_RESULT_TABLE} WHERE record_id = ?"
        result = self.db.execute(cmnd, (unique_id,)).fetchone()

        return result["md5"] if result else None

    def write_citations(self, *, data: tuple[CitationBase, ...]) -> None:
        if not data:
            return
        if not self._has_citations_table():
            self.db.execute(
                "CREATE TABLE IF NOT EXISTS citations"
                "(citation_id INTEGER PRIMARY KEY, data TEXT)",
            )
        from citeable import to_jsons

        json_data = to_jsons(data)
        existing = self.db.execute("SELECT citation_id FROM citations").fetchone()
        if existing:
            self.db.execute(
                "UPDATE citations SET data=? WHERE citation_id=?",
                (json_data, existing["citation_id"]),
            )
        else:
            self.db.execute("INSERT INTO citations(data) VALUES (?)", (json_data,))

    def _load_citations(self) -> list[CitationBase]:
        from citeable import from_jsons

        if not self._has_citations_table():
            return []
        result = self.db.execute("SELECT data FROM citations").fetchone()
        if not result:
            return []
        return from_jsons(result["data"])

    def _has_citations_table(self) -> bool:
        result = self.db.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='citations'",
        ).fetchone()
        return result is not None

    def _describe(self) -> dict[str, object]:
        if self.locked and self._lock_id != os.getpid():
            title = f"Locked db store. Locked to pid={self._lock_id}, current pid={os.getpid()}."
        elif self.locked:
            title = "Locked to the current process."
        else:
            title = "Unlocked db store."
        result = super()._describe()
        result["title"] = title
        return result

    @property
    def record_type(self) -> str:
        """class name of completed results"""
        result = self.db.execute("SELECT record_type FROM state").fetchone()
        return result["record_type"]

    @record_type.setter
    def record_type(self, obj: object) -> None:
        from scinexus.misc import get_object_provenance

        rt = self.record_type
        if self.mode is OVERWRITE and rt:
            msg = f"cannot overwrite existing record_type {rt}"
            raise OSError(msg)

        n = get_object_provenance(obj)
        self.db.execute("UPDATE state SET record_type=? WHERE state_id=1", (n,))

    def _summary_not_completed(self) -> list[dict]:
        """returns a list of dicts summarising not completed results"""
        from scinexus.data_store import summary_not_completeds
        from scinexus.io import DEFAULT_DESERIALISER

        return summary_not_completeds(
            self.not_completed,
            deserialise=DEFAULT_DESERIALISER,
        )

1	from __future__ import annotations	6✔
2
3	import contextlib	6✔
4	import datetime	6✔
5	import os	6✔
6	import re	6✔
7	import sqlite3	6✔
8	import weakref	6✔
9	from pathlib import Path	6✔
10	from typing import TYPE_CHECKING, Any	6✔
11
12	from scitrack import get_text_hexdigest # type: ignore[import-untyped]	6✔
13
14	from scinexus.data_store import (	6✔
15	_LOG_TABLE,
16	APPEND,
17	OVERWRITE,
18	READONLY,
19	DataMember,
20	DataMemberABC,
21	DataStoreABC,
22	DataStoreDirectory,
23	Mode,
24	)
25	from scinexus.misc import extend_docstring_from	6✔
26
27	if TYPE_CHECKING: # pragma: no cover
28	from citeable import CitationBase
29
30	_RESULT_TABLE = "results"	6✔
31	_MEMORY = ":memory:"	6✔
32	_mem_pattern = re.compile(r"^\s[:]{0,1}memory[:]{0,1}\s$")	6✔
33	NoneType = type(None)	6✔
34
35	# dealing with python3.12 deprecation of datetime objects and their sqlite3 handling
36
37
38	def _datetime_to_iso(timestamp: datetime.datetime) -> str:	6✔
39	"""timestamp in ISO 8601 format"""
40	return timestamp.isoformat()	6✔
41
42
43	sqlite3.register_adapter(datetime.datetime, _datetime_to_iso)	6✔
44
45
46	def _datetime_from_iso(data: bytes) -> datetime.datetime:	6✔
47	"""timestamp from ISO 8601 format"""
48	return datetime.datetime.fromisoformat(data.decode())	6✔
49
50
51	sqlite3.register_converter("timestamp", _datetime_from_iso)	6✔
52
53
54	# create db
55	def open_sqlite_db_rw(path: str \| Path) -> sqlite3.Connection:	6✔
56	"""creates a new sqlitedb for read/write at path, can be an in-memory db
57
58	Notes
59	-----
60	This function embeds the schema. There are three tables:
61
62	- results: analysis objects, may be completed or not completed
63	- logs: log-file contents
64	- state: whether db is locked to a process
65
66	Returns
67	-------
68	Handle to a sqlite3 session
69	"""
70	db = sqlite3.connect(	6✔
71	path,
72	isolation_level=None,
73	detect_types=sqlite3.PARSE_DECLTYPES \| sqlite3.PARSE_COLNAMES,
74	)
75	db.row_factory = sqlite3.Row	6✔
76	create_template = "CREATE TABLE IF NOT EXISTS {};"	6✔
77	# note it is essential to use INTEGER for the autoincrement of primary key to work
78	creates = [	6✔
79	"state(state_id INTEGER PRIMARY KEY, record_type TEXT, lock_pid INTEGER)",
80	f"{_LOG_TABLE}(log_id INTEGER PRIMARY KEY, log_name TEXT, date timestamp, data BLOB)",
81	f"{_RESULT_TABLE}(record_id TEXT PRIMARY KEY, log_id INTEGER, md5 BLOB, is_completed INTEGER, data BLOB)",
82	"citations(citation_id INTEGER PRIMARY KEY, data TEXT)",
83	]
84	for table in creates:	6✔
85	db.execute(create_template.format(table))	6✔
86	return db	6✔
87
88
89	def open_sqlite_db_ro(path: str \| Path) -> sqlite3.Connection:	6✔
90	"""returns db opened as read only
91	Returns
92	-------
93	Handle to a sqlite3 session
94	"""
95	db = sqlite3.connect(	6✔
96	f"file:{path}?mode=ro",
97	isolation_level=None,
98	detect_types=sqlite3.PARSE_DECLTYPES \| sqlite3.PARSE_COLNAMES,
99	uri=True,
100	)
101	db.row_factory = sqlite3.Row	6✔
102	if not has_valid_schema(db):	6✔
103	msg = "database does not have a valid schema"	6✔
104	raise ValueError(msg)	6✔
105	return db	6✔
106
107
108	def has_valid_schema(db: sqlite3.Connection) -> bool:	6✔
109	# TODO: should be a full schema check
110	query = "SELECT name FROM sqlite_master WHERE type='table'"	6✔
111	result = db.execute(query).fetchall()	6✔
112	table_names = {r["name"] for r in result}	6✔
113	_required = {_RESULT_TABLE, _LOG_TABLE, "state"}	6✔
114	_optional = {"citations"}	6✔
115	return _required <= table_names <= (_required \| _optional)	6✔
116
117
118	class DataStoreSqlite(DataStoreABC):	6✔
119	store_suffix = "sqlitedb"	6✔
120
121	def __init__(	6✔
122	self,
123	source: str \| Path,
124	mode: Mode \| str = READONLY,
125	limit: int \| None = None,
126	verbose: bool = False,
127	) -> None:
128	if _mem_pattern.search(str(source)):	6✔
129	self._source: str \| Path = _MEMORY	6✔
130	else:
131	source = Path(source).expanduser()	6✔
132	self._source = (	6✔
133	source
134	if source.suffix[1:] == self.store_suffix # sliced to remove "."
135	else Path(f"{source}.{self.store_suffix}")
136	)
137	self._mode = Mode(mode)	6✔
138	if mode is not READONLY and limit is not None:	6✔
139	msg = "Using limit argument is only valid for readonly datastores"	6✔
140	raise ValueError(	6✔
141	msg,
142	)
143	self._limit = limit	6✔
144	self._verbose = verbose	6✔
145	self._db: sqlite3.Connection \| None = None	6✔
146	self._open = False	6✔
147	self._log_id: int \| None = None	6✔
148	weakref.finalize(self, self.close)	6✔
149
150	def __getstate__(self) -> dict[str, object]:	6✔
151	return {**self._init_vals}	6✔
152
153	def __setstate__(self, state: dict[str, Any]) -> None:	6✔
154	# this will reset connections to read only db's
155	obj = self.__class__(**state)	6✔
156	self.__dict__.update(obj.__dict__)	6✔
157
158	def __del__(self) -> None:	6✔
159	"""close the db connection when the object is deleted"""
160	self.close()	6✔
161
162	@property	6✔
163	def source(self) -> str \| Path:	6✔
164	"""string that references connecting to data store, override in subclass constructor"""
165	return self._source	6✔
166
167	@property	6✔
168	def mode(self) -> Mode:	6✔
169	"""string that references datastore mode, override in override in subclass constructor"""
170	return self._mode	6✔
171
172	@property	6✔
173	def limit(self) -> int \| None:	6✔
174	return self._limit	6✔
175
176	@property	6✔
177	def db(self) -> sqlite3.Connection:	6✔
178	if self._db is None:	6✔
179	db_func = open_sqlite_db_ro if self.mode is READONLY else open_sqlite_db_rw	6✔
180	self._db = db_func(self.source)	6✔
181	self._open = True	6✔
182	self.lock()	6✔
183
184	if self._db is None:	6✔
NEW 185	msg = "database connection is unexpectedly None"	×
NEW 186	raise ValueError(msg)	×
187	return self._db	6✔
188
189	def _init_log(self) -> None:	6✔
190	timestamp = datetime.datetime.now(tz=datetime.UTC)	6✔
191	self.db.execute(f"INSERT INTO {_LOG_TABLE}(date) VALUES (?)", (timestamp,))	6✔
192	self._log_id = self.db.execute(	6✔
193	f"SELECT log_id FROM {_LOG_TABLE} where date = ?",
194	(timestamp,),
195	).fetchone()["log_id"]
196
197	def close(self) -> None:	6✔
198	if getattr(self, "_db", None) is None:	6✔
199	return	6✔
200	if self._db is None:	6✔
201	msg = "database connection is unexpectedly None"	×
202	raise RuntimeError(msg)	×
203	with contextlib.suppress(sqlite3.ProgrammingError):	6✔
204	self._db.close()	6✔
205	self._open = False	6✔
206
207	def read(self, unique_id: str) -> str \| bytes:	6✔
208	"""
209	identifier string formed from Path(table_name) / identifier
210	"""
211	uid_path = Path(unique_id)	6✔
212	table_name = str(uid_path.parent)	6✔
213	if table_name not in (	6✔
214	".",
215	_LOG_TABLE,
216	):
217	msg = f"unknown table for {str(uid_path)!r}"	6✔
218	raise ValueError(msg)	6✔
219
220	if table_name != _LOG_TABLE:	6✔
221	cmnd = f"SELECT * FROM {_RESULT_TABLE} WHERE record_id = ?"	6✔
222	result = self.db.execute(cmnd, (uid_path.name,)).fetchone()	6✔
223	return result["data"]	6✔
224
225	cmnd = f"SELECT * FROM {_LOG_TABLE} WHERE log_name = ?"	6✔
226	result = self.db.execute(cmnd, (uid_path.name,)).fetchone()	6✔
227
228	return result["data"]	6✔
229
230	@property	6✔
231	def completed(self) -> list[DataMemberABC]:	6✔
232	if not self._completed:	6✔
233	self._completed = self._select_members(	6✔
234	table_name=_RESULT_TABLE,
235	is_completed=True,
236	)
237	return self._completed	6✔
238
239	@property	6✔
240	def not_completed(self) -> list[DataMemberABC]:	6✔
241	"""returns database records of type NotCompleted"""
242	if not self._not_completed:	6✔
243	self._not_completed = self._select_members(	6✔
244	table_name=_RESULT_TABLE,
245	is_completed=False,
246	)
247	return self._not_completed	6✔
248
249	def _select_members(	6✔
250	self,
251	*,
252	table_name: str,
253	is_completed: bool,
254	) -> list[DataMemberABC]:
255	limit = f"LIMIT {self.limit}" if self.limit else ""	6✔
256	cmnd = self.db.execute(	6✔
257	f"SELECT record_id FROM {table_name} WHERE is_completed=? {limit}",
258	(is_completed,),
259	)
260	return [	6✔
261	DataMember(data_store=self, unique_id=r["record_id"])
262	for r in cmnd.fetchall()
263	]
264
265	@property	6✔
266	def logs(self) -> list[DataMemberABC]:	6✔
267	"""returns all log records"""
268	cmnd = self.db.execute(f"SELECT log_name FROM {_LOG_TABLE}")	6✔
269	return [	6✔
270	DataMember(data_store=self, unique_id=Path(_LOG_TABLE) / r["log_name"])
271	for r in cmnd.fetchall()
272	if r["log_name"]
273	]
274
275	def _write(	6✔
276	self,
277	*,
278	table_name: str,
279	unique_id: str,
280	data: str \| bytes,
281	is_completed: bool,
282	) -> DataMemberABC \| None:
283	"""
284	Parameters
285	----------
286	table_name: str
287	name of table to save data. It must be _RESULT_TABLE or _LOG_TABLE.
288	unique_id : str
289	unique identifier that data will be saved under.
290	data: str
291	data to be saved.
292	is_completed: bool
293	flag to identify NotCompleted results
294
295	Returns
296	-------
297	DataMember instance or None when writing to _LOG_TABLE
298	"""
299	if self._log_id is None:	6✔
300	self._init_log()	6✔
301
302	if table_name == _LOG_TABLE:	6✔
303	# TODO how to evaluate whether writing a new log?
304	cmnd = f"UPDATE {table_name} SET data =?, log_name =? WHERE log_id=?"	6✔
305	self.db.execute(cmnd, (data, unique_id, self._log_id))	6✔
306	return None	6✔
307
308	md5 = get_text_hexdigest(data)	6✔
309
310	if unique_id in self and self.mode is not APPEND:	6✔
311	cmnd = f"UPDATE {table_name} SET data= ?, log_id=?, md5=? WHERE record_id=?"	6✔
312	self.db.execute(cmnd, (data, self._log_id, md5, unique_id))	6✔
313	else:
314	cmnd = f"INSERT INTO {table_name} (record_id,data,log_id,md5,is_completed) VALUES (?,?,?,?,?)"	6✔
315	self.db.execute(cmnd, (unique_id, data, self._log_id, md5, is_completed))	6✔
316
317	return DataMember(data_store=self, unique_id=unique_id)	6✔
318
319	def drop_not_completed(self, *, unique_id: str \| None = None) -> None:	6✔
320	vals: tuple[int] \| tuple[int, str]
321	if not unique_id:	6✔
322	cmnd = f"DELETE FROM {_RESULT_TABLE} WHERE is_completed=?"	6✔
323	vals = (0,)	6✔
324	else:
325	cmnd = f"DELETE FROM {_RESULT_TABLE} WHERE is_completed=? AND record_id=?"	6✔
326	vals = (0, unique_id)	6✔
327	self.db.execute(cmnd, vals)	6✔
328	self._not_completed = []	6✔
329
330	@property	6✔
331	def _lock_id(self) -> int \| None:	6✔
332	"""returns lock_pid"""
333	result = self.db.execute("SELECT lock_pid FROM state").fetchone()	6✔
334	return result[0] if result else result	6✔
335
336	@property	6✔
337	def locked(self) -> bool:	6✔
338	"""returns if lock_pid is NULL or doesn't exist."""
339	return self._lock_id is not None	6✔
340
341	def lock(self) -> None:	6✔
342	"""if writable, and not locked, locks the database to this pid"""
343	if self.mode is READONLY:	6✔
344	return	6✔
345	if self._db is None:	6✔
346	msg = "database connection is unexpectedly None"	6✔
347	raise RuntimeError(msg)	6✔
348	result = self._db.execute("SELECT state_id,lock_pid FROM state").fetchall()	6✔
349	locked = result[0]["lock_pid"] if result else None	6✔
350	if locked and self.mode is OVERWRITE:	6✔
351	msg = (	6✔
352	f"You are trying to OVERWRITE {str(self.source)!r} which is "
353	"locked. Use APPEND mode or unlock."
354	)
355	raise OSError(	6✔
356	msg,
357	)
358
359	if result:	6✔
360	# we will update an existing
361	state_id = result[0]["state_id"]	6✔
362	cmnd = "UPDATE state SET lock_pid=? WHERE state_id=?"	6✔
363	vals = [os.getpid(), state_id]	6✔
364	else:
365	cmnd = "INSERT INTO state(lock_pid) VALUES (?)"	6✔
366	vals = [os.getpid()]	6✔
367	self._db.execute(cmnd, tuple(vals))	6✔
368
369	def unlock(self, force: bool = False) -> None:	6✔
370	"""remove a lock if pid matches. If force, ignores pid. ignored if mode is READONLY"""
371	if self.mode is READONLY:	6✔
372	return	6✔
373
374	lock_id = self._lock_id	6✔
375	if lock_id is None:	6✔
376	return	6✔
377
378	if lock_id == os.getpid() or force:	6✔
379	self.db.execute("UPDATE state SET lock_pid=NULL WHERE state_id=1")	6✔
380
381	return	6✔
382
383	@extend_docstring_from(DataStoreDirectory.write)	6✔
384	def write(self, *, unique_id: str, data: str \| bytes) -> DataMemberABC: # type: ignore[override]	6✔
385	if unique_id.startswith(_RESULT_TABLE):	6✔
386	unique_id = Path(unique_id).name	6✔
387
388	super().write(unique_id=unique_id, data=data)	6✔
389
390	self.drop_not_completed(unique_id=unique_id)	6✔
391
392	member = self._write(	6✔
393	table_name=_RESULT_TABLE,
394	unique_id=unique_id,
395	data=data,
396	is_completed=True,
397	)
398	if member is None:	6✔
399	msg = "write to results table failed to produce a member"	×
400	raise RuntimeError(msg)	×
401	if member not in self._completed:	6✔
402	self._completed.append(member)	6✔
403	return member	6✔
404
405	@extend_docstring_from(DataStoreDirectory.write_log)	6✔
406	def write_log(self, *, unique_id: str, data: str \| bytes) -> None:	6✔
407	if unique_id.startswith(_LOG_TABLE):	6✔
408	unique_id = Path(unique_id).name	6✔
409
410	super().write_log(unique_id=unique_id, data=data)	6✔
411	_ = self._write(	6✔
412	table_name=_LOG_TABLE,
413	unique_id=unique_id,
414	data=data,
415	is_completed=False,
416	)
417
418	@extend_docstring_from(DataStoreDirectory.write_not_completed)	6✔
419	def write_not_completed( # type: ignore[override]	6✔
420	self, *, unique_id: str, data: str \| bytes
421	) -> DataMemberABC:
422	if unique_id.startswith(_RESULT_TABLE):	6✔
423	unique_id = Path(unique_id).name	6✔
424
425	super().write_not_completed(unique_id=unique_id, data=data)	6✔
426	member = self._write(	6✔
427	table_name=_RESULT_TABLE,
428	unique_id=unique_id,
429	data=data,
430	is_completed=False,
431	)
432	if member is None:	6✔
433	msg = "write to results table failed to produce a member"	×
434	raise RuntimeError(msg)	×
435	self._not_completed.append(member)	6✔
436	return member	6✔
437
438	def md5(self, unique_id: str) -> str \| None:	6✔
439	"""
440	Parameters
441	----------
442	unique_id
443	name of data store member
444	Returns
445	-------
446	md5 checksum for the member, if available, None otherwise
447	"""
448	cmnd = f"SELECT * FROM {_RESULT_TABLE} WHERE record_id = ?"	6✔
449	result = self.db.execute(cmnd, (unique_id,)).fetchone()	6✔
450
451	return result["md5"] if result else None	6✔
452
453	def write_citations(self, *, data: tuple[CitationBase, ...]) -> None:	6✔
454	if not data:	6✔
455	return	6✔
456	if not self._has_citations_table():	6✔
457	self.db.execute(	×
458	"CREATE TABLE IF NOT EXISTS citations"
459	"(citation_id INTEGER PRIMARY KEY, data TEXT)",
460	)
461	from citeable import to_jsons	6✔
462
463	json_data = to_jsons(data)	6✔
464	existing = self.db.execute("SELECT citation_id FROM citations").fetchone()	6✔
465	if existing:	6✔
466	self.db.execute(	6✔
467	"UPDATE citations SET data=? WHERE citation_id=?",
468	(json_data, existing["citation_id"]),
469	)
470	else:
471	self.db.execute("INSERT INTO citations(data) VALUES (?)", (json_data,))	6✔
472
473	def _load_citations(self) -> list[CitationBase]:	6✔
474	from citeable import from_jsons	6✔
475
476	if not self._has_citations_table():	6✔
477	return []	6✔
478	result = self.db.execute("SELECT data FROM citations").fetchone()	6✔
479	if not result:	6✔
480	return []	6✔
481	return from_jsons(result["data"])	6✔
482
483	def _has_citations_table(self) -> bool:	6✔
484	result = self.db.execute(	6✔
485	"SELECT name FROM sqlite_master WHERE type='table' AND name='citations'",
486	).fetchone()
487	return result is not None	6✔
488
489	def _describe(self) -> dict[str, object]:	6✔
490	if self.locked and self._lock_id != os.getpid():	6✔
491	title = f"Locked db store. Locked to pid={self._lock_id}, current pid={os.getpid()}."	6✔
492	elif self.locked:	6✔
493	title = "Locked to the current process."	6✔
494	else:
495	title = "Unlocked db store."	6✔
496	result = super()._describe()	6✔
497	result["title"] = title	6✔
498	return result	6✔
499
500	@property	6✔
501	def record_type(self) -> str:	6✔
502	"""class name of completed results"""
503	result = self.db.execute("SELECT record_type FROM state").fetchone()	6✔
504	return result["record_type"]	6✔
505
506	@record_type.setter	6✔
507	def record_type(self, obj: object) -> None:	6✔
508	from scinexus.misc import get_object_provenance	6✔
509
510	rt = self.record_type	6✔
511	if self.mode is OVERWRITE and rt:	6✔
512	msg = f"cannot overwrite existing record_type {rt}"	6✔
513	raise OSError(msg)	6✔
514
515	n = get_object_provenance(obj)	6✔
516	self.db.execute("UPDATE state SET record_type=? WHERE state_id=1", (n,))	6✔
517
518	def _summary_not_completed(self) -> list[dict]:	6✔
519	"""returns a list of dicts summarising not completed results"""
520	from scinexus.data_store import summary_not_completeds	6✔
521	from scinexus.io import DEFAULT_DESERIALISER	6✔
522
523	return summary_not_completeds(	6✔
524	self.not_completed,
525	deserialise=DEFAULT_DESERIALISER,
526	)

cogent3 / scinexus / 23890155859

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous