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 sqlite3 import Row
   5 from plomtask.db import DatabaseConnection
   6
   7
   8 class VersionedAttribute:
   9     """Attributes whose values are recorded as a timestamped history."""
  10
  11     def __init__(self,
  12                  parent: Any, table_name: str, default: str | float) -> None:
  13         self.parent = parent
  14         self.table_name = table_name
  15         self.default = default
  16         self.history: dict[str, str | float] = {}
  17
  18     @property
  19     def _newest_timestamp(self) -> str:
  20         """Return most recent timestamp."""
  21         return sorted(self.history.keys())[-1]
  22
  23     @property
  24     def newest(self) -> str | float:
  25         """Return most recent value, or self.default if self.history empty."""
  26         if 0 == len(self.history):
  27             return self.default
  28         return self.history[self._newest_timestamp]
  29
  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
  35
  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]
  39
  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):
  44             return self.default
  45         selected_timestamp = sorted_timestamps[0]
  46         for timestamp in sorted_timestamps[1:]:
  47             if timestamp > queried_time:
  48                 break
  49             selected_timestamp = timestamp
  50         return self.history[selected_timestamp]
  51
  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_,
  55                                   [[item[0], item[1]]
  56                                    for item in self.history.items()])