Stricter versioning: always store migrate version record, file must be within versions included in migrations. Log only for migrations that made changes. Allow deleting older migration functions when no longer required.

kanta/kantaimpl.py

@@ -12,7 +12,7 @@ from typing import Any, Generic, TypeVar
 
 from kanta.callbacks import CallbackRegistry, InjectionContext
 from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
-from kanta.migrate import MigrationRegistry
+from kanta.migrations import Migrations
 from kanta.persistence import PersistenceMixin
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
@@ -28,18 +28,18 @@ class KantaImpl(PersistenceMixin, Generic[T]):
     def __init__(self, **kwargs: Any):
         self.data_type = kwargs.pop("type")
         self.data: T = kwargs.pop("data")
-        self.migrations = kwargs.pop("migrations", None)
         self._kanta = kwargs.pop("kanta", None)
+        migrations = kwargs.pop("migrations", None)
         self.ctx = SimpleNamespace()
         super().__init__(**kwargs)
-        self.migration_registry: MigrationRegistry | None = None
+        self.migrations: Migrations | None = None
-        if self.migrations is not None:
+        if migrations is not None:
             module = (
-                importlib.import_module(self.migrations)
+                importlib.import_module(migrations)
-                if isinstance(self.migrations, str)
+                if isinstance(migrations, str)
-                else self.migrations
+                else migrations
             )
-            self.migration_registry = MigrationRegistry.from_module(module)
+            self.migrations = Migrations.from_module(module)
 
         self.in_transaction = False
         self.transaction_snapshot: dict[str, Any] | None = None
@@ -55,9 +55,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         )
 
         self.statedict = struct_to_dict(self.data, serializer=self.serializer)
-        self.version = (
+        self.version = self.migrations.dbver if self.migrations is not None else 0
-            self.migration_registry.dbver if self.migration_registry is not None else 0
-        )
 
     def add_bootstrap(
         self,
@@ -141,10 +139,11 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     cause_type=type(e).__name__,
                 ) from e
 
