X-Git-Url: https://plomlompom.com/repos/foo.html?a=blobdiff_plain;ds=inline;f=plomtask%2Fversioned_attributes.py;fp=plomtask%2Fversioned_attributes.py;h=d3c364942e3d76edf0eb9618df8366866c603709;hb=1a9178b2b7bb66e77f3df01a0c7b2812839637ba;hp=b3442d7df1030b72b2990a21e12ffa12302aa9d1;hpb=5ff672be7e421fbbde8132a66ebb9b734d149a69;p=plomtask
diff --git a/plomtask/versioned_attributes.py b/plomtask/versioned_attributes.py
index b3442d7..d3c3649 100644
--- a/plomtask/versioned_attributes.py
+++ b/plomtask/versioned_attributes.py
@@ -4,6 +4,7 @@ from typing import Any
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'
@@ -30,6 +31,32 @@ class VersionedAttribute:
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.