From: Christian Heller Date: Wed, 1 May 2024 23:11:36 +0000 (+0200) Subject: Move VersionedAttributes code into appropriotely named module. X-Git-Url: https://plomlompom.com/repos/%7B%7Bdb.prefix%7D%7D/%7B%7B%20web_path%20%7D%7D/static/%7B%7Bprefix%7D%7D/ledger?a=commitdiff_plain;h=a0244482692edf794960f3c2af9461be2ad15e8a;p=plomtask Move VersionedAttributes code into appropriotely named module. --- diff --git a/plomtask/conditions.py b/plomtask/conditions.py index 45aa62b..539db9c 100644 --- a/plomtask/conditions.py +++ b/plomtask/conditions.py @@ -3,7 +3,7 @@ from __future__ import annotations from typing import Any from sqlite3 import Row from plomtask.db import DatabaseConnection, BaseModel -from plomtask.misc import VersionedAttribute +from plomtask.versioned_attributes import VersionedAttribute from plomtask.exceptions import HandledException diff --git a/plomtask/misc.py b/plomtask/misc.py deleted file mode 100644 index ab39df0..0000000 --- a/plomtask/misc.py +++ /dev/null @@ -1,68 +0,0 @@ -"""Attributes whose values are recorded as a timestamped history.""" -from datetime import datetime -from typing import Any -from sqlite3 import Row -from time import sleep -from plomtask.db import DatabaseConnection - -TIMESTAMP_FMT = '%Y-%m-%d %H:%M:%S.%f' - - -class VersionedAttribute: - """Attributes whose values are recorded as a timestamped history.""" - - def __init__(self, - parent: Any, table_name: str, default: str | float) -> None: - self.parent = parent - self.table_name = table_name - self.default = default - self.history: dict[str, str | float] = {} - - @property - def _newest_timestamp(self) -> str: - """Return most recent timestamp.""" - return sorted(self.history.keys())[-1] - - @property - def newest(self) -> str | float: - """Return most recent value, or self.default if self.history empty.""" - if 0 == len(self.history): - return self.default - return self.history[self._newest_timestamp] - - def set(self, value: str | float) -> None: - """Add to self.history if and only if not same value as newest one. - - Note that we wait one micro-second, as timestamp comparison to check - most recent elements only goes up to that precision. - - Also note that we don't check against .newest because that may make us - compare value against .default even if not set. We want to be able to - explicitly set .default as the first element. - """ - sleep(0.00001) - if 0 == len(self.history) \ - or value != self.history[self._newest_timestamp]: - self.history[datetime.now().strftime(TIMESTAMP_FMT)] = value - - def history_from_row(self, row: Row) -> None: - """Extend self.history from expected table row format.""" - self.history[row[1]] = row[2] - - def at(self, queried_time: str) -> str | float: - """Retrieve value of timestamp nearest queried_time from the past.""" - sorted_timestamps = sorted(self.history.keys()) - if 0 == len(sorted_timestamps): - return self.default - selected_timestamp = sorted_timestamps[0] - for timestamp in sorted_timestamps[1:]: - if timestamp > queried_time: - break - selected_timestamp = timestamp - return self.history[selected_timestamp] - - def save(self, db_conn: DatabaseConnection) -> None: - """Save as self.history entries, but first wipe old ones.""" - db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_, - [[item[0], item[1]] - for item in self.history.items()]) diff --git a/plomtask/processes.py b/plomtask/processes.py index 375a0be..21e2d81 100644 --- a/plomtask/processes.py +++ b/plomtask/processes.py @@ -4,7 +4,7 @@ from dataclasses import dataclass from typing import Set, Any from sqlite3 import Row from plomtask.db import DatabaseConnection, BaseModel -from plomtask.misc import VersionedAttribute +from plomtask.versioned_attributes import VersionedAttribute from plomtask.conditions import Condition, ConditionsRelations from plomtask.exceptions import (NotFoundException, BadFormatException, HandledException) diff --git a/plomtask/versioned_attributes.py b/plomtask/versioned_attributes.py new file mode 100644 index 0000000..ab39df0 --- /dev/null +++ b/plomtask/versioned_attributes.py @@ -0,0 +1,68 @@ +"""Attributes whose values are recorded as a timestamped history.""" +from datetime import datetime +from typing import Any +from sqlite3 import Row +from time import sleep +from plomtask.db import DatabaseConnection + +TIMESTAMP_FMT = '%Y-%m-%d %H:%M:%S.%f' + + +class VersionedAttribute: + """Attributes whose values are recorded as a timestamped history.""" + + def __init__(self, + parent: Any, table_name: str, default: str | float) -> None: + self.parent = parent + self.table_name = table_name + self.default = default + self.history: dict[str, str | float] = {} + + @property + def _newest_timestamp(self) -> str: + """Return most recent timestamp.""" + return sorted(self.history.keys())[-1] + + @property + def newest(self) -> str | float: + """Return most recent value, or self.default if self.history empty.""" + if 0 == len(self.history): + return self.default + return self.history[self._newest_timestamp] + + def set(self, value: str | float) -> None: + """Add to self.history if and only if not same value as newest one. + + Note that we wait one micro-second, as timestamp comparison to check + most recent elements only goes up to that precision. + + Also note that we don't check against .newest because that may make us + compare value against .default even if not set. We want to be able to + explicitly set .default as the first element. + """ + sleep(0.00001) + if 0 == len(self.history) \ + or value != self.history[self._newest_timestamp]: + self.history[datetime.now().strftime(TIMESTAMP_FMT)] = value + + def history_from_row(self, row: Row) -> None: + """Extend self.history from expected table row format.""" + self.history[row[1]] = row[2] + + def at(self, queried_time: str) -> str | float: + """Retrieve value of timestamp nearest queried_time from the past.""" + sorted_timestamps = sorted(self.history.keys()) + if 0 == len(sorted_timestamps): + return self.default + selected_timestamp = sorted_timestamps[0] + for timestamp in sorted_timestamps[1:]: + if timestamp > queried_time: + break + selected_timestamp = timestamp + return self.history[selected_timestamp] + + def save(self, db_conn: DatabaseConnection) -> None: + """Save as self.history entries, but first wipe old ones.""" + db_conn.rewrite_relations(self.table_name, 'parent', self.parent.id_, + [[item[0], item[1]] + for item in self.history.items()]) diff --git a/tests/versioned_attributes.py b/tests/versioned_attributes.py index fdb1939..5238d70 100644 --- a/tests/versioned_attributes.py +++ b/tests/versioned_attributes.py @@ -3,7 +3,7 @@ from unittest import TestCase from time import sleep from datetime import datetime from tests.utils import TestCaseWithDB -from plomtask.misc import VersionedAttribute, TIMESTAMP_FMT +from plomtask.versioned_attributes import VersionedAttribute, TIMESTAMP_FMT from plomtask.db import BaseModel SQL_TEST_TABLE = '''