from sqlite3 import Row
from time import sleep
from plomtask.db import DatabaseConnection
+from plomtask.exceptions import HandledException, BadFormatException
TIMESTAMP_FMT = '%Y-%m-%d %H:%M:%S.%f'
self.default = default
self.history: dict[str, str | float] = {}
+ def __hash__(self) -> int:
+ history_tuples = tuple((k, v) for k, v in self.history.items())
+ hashable = (self.parent.id_, self.table_name, self.default,
+ history_tuples)
+ return hash(hashable)
+
+ @property
+ def as_dict(self) -> dict[str, object]:
+ """Return self as (json.dumps-coompatible) dict."""
+ d = {'parent_process_id': self.parent.id_,
+ 'table_name': self.table_name,
+ 'history': self.history}
+ return d
+
@property
def _newest_timestamp(self) -> str:
"""Return most recent timestamp."""
return self.default
return self.history[self._newest_timestamp]
+ def reset_timestamp(self, old_str: str, new_str: str) -> None:
+ """Rename self.history key (timestamp) old to new.
+
+ Chronological sequence of keys must be preserved, i.e. cannot move
+ key before earlier or after later timestamp.
+ """
+ try:
+ new = datetime.strptime(new_str, TIMESTAMP_FMT)
+ old = datetime.strptime(old_str, TIMESTAMP_FMT)
+ except ValueError as exc:
+ raise BadFormatException('Timestamp of illegal format.') from exc
+ timestamps = list(self.history.keys())
+ if old_str not in timestamps:
+ raise HandledException(f'Timestamp {old} not found in history.')
+ sorted_timestamps = sorted([datetime.strptime(t, TIMESTAMP_FMT)
+ for t in timestamps])
+ expected_position = sorted_timestamps.index(old)
+ sorted_timestamps.remove(old)
+ sorted_timestamps += [new]
+ sorted_timestamps.sort()
+ if sorted_timestamps.index(new) != expected_position:
+ raise HandledException('Timestamp not respecting chronology.')
+ value = self.history[old_str]
+ del self.history[old_str]
+ self.history[new_str] = value
+
def set(self, value: str | float) -> None:
"""Add to self.history if and only if not same value as newest one.
def at(self, queried_time: str) -> str | float:
"""Retrieve value of timestamp nearest queried_time from the past."""
+ if len(queried_time) == 10:
+ queried_time += ' 23:59:59.999'
sorted_timestamps = sorted(self.history.keys())
if 0 == len(sorted_timestamps):
return self.default
db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_,
[[item[0], item[1]]
for item in self.history.items()])
+
+ def remove(self, db_conn: DatabaseConnection) -> None:
+ """Remove from DB."""
+ db_conn.delete_where(self.table_name, 'parent', self.parent.id_)