plomlompom.com Git - plomtask/blob - plomtask/misc.py

   1 """Attributes whose values are recorded as a timestamped history."""
   2 from datetime import datetime
   3 from typing import Any
   4 from plomtask.db import DatabaseConnection
   5
   6
   7 class VersionedAttribute:
   8     """Attributes whose values are recorded as a timestamped history."""
   9
  10     def __init__(self,
  11                  parent: Any, table_name: str, default: str | float) -> None:
  12         self.parent = parent
  13         self.table_name = table_name
  14         self.default = default
  15         self.history: dict[str, str | float] = {}
  16
  17     @property
  18     def _newest_timestamp(self) -> str:
  19         """Return most recent timestamp."""
  20         return sorted(self.history.keys())[-1]
  21
  22     @property
  23     def newest(self) -> str | float:
  24         """Return most recent value, or self.default if self.history empty."""
  25         if 0 == len(self.history):
  26             return self.default
  27         return self.history[self._newest_timestamp]
  28
  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
  34
  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):
  39             return self.default
  40         selected_timestamp = sorted_timestamps[0]
  41         for timestamp in sorted_timestamps[1:]:
  42             if timestamp > queried_time:
  43                 break
  44             selected_timestamp = timestamp
  45         return self.history[selected_timestamp]
  46
  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 = ?',
  50                      (self.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))