From 123e910873fa20efa37158c530ec6e317b614366 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 6 Aug 2026 23:55:09 +0000 Subject: [PATCH] Add @kanta.clock hook replacing the UTC clock for all record and snapshot timestamps --- kanta/kanta.py | 19 ++++++++ kanta/kantaimpl.py | 6 ++- kanta/persistence.py | 34 ++++++++++++- kanta/snapshot.py | 9 +++- tests/test_clock.py | 114 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 177 insertions(+), 5 deletions(-) create mode 100644 tests/test_clock.py diff --git a/kanta/kanta.py b/kanta/kanta.py index 93eb19b..ea76bdf 100644 --- a/kanta/kanta.py +++ b/kanta/kanta.py @@ -253,6 +253,25 @@ class Kanta(Generic[T]): return _register return _register(fn) + def clock(self, fn=None): + """Register a clock callback replacing the default UTC clock. + + Can be used as ``@kanta.clock``. The callback takes no arguments and + must return a :class:`~datetime.datetime`; its value is used for all + record timestamps (``ts``, and ``m`` when ``mtime`` is ``True``) and + snapshot timestamps. Register before :meth:`open` so that bootstrap + and migration records use the custom clock as well. This is mainly + useful for tests and reproducible demos. + """ + + def _register(callback): + self._impl.add_clock(callback) + return callback + + if fn is None: + return _register + return _register(fn) + def logmigr(self, fn=None): """Register a migration logging callback. diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index 86b70bd..aedc4bb 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -296,7 +296,11 @@ class KantaImpl(PersistenceMixin, Generic[T]): self.snapshot.request_force() await self.flush() self.snapshot.maybe_write( - self.file, self.version, self.statedict, m=self.mtime + self.file, + self.version, + self.statedict, + m=self.mtime, + now=self.now(), ) if migrations_ran and migration_result is not None: diff --git a/kanta/persistence.py b/kanta/persistence.py index 5113256..76a5fe8 100644 --- a/kanta/persistence.py +++ b/kanta/persistence.py @@ -4,8 +4,10 @@ from __future__ import annotations import asyncio import copy +import inspect import logging from collections import deque +from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -41,6 +43,7 @@ class PersistenceMixin: opened: bool readonly: bool mtime: datetime | None + clock: Callable[[], datetime] | None def __init__(self, **kwargs: Any) -> None: """Initialize persistence-owned state used by mixin methods.""" @@ -62,6 +65,31 @@ class PersistenceMixin: self.flush_interval = flush_interval self.version = 0 self.mtime: datetime | None = None + self.clock: Callable[[], datetime] | None = None + + def add_clock(self, callback) -> None: + """Register a clock callback ``() -> datetime`` replacing the UTC clock.""" + if not callable(callback): + raise TypeError("clock callback must be callable") + for param in inspect.signature(callback).parameters.values(): + if param.default is inspect.Parameter.empty and param.kind in ( + param.POSITIONAL_ONLY, + param.POSITIONAL_OR_KEYWORD, + param.KEYWORD_ONLY, + ): + raise TypeError("clock callback must not require arguments") + self.clock = callback + + def now(self) -> datetime: + """Current time from the registered clock (default: UTC now).""" + if self.clock is None: + return datetime.now(UTC) + ts = self.clock() + if not isinstance(ts, datetime): + raise TypeError( + f"clock callback must return a datetime, got {type(ts).__name__}" + ) + return ts def add_fatal_error(self, callback) -> None: """Register one fatal error callback in call order.""" @@ -100,7 +128,9 @@ class PersistenceMixin: def maybe_snapshot(self) -> None: """Evaluate and possibly write a snapshot from current state.""" - self.snapshot.maybe_write(self.file, self.version, self.statedict, m=self.mtime) + self.snapshot.maybe_write( + self.file, self.version, self.statedict, m=self.mtime, now=self.now() + ) def queue_change( self, @@ -128,7 +158,7 @@ class PersistenceMixin: The queued :class:`ChangeRecord`, or ``None`` if the diff was empty and *force* is ``False``. """ - now = datetime.now(UTC) + now = self.now() if mtime is True: m = now diff --git a/kanta/snapshot.py b/kanta/snapshot.py index 8896f50..c6389fe 100644 --- a/kanta/snapshot.py +++ b/kanta/snapshot.py @@ -38,11 +38,16 @@ class SnapshotState: self.changes += count def maybe_write( - self, file, version: int, state: dict, m: datetime | None = None + self, + file, + version: int, + state: dict, + m: datetime | None = None, + now: datetime | None = None, ) -> None: """Write snapshot when thresholds/time policy allows it.""" force = self._force_pending - now = datetime.now(UTC) + now = now if now is not None else datetime.now(UTC) if not force: if self.changes < self._min_diffs: return diff --git a/tests/test_clock.py b/tests/test_clock.py new file mode 100644 index 0000000..10cb4dc --- /dev/null +++ b/tests/test_clock.py @@ -0,0 +1,114 @@ +from datetime import UTC, datetime, timedelta + +import pytest + +from .support import ( + Data, + make_kanta, + make_migrations_module, + read_changes, + read_last_snapshot, +) + +T0 = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) + + +def test_clock_rejects_non_callable(tmp_path, format_config): + kanta = make_kanta(tmp_path / "test.db", Data, format_config) + + with pytest.raises(TypeError, match="must be callable"): + kanta.clock(42) + + +def test_clock_rejects_required_argument(tmp_path, format_config): + kanta = make_kanta(tmp_path / "test.db", Data, format_config) + + with pytest.raises(TypeError, match="must not require arguments"): + + @kanta.clock + def fake_now(tz) -> datetime: + return T0 + + +@pytest.mark.asyncio +async def test_clock_rejects_non_datetime_result(tmp_path, format_config): + kanta = make_kanta(tmp_path / "test.db", Data, format_config) + + @kanta.clock + def fake_now() -> datetime: + return "noon" + + with pytest.raises(TypeError, match="must return a datetime"): + await kanta.open(log=False) + + +@pytest.mark.asyncio +async def test_clock_controls_record_timestamps(tmp_path, format_config): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data, format_config) + current = T0 + + @kanta.clock + def fake_now() -> datetime: + return current + + await kanta.open(log=False) + current = T0 + timedelta(hours=1) + with kanta.transaction(action="update") as data: + data.counter = 1 + current = T0 + timedelta(hours=2) + with kanta.transaction(action="repair", mtime=False) as data: + data.counter = 2 + await kanta.close() + + bootstrap, update, repair = read_changes(path, format_config) + assert bootstrap.ts == T0 + assert bootstrap.m == T0 + assert update.ts == T0 + timedelta(hours=1) + assert update.m == T0 + timedelta(hours=1) + # System operation: stamped by the clock, but m is not updated. + assert repair.ts == T0 + timedelta(hours=2) + assert repair.m is None + assert kanta.mtime == T0 + timedelta(hours=1) + + +@pytest.mark.asyncio +async def test_clock_controls_migration_and_snapshot_timestamps( + tmp_path, format_config +): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data, format_config) + + @kanta.clock + def fake_now() -> datetime: + return T0 + + await kanta.open(log=False) + await kanta.close() + + def migrate_v1(d): + """Bump counter""" + d["counter"] = 1 + + migrations = make_migrations_module("clock_migrations", "migrate_v1", migrate_v1) + t1 = T0 + timedelta(days=1) + kanta2 = make_kanta(path, Data, format_config, migrations=migrations) + + @kanta2.clock + def fake_now2() -> datetime: + return t1 + + await kanta2.open(log=False) + await kanta2.close() + + migrate_records = [ + r for r in read_changes(path, format_config) if r.a.startswith("migrate:") + ] + assert migrate_records + assert all(r.ts == t1 for r in migrate_records) + + snapshot = read_last_snapshot(path, format_config) + assert snapshot is not None + assert snapshot.ts == t1 + # mtime is carried forward from the last real modification. + assert snapshot.m == T0