1 """Attributes whose values are recorded as a timestamped history."""
2 from datetime import datetime
4 from plomtask.db import DatabaseConnection
7 class VersionedAttribute:
8 """Attributes whose values are recorded as a timestamped history."""
11 parent: Any, table_name: str, default: str | float) -> None:
13 self.table_name = table_name
14 self.default = default
15 self.history: dict[str, str | float] = {}
18 def _newest_timestamp(self) -> str:
19 """Return most recent timestamp."""
20 return sorted(self.history.keys())[-1]
23 def newest(self) -> str | float:
24 """Return most recent value, or self.default if self.history empty."""
25 if 0 == len(self.history):
27 return self.history[self._newest_timestamp]
29 def set(self, value: str | float) -> None:
30 """Add to self.history if and only if not same value as newest one."""
31 if 0 == len(self.history) \
32 or value != self.history[self._newest_timestamp]:
33 self.history[datetime.now().strftime('%Y-%m-%d %H:%M:%S')] = value
35 def at(self, queried_time: str) -> str | float:
36 """Retrieve value of timestamp nearest queried_time from the past."""
37 sorted_timestamps = sorted(self.history.keys())
38 if 0 == len(sorted_timestamps):
40 selected_timestamp = sorted_timestamps[0]
41 for timestamp in sorted_timestamps[1:]:
42 if timestamp > queried_time:
44 selected_timestamp = timestamp
45 return self.history[selected_timestamp]
47 def save(self, db_conn: DatabaseConnection) -> None:
48 """Save as self.history entries, but first wipe old ones."""
49 db_conn.exec(f'DELETE FROM {self.table_name} WHERE parent_id = ?',
51 for timestamp, value in self.history.items():
52 db_conn.exec(f'INSERT INTO {self.table_name} VALUES (?, ?, ?)',
53 (self.parent.id_, timestamp, value))