X-Git-Url: https://plomlompom.com/repos/foo.html?a=blobdiff_plain;f=plomtask%2Fmisc.py;h=5759c0d8b6bcc5915f84b2e94a533eefa70954e7;hb=e825a876e82ffbedf0234f4dfb6d6055d9e29241;hp=bf071885278471e421a346791bb9db59989097d5;hpb=34741b65438149b4e02f1e2bb4f8fdba5df5a667;p=plomtask
diff --git a/plomtask/misc.py b/plomtask/misc.py
index bf07188..5759c0d 100644
--- a/plomtask/misc.py
+++ b/plomtask/misc.py
@@ -1,6 +1,7 @@
"""Attributes whose values are recorded as a timestamped history."""
from datetime import datetime
from typing import Any
+from sqlite3 import Row
from plomtask.db import DatabaseConnection
@@ -32,6 +33,10 @@ class VersionedAttribute:
or value != self.history[self._newest_timestamp]:
self.history[datetime.now().strftime('%Y-%m-%d %H:%M:%S')] = value
+ def history_from_row(self, row: Row) -> None:
+ """Extend self.history from expected table row format."""
+ self.history[row[1]] = row[2]
+
def at(self, queried_time: str) -> str | float:
"""Retrieve value of timestamp nearest queried_time from the past."""
sorted_timestamps = sorted(self.history.keys())
@@ -46,8 +51,6 @@ class VersionedAttribute:
def save(self, db_conn: DatabaseConnection) -> None:
"""Save as self.history entries, but first wipe old ones."""
- db_conn.exec(f'DELETE FROM {self.table_name} WHERE parent_id = ?',
- (self.parent.id_,))
- for timestamp, value in self.history.items():
- db_conn.exec(f'INSERT INTO {self.table_name} VALUES (?, ?, ?)',
- (self.parent.id_, timestamp, value))
+ db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
+ [[item[0], item[1]]
+ for item in self.history.items()])