-            if self.migration_registry is not None:
+            migrations_ran = False
-                rr.version = self.migration_registry.apply(
+            if self.migrations is not None:
-                    rr.state, rr.version, self._kanta
+                previous_version = rr.version
-                )
+                rr.version = self.migrations.apply(rr.state, rr.version, self._kanta)
+                migrations_ran = rr.version != previous_version
 
             self.statedict = copy.deepcopy(rr.state)
             self.data = restore_data_in_place(
@@ -159,6 +158,13 @@ class KantaImpl(PersistenceMixin, Generic[T]):
             if self.readonly:
                 self.statedict = copy.deepcopy(normalized)
             else:
+                if migrations_ran:
+                    self.queue_change(
+                        f"migrate:v{self.version}",
+                        self.statedict,
+                        mtime=False,
+                        force=True,
+                    )
                 self.queue_change("migrate:msgspec", normalized, mtime=False)
             self.snapshot.ts = (
                 datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)

kanta/migrations.py

@@ -6,37 +6,44 @@ or by prefix. Each runs exactly once based on the current version.
 
 from __future__ import annotations
 
+import copy
 import importlib
 import inspect
 import logging
 from types import ModuleType
 from typing import Any
 
+from kanta.exceptions import DatabaseError
+
 _logger = logging.getLogger(__name__)
 
+# Cache registries by imported module object so that many Kanta instances using
+# the same migrations module do not re-scan it each time.
+_module_registry_cache: dict[ModuleType, Migrations] = {}
 
-class MigrationRegistry:
+
+class Migrations:
     """Registry of schema migration functions.
 
     Usage::
 
-        registry = MigrationRegistry()
+        migrations = Migrations()
 
-        @registry.register
+        @migrations.register
         def migrate_v1(d: dict, kanta) -> None:
             d.setdefault("version", 1)
             kanta.ctx.note = "migrated"
 
-        @registry.register
+        @migrations.register
         def migrate_v2(d: dict) -> None:
             d.setdefault("version", 2)
 
-        new_version = registry.apply(state, current_version=0, kanta=kanta)
+        new_version = migrations.apply(state, current_version=0, kanta=kanta)
 
     Or load from a module::
 
-        registry = MigrationRegistry.from_module("myapp.migrations")
+        migrations = Migrations.from_module("myapp.migrations")
-        new_version = registry.apply(state, current_version=0, kanta=kanta)
+        new_version = migrations.apply(state, current_version=0, kanta=kanta)
     """
 
     def __init__(self) -> None:
@@ -59,24 +66,30 @@ class MigrationRegistry:
         return fn
 
     @classmethod
-    def from_module(cls, module: str | ModuleType) -> MigrationRegistry:
+    def from_module(cls, module: str | ModuleType) -> Migrations:
-        """Create a registry by scanning a module for ``migrate_vN`` functions.
+        """Create or retrieve a cached registry by scanning a module.
 
         Args:
             module: A module name (string) or an imported module object.
         """
-        reg = cls()
         if isinstance(module, str):
             mod = importlib.import_module(module)
         else:
             mod = module
 
+        try:
+            return _module_registry_cache[mod]
+        except KeyError:
+            pass
+
+        reg = cls()
         for name in dir(mod):
             if name.startswith("migrate_v"):
                 fn = getattr(mod, name)
                 if callable(fn):
                     version = reg._migration_version(fn)
                     reg._migrations[version] = fn
+        _module_registry_cache[mod] = reg
         return reg
 
     @property
@@ -84,6 +97,11 @@ class MigrationRegistry:
         """Current schema version (= highest discovered migration, or 0)."""
         return max(self._migrations.keys(), default=0)
 
+    @property
+    def minver(self) -> int:
+        """Minimum supported current version (first migration minus 1, or 0)."""
+        return min(self._migrations.keys(), default=1) - 1
+
     @staticmethod
     def _call_migration(fn: Any, data_dict: dict[str, Any], kanta: Any) -> None:
         """Call *fn* with the data dict and, if accepted, the Kanta instance."""
@@ -104,19 +122,33 @@ class MigrationRegistry:
     ) -> int:
         """Apply pending migrations to *data_dict* in place.
 
+        Missing intermediate migration steps are silently skipped.
+
+        Raises:
+            DatabaseError: If the database version is newer than the highest
+                supported version or older than the minimum supported version.
+
         Returns the new version after all migrations.
         """
-        while current_version < self.dbver:
+        if current_version > self.dbver:
-            next_version = current_version + 1
+            raise DatabaseError(
-            fn = self._migrations.get(next_version)
+                f"Database version v{current_version} is newer than the "
-            if fn is None:
+                f"highest supported version v{self.dbver}"
-                raise ValueError(
+            )
-                    f"Missing migration step migrate_v{next_version} "
+        if current_version < self.minver:
-                    f"(highest discovered is v{self.dbver})"
+            raise DatabaseError(
-                )
+                f"Database version v{current_version} is older than the "
+                f"minimum supported version v{self.minver}"
+            )
+
+        for version in sorted(self._migrations.keys()):
+            if version <= current_version:
+                continue
+            fn = self._migrations[version]
+            before = copy.deepcopy(data_dict) if not silent else None
             self._call_migration(fn, data_dict, kanta)
-            current_version = next_version
+            current_version = version
-            if not silent:
+            if not silent and before != data_dict:
                 desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
                 _logger.info("Applied migration %s: %s", fn.__name__, desc)
         return current_version

kanta/persistence.py

@@ -140,8 +140,10 @@ class PersistenceMixin:
             raise TypeError("mtime must be True, False, or a datetime")
 
         diff = compute_diff(self.statedict, current)
-        if not diff and not force:
+        if not diff:
-            return None
+            if not force:
+                return None
+            diff = {}
 
         record = ChangeRecord(
             ts=now,

tests/test_kanta_integration.py

@@ -17,6 +17,7 @@ from .support import (
     change_actions,
     fixed_change,
     make_kanta,
+    make_migrations_module,
     read_changes,
     seed_single_change,
 )
@@ -473,6 +474,43 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
     assert "migrate:msgspec" in change_actions(path, format_config)
 
 
+@pytest.mark.asyncio
+async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path, fixed_change("init", {"counter": 0, "users": {}}), format_config
+    )
+
+    def migrate_v1(d, kanta):
+        """No-op migration that only bumps the schema version."""
+        pass
+
+    mod = make_migrations_module("empty_migration_mod", "migrate_v1", migrate_v1)
+
+    try:
+        kanta = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta.open()
+        assert kanta.version == 1
+        await kanta.flush()
+        await kanta.close()
+
+        records = read_changes(path, format_config)
+        migration_records = [r for r in records if r.a.startswith("migrate")]
+        assert len(migration_records) == 1
+        assert migration_records[0].v == 1
+        assert migration_records[0].diff == {}
+
+        kanta2 = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta2.open()
+        assert kanta2.version == 1
+        await kanta2.close()
+
+        records2 = read_changes(path, format_config)
+        assert len([r for r in records2 if r.a.startswith("migrate")]) == 1
+    finally:
+        sys.modules.pop("empty_migration_mod", None)
+
+
 @pytest.mark.asyncio
 async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
     path = tmp_path / "test.db"

tests/test_migrations.py

@@ -1,6 +1,10 @@
+import logging
 from types import ModuleType, SimpleNamespace
 
-from kanta.migrate import MigrationRegistry
+import pytest
+
+from kanta.exceptions import DatabaseError
+from kanta.migrations import Migrations
 
 
 class _DummyKanta:
@@ -9,7 +13,7 @@ class _DummyKanta:
 
 
 def test_register_and_apply():
-    reg = MigrationRegistry()
+    reg = Migrations()
     kanta = _DummyKanta()
 
     @reg.register
@@ -27,7 +31,7 @@ def test_register_and_apply():
 
 
 def test_no_migrations_needed():
-    reg = MigrationRegistry()
+    reg = Migrations()
     kanta = _DummyKanta()
 
     @reg.register
@@ -52,7 +56,7 @@ def test_from_module():
     mod.__dict__["migrate_v1"] = migrate_v1
     mod.__dict__["migrate_v2"] = migrate_v2
 
-    reg = MigrationRegistry.from_module(mod)
+    reg = Migrations.from_module(mod)
     assert reg.dbver == 2
 
     state = {}
@@ -62,7 +66,7 @@ def test_from_module():
 
 
 def test_migrations_can_use_kanta_ctx():
-    reg = MigrationRegistry()
+    reg = Migrations()
     kanta = _DummyKanta()
 
     @reg.register
@@ -78,7 +82,7 @@ def test_migrations_can_use_kanta_ctx():
 
 
 def test_migration_can_omit_kanta_argument():
-    reg = MigrationRegistry()
+    reg = Migrations()
     kanta = _DummyKanta()
 
     @reg.register
@@ -89,3 +93,112 @@ def test_migration_can_omit_kanta_argument():
     new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
     assert new_ver == 1
     assert state["x"] == 1
+
+
+def test_version_too_new():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v2 is newer than the highest supported version v1",
+    ):
+        reg.apply({}, current_version=2, kanta=kanta, silent=True)
+
+
+def test_version_too_old():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v1 is older than the minimum supported version v2",
+    ):
+        reg.apply({}, current_version=1, kanta=kanta, silent=True)
+
+
+def test_missing_middle_migration_is_skipped():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v3(d):
+        d["y"] = 3
+
+    state = {"x": 1}
+    new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 1
+    assert state["y"] == 3
+
+
+def test_old_migrations_deleted_current_supported():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    state = {"x": 2}
+    new_ver = reg.apply(state, current_version=2, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 3
+
+
+def test_migration_log_only_when_changed(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """Set x."""
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v2(d):
+        """No-op."""
+        pass
+
+    @reg.register
+    def migrate_v3(d):
+        """Set y."""
+        d["y"] = 3
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(messages) == 2
+    assert "migrate_v1" in messages[0]
+    assert "Set x" in messages[0]
+    assert "migrate_v3" in messages[1]
+    assert "Set y" in messages[1]
+
+
+def test_no_op_migration_produces_no_log(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """No-op."""
+        pass
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages