X-Git-Url: https://plomlompom.com/repos/foo.html?a=blobdiff_plain;f=plomtask%2Fmisc.py;h=efa8898efd97d51ad685307902f135dc920c3ee7;hb=89624d5e05480c832a079008bbb9992f411be0dd;hp=1b780e29d90af5c17820f642fc445e27b9e8cc25;hpb=b557c789f4eec704db0e6276390395fac5d8db9e;p=plomtask

diff --git a/plomtask/misc.py b/plomtask/misc.py
index 1b780e2..efa8898 100644
--- a/plomtask/misc.py
+++ b/plomtask/misc.py
@@ -1,8 +1,51 @@
-"""
-Whatever fits nowhere else, and/or is too small/trivial
-to merit its own module at this point.
-"""
+"""Attributes whose values are recorded as a timestamped history."""
+from datetime import datetime
+from typing import Any
+from plomtask.db import DatabaseConnection
 
 
-class HandledException(Exception):
-    """To identify Exceptions based on expected (if faulty) user behavior."""
+class VersionedAttribute:
+    """Attributes whose values are recorded as a timestamped history."""
+
+    def __init__(self,
+                 parent: Any, table_name: str, default: str | float) -> None:
+        self.parent = parent
+        self.table_name = table_name
+        self.default = default
+        self.history: dict[str, str | float] = {}
+
+    @property
+    def _newest_timestamp(self) -> str:
+        """Return most recent timestamp."""
+        return sorted(self.history.keys())[-1]
+
+    @property
+    def newest(self) -> str | float:
+        """Return most recent value, or self.default if self.history empty."""
+        if 0 == len(self.history):
+            return self.default
+        return self.history[self._newest_timestamp]
+
+    def set(self, value: str | float) -> None:
+        """Add to self.history if and only if not same value as newest one."""
+        if 0 == len(self.history) \
+                or value != self.history[self._newest_timestamp]:
+            self.history[datetime.now().strftime('%Y-%m-%d %H:%M:%S')] = value
+
+    def at(self, queried_time: str) -> str | float:
+        """Retrieve value of timestamp nearest queried_time from the past."""
+        sorted_timestamps = sorted(self.history.keys())
+        if 0 == len(sorted_timestamps):
+            return self.default
+        selected_timestamp = sorted_timestamps[0]
+        for timestamp in sorted_timestamps[1:]:
+            if timestamp > queried_time:
+                break
+            selected_timestamp = timestamp
+        return self.history[selected_timestamp]
+
+    def save(self, db_conn: DatabaseConnection) -> None:
+        """Save as self.history entries, but first wipe old ones."""
+        db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
+                                  [[item[0], item[1]]
+                                   for item in self.history.items()])