home · contact · privacy
Improve template layouts.
[plomtask] / plomtask / versioned_attributes.py
1 """Attributes whose values are recorded as a timestamped history."""
2 from datetime import datetime
3 from typing import Any
4 from sqlite3 import Row
5 from time import sleep
6 from plomtask.db import DatabaseConnection
7
8 TIMESTAMP_FMT = '%Y-%m-%d %H:%M:%S.%f'
9
10
11 class VersionedAttribute:
12     """Attributes whose values are recorded as a timestamped history."""
13
14     def __init__(self,
15                  parent: Any, table_name: str, default: str | float) -> None:
16         self.parent = parent
17         self.table_name = table_name
18         self.default = default
19         self.history: dict[str, str | float] = {}
20
21     @property
22     def _newest_timestamp(self) -> str:
23         """Return most recent timestamp."""
24         return sorted(self.history.keys())[-1]
25
26     @property
27     def newest(self) -> str | float:
28         """Return most recent value, or self.default if self.history empty."""
29         if 0 == len(self.history):
30             return self.default
31         return self.history[self._newest_timestamp]
32
33     def set(self, value: str | float) -> None:
34         """Add to self.history if and only if not same value as newest one.
35
36         Note that we wait one micro-second, as timestamp comparison to check
37         most recent elements only goes up to that precision.
38
39         Also note that we don't check against .newest because that may make us
40         compare value against .default even if not set. We want to be able to
41         explicitly set .default as the first element.
42         """
43         sleep(0.00001)
44         if 0 == len(self.history) \
45                 or value != self.history[self._newest_timestamp]:
46             self.history[datetime.now().strftime(TIMESTAMP_FMT)] = value
47
48     def history_from_row(self, row: Row) -> None:
49         """Extend self.history from expected table row format."""
50         self.history[row[1]] = row[2]
51
52     def at(self, queried_time: str) -> str | float:
53         """Retrieve value of timestamp nearest queried_time from the past."""
54         sorted_timestamps = sorted(self.history.keys())
55         if 0 == len(sorted_timestamps):
56             return self.default
57         selected_timestamp = sorted_timestamps[0]
58         for timestamp in sorted_timestamps[1:]:
59             if timestamp > queried_time:
60                 break
61             selected_timestamp = timestamp
62         return self.history[selected_timestamp]
63
64     def save(self, db_conn: DatabaseConnection) -> None:
65         """Save as self.history entries, but first wipe old ones."""
66         db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
67                                   [[item[0], item[1]]
68                                    for item in self.history.items()])
69
70     def remove(self, db_conn: DatabaseConnection) -> None:
71         """Remove from DB."""
72         db_conn.delete_where(self.table_name, 'parent', self.parent.id_)