1 """Attributes whose values are recorded as a timestamped history."""
2 from datetime import datetime
4 from sqlite3 import Row
5 from plomtask.db import DatabaseConnection
8 class VersionedAttribute:
9 """Attributes whose values are recorded as a timestamped history."""
12 parent: Any, table_name: str, default: str | float) -> None:
14 self.table_name = table_name
15 self.default = default
16 self.history: dict[str, str | float] = {}
19 def _newest_timestamp(self) -> str:
20 """Return most recent timestamp."""
21 return sorted(self.history.keys())[-1]
24 def newest(self) -> str | float:
25 """Return most recent value, or self.default if self.history empty."""
26 if 0 == len(self.history):
28 return self.history[self._newest_timestamp]
30 def set(self, value: str | float) -> None:
31 """Add to self.history if and only if not same value as newest one."""
32 if 0 == len(self.history) \
33 or value != self.history[self._newest_timestamp]:
34 self.history[datetime.now().strftime('%Y-%m-%d %H:%M:%S')] = value
36 def history_from_row(self, row: Row) -> None:
37 """Extend self.history from expected table row format."""
38 self.history[row[1]] = row[2]
40 def at(self, queried_time: str) -> str | float:
41 """Retrieve value of timestamp nearest queried_time from the past."""
42 sorted_timestamps = sorted(self.history.keys())
43 if 0 == len(sorted_timestamps):
45 selected_timestamp = sorted_timestamps[0]
46 for timestamp in sorted_timestamps[1:]:
47 if timestamp > queried_time:
49 selected_timestamp = timestamp
50 return self.history[selected_timestamp]
52 def save(self, db_conn: DatabaseConnection) -> None:
53 """Save as self.history entries, but first wipe old ones."""
54 db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
56 for item in self.history.items()])