Use depot instead of dbm for caching.

- Introduce RawData depot object for storing any bytes
- Store screenplay results as RawData by it's md5sum

hworker/depot/database/functions.py

@@ -13,6 +13,7 @@
 ObjectSuccessor = TypeVar("ObjectSuccessor", bound=objects.StoreObject)
 
 _object_to_model_class: dict[type[objects.StoreObject] : type[Base]] = {
+    objects.RawData: RawData,
     objects.Homework: Homework,
     objects.Check: Check,
     objects.Solution: Solution,

hworker/depot/database/models.py

@@ -27,6 +27,17 @@ def __init__(self, ID: str = None, USER_ID: str = None, TASK_ID: str = None, tim
         self.timestamp = timestamp
 
 
+class RawData(Base):
+    __tablename__ = "rawdata"
+
+    content: Mapped[bytes] = mapped_column(LargeBinary)
+
+    # noinspection PyTypeChecker
+    def __init__(self, content: bytes = None, **kwargs):
+        super().__init__(**kwargs)
+        self.content = content
+
+
 class Homework(Base):
     __tablename__ = "homework"
 

hworker/depot/objects.py

@@ -84,6 +84,16 @@ def __matmul__(self, other):
         return list(self.items()) == list(other.items())
 
 
+class RawData(StoreObject):
+    content: bytes
+    _is_versioned: bool = False
+
+    def __init__(self, content: bytes = None, **kwargs):
+        fields = {"USER_ID": ".", "TASK_ID": ".", "timestamp": datetime.datetime.now().timestamp()}
+        super().__init__(**(fields | kwargs))
+        self.content = content
+
+
 class FileObject:
     content: bytes
     timestamp: float

hworker/make/screenplay.py

@@ -10,15 +10,13 @@
 from tempfile import TemporaryDirectory as tmpdir
 from ..log import get_logger
 from subprocess import run
-import dbm.gnu
-from ..config import get_check_directory
 import hashlib
+from ..depot import store, search
+from ..depot.objects import RawData, Criteria
 
 # report.01.second/./BOTH.txt
 REname = re.compile(r"report....(\w+)[./]+(\w+)[.]txt")
 log = get_logger(__name__)
-(check_directory := Path(get_check_directory())).mkdir(parents=True, exist_ok=True)
-cache = dbm.gnu.open(check_directory / "cache.gdbm", "csu")
 
 
 def screendump(command: str, directory: Path) -> bytes:
@@ -34,8 +32,8 @@ def screendump(command: str, directory: Path) -> bytes:
 def screenplay(both: bytes, timer: bytes) -> bytes:
     """Run scritreplay on both / timer data and dump result's screen buffer."""
     md5 = hashlib.md5(both + timer).hexdigest()
-    if answer := cache.get(md5, None):
-        return answer
+    if answer := search(RawData, Criteria("ID", "==", md5), first=True):
+        return answer.content
     with tmpdir() as Dname:
         D = Path(Dname)
         B, T = D / "BOTH.txt", D / "TIME.txt"
@@ -44,7 +42,8 @@ def screenplay(both: bytes, timer: bytes) -> bytes:
         answer = screendump(f"scriptreplay -m 0.001 -t {T} -B {B}", D)
         columns = int(re.sub(rb'.*COLUMNS="(\d+)".*', rb"\1", both[: both.index(b"\n")]))
         rejoin = rf"(^.{{{columns}}})\n".encode()
-        answer = cache[md5] = re.sub(rejoin, rb"\1", answer, flags=re.MULTILINE)
+        answer = re.sub(rejoin, rb"\1", answer, flags=re.MULTILINE)
+        store(RawData(ID=md5, content=answer))
         return answer
 
 

tests/test_screenplay.py

@@ -3,7 +3,10 @@
 Scriptreplay
 '''
 import pytest
+import hashlib
 from hworker.make.screenplay import screenplay, screendump, screenplay_all
+from hworker.depot import search
+from hworker.depot.objects import RawData, Criteria
 from pathlib import Path
 
 
@@ -50,10 +53,18 @@ def test_screendump_esc(self, tmp_path):
     def test_screenplay(self, simplescript):
         """ screenplay echo"""
         assert screenplay(simplescript[0], simplescript[1]).strip() == simplescript[2]
+        assert screenplay(simplescript[0], simplescript[1]).strip() == simplescript[2]
 
     def test_screenplay_esc(self, escapescript):
         """screenplay with color"""
         assert screenplay(escapescript[0], escapescript[1]).strip() == escapescript[2]
+        assert screenplay(escapescript[0], escapescript[1]).strip() == escapescript[2]
+
+    def test_screenplay_cache(self, escapescript):
+        """screnplay and then check for cached play"""
+        assert screenplay(escapescript[0], escapescript[1]).strip() == escapescript[2]
+        md5 = hashlib.md5(escapescript[0] + escapescript[1]).hexdigest()
+        assert search(RawData, Criteria("ID", "==", md5), first=True).content.strip() == escapescript[2]
 
     def test_screenplay_all(self, solution, simplescript, escapescript):
         """untarred solution"